MetaGPT

1
#!/usr/bin/env python
2
# -*- coding: utf-8 -*-
3
# @Desc   : the implement of serialization and deserialization
4

5
import copy
6
import pickle
7

8
from metagpt.utils.common import import_class
9

10

11
def actionoutout_schema_to_mapping(schema: dict) -> dict:
12
    """
13
    directly traverse the `properties` in the first level.
14
    schema structure likes
15
    ```
16
    {
17
        "title":"prd",
18
        "type":"object",
19
        "properties":{
20
            "Original Requirements":{
21
                "title":"Original Requirements",
22
                "type":"string"
23
            },
24
        },
25
        "required":[
26
            "Original Requirements",
27
        ]
28
    }
29
    ```
30
    """
31
    mapping = dict()
32
    for field, property in schema["properties"].items():
33
        if property["type"] == "string":
34
            mapping[field] = (str, ...)
35
        elif property["type"] == "array" and property["items"]["type"] == "string":
36
            mapping[field] = (list[str], ...)
37
        elif property["type"] == "array" and property["items"]["type"] == "array":
38
            # here only consider the `list[list[str]]` situation
39
            mapping[field] = (list[list[str]], ...)
40
    return mapping
41

42

43
def actionoutput_mapping_to_str(mapping: dict) -> dict:
44
    new_mapping = {}
45
    for key, value in mapping.items():
46
        new_mapping[key] = str(value)
47
    return new_mapping
48

49

50
def actionoutput_str_to_mapping(mapping: dict) -> dict:
51
    new_mapping = {}
52
    for key, value in mapping.items():
53
        if value == "(<class 'str'>, Ellipsis)":
54
            new_mapping[key] = (str, ...)
55
        else:
56
            new_mapping[key] = eval(value)  # `"'(list[str], Ellipsis)"` to `(list[str], ...)`
57
    return new_mapping
58

59

60
def serialize_message(message: "Message"):
61
    message_cp = copy.deepcopy(message)  # avoid `instruct_content` value update by reference
62
    ic = message_cp.instruct_content
63
    if ic:
64
        # model create by pydantic create_model like `pydantic.main.prd`, can't pickle.dump directly
65
        schema = ic.model_json_schema()
66
        mapping = actionoutout_schema_to_mapping(schema)
67

68
        message_cp.instruct_content = {"class": schema["title"], "mapping": mapping, "value": ic.model_dump()}
69
    msg_ser = pickle.dumps(message_cp)
70

71
    return msg_ser
72

73

74
def deserialize_message(message_ser: str) -> "Message":
75
    message = pickle.loads(message_ser)
76
    if message.instruct_content:
77
        ic = message.instruct_content
78
        actionnode_class = import_class("ActionNode", "metagpt.actions.action_node")  # avoid circular import
79
        ic_obj = actionnode_class.create_model_class(class_name=ic["class"], mapping=ic["mapping"])
80
        ic_new = ic_obj(**ic["value"])
81
        message.instruct_content = ic_new
82

83
    return message
84