add mg ser&deser

metagpt/utils/serialize.py

@@ -4,13 +4,13 @@
 
 import copy
 import pickle
-from typing import Dict, List
 
 from metagpt.actions.action_output import ActionOutput
 from metagpt.schema import Message
+from metagpt.actions.action import Action
 
 
-def actionoutout_schema_to_mapping(schema: Dict) -> Dict:
+def actionoutout_schema_to_mapping(schema: dict) -> dict:
     """
     directly traverse the `properties` in the first level.
     schema structure likes
@@ -35,13 +35,47 @@ def actionoutout_schema_to_mapping(schema: Dict) -> Dict:
         if property["type"] == "string":
             mapping[field] = (str, ...)
         elif property["type"] == "array" and property["items"]["type"] == "string":
-            mapping[field] = (List[str], ...)
+            mapping[field] = (list[str], ...)
         elif property["type"] == "array" and property["items"]["type"] == "array":
-            # here only consider the `List[List[str]]` situation
-            mapping[field] = (List[List[str]], ...)
+            # here only consider the `list[list[str]]` situation
+            mapping[field] = (list[list[str]], ...)
     return mapping
 
 
+def actionoutput_mapping_to_str(mapping: dict) -> dict:
+    new_mapping = {}
+    for key, value in mapping.items():
+        new_mapping[key] = str(value)
+    return new_mapping
+
+
+def actionoutput_str_to_mapping(mapping: dict) -> dict:
+    new_mapping = {}
+    for key, value in mapping.items():
+        if value == "(<class 'str'>, Ellipsis)":
+            new_mapping[key] = (str, ...)
+        else:
+            new_mapping[key] = eval(value)  # `"'(list[str], Ellipsis)"` to `(list[str], ...)`
+    return new_mapping
+
+
+def serialize_general_message(message: Message) -> dict:
+    """ serialize Message, not to save"""
+    message_cp = copy.deepcopy(message)
+    ic = message_cp.instruct_content
+    if ic:
+        # model create by pydantic create_model like `pydantic.main.prd`, can't pickle.dump directly
+        schema = ic.schema()
+        mapping = actionoutout_schema_to_mapping(schema)
+        mapping = actionoutput_mapping_to_str(mapping)
+
+        message_cp.instruct_content = {"class": schema["title"], "mapping": mapping, "value": ic.dict()}
+    cb = message_cp.cause_by
+    if cb:
+        message_cp.cause_by = cb.ser_class()
+    return message_cp.dict()
+
+
 def serialize_message(message: Message):
     message_cp = copy.deepcopy(message)  # avoid `instruct_content` value update by reference
     ic = message_cp.instruct_content
@@ -56,6 +90,24 @@ def serialize_message(message: Message):
     return msg_ser
 
 
+def deserialize_general_message(message_dict: dict) -> Message:
+    """ deserialize Message, not to load"""
+    instruct_content = message_dict.pop("instruct_content")
+    cause_by = message_dict.pop("cause_by")
+
+    message = Message(**message_dict)
+    if instruct_content:
+        ic = instruct_content
+        mapping = actionoutput_str_to_mapping(ic["mapping"])
+        ic_obj = ActionOutput.create_model_class(class_name=ic["class"], mapping=mapping)
+        ic_new = ic_obj(**ic["value"])
+        message.instruct_content = ic_new
+    if cause_by:
+        message.cause_by = Action.deser_class(cause_by)
+
+    return message
+
+
 def deserialize_message(message_ser: str) -> Message:
     message = pickle.loads(message_ser)
     if message.instruct_content:

metagpt/utils/utils.py

@@ -3,7 +3,10 @@
 # @Desc   :
 
 import typing
-
+from typing import Any
+import json
+from pathlib import Path
+import importlib
 from tenacity import _utils
 
 
@@ -20,3 +23,36 @@ def general_after_log(logger: "loguru.Logger", sec_format: str = "%0.3f") -> typ
         )
 
     return log_it
+
+
+def read_json_file(json_file: str, encoding=None) -> list[Any]:
+    if not Path(json_file).exists():
+        raise FileNotFoundError(f"json_file: {json_file} not exist, return []")
+
+    with open(json_file, "r", encoding=encoding) as fin:
+        try:
+            data = json.load(fin)
+        except Exception as exp:
+            raise ValueError(f"read json file: {json_file} failed")
+    return data
+
+
+def write_json_file(json_file: str, data: list, encoding=None):
+    folder_path = Path(json_file).parent
+    if not folder_path.exists():
+        folder_path.mkdir(parents=True, exist_ok=True)
+
+    with open(json_file, "w", encoding=encoding) as fout:
+        json.dump(data, fout, ensure_ascii=False, indent=4)
+
+
+def import_class(class_name: str, module_name: str) -> type:
+    module = importlib.import_module(module_name)
+    a_class = getattr(module, class_name)
+    return a_class
+
+
+def import_class_inst(class_name: str, module_name: str, *args, **kwargs) -> object:
+    a_class = import_class(class_name, module_name)
+    class_inst = a_class(*args, **kwargs)
+    return class_inst