mirror of
https://github.com/FoundationAgents/MetaGPT.git
synced 2026-07-03 10:04:37 +08:00
84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
# @Desc : the implement of serialization and deserialization
|
|
|
|
import copy
|
|
import pickle
|
|
|
|
from metagpt.utils.common import import_class
|
|
|
|
|
|
def actionoutout_schema_to_mapping(schema: dict) -> dict:
|
|
"""
|
|
directly traverse the `properties` in the first level.
|
|
schema structure likes
|
|
```
|
|
{
|
|
"title":"prd",
|
|
"type":"object",
|
|
"properties":{
|
|
"Original Requirements":{
|
|
"title":"Original Requirements",
|
|
"type":"string"
|
|
},
|
|
},
|
|
"required":[
|
|
"Original Requirements",
|
|
]
|
|
}
|
|
```
|
|
"""
|
|
mapping = dict()
|
|
for field, property in schema["properties"].items():
|
|
if property["type"] == "string":
|
|
mapping[field] = (str, ...)
|
|
elif property["type"] == "array" and property["items"]["type"] == "string":
|
|
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]], ...)
|
|
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_message(message: "Message"):
|
|
message_cp = copy.deepcopy(message) # avoid `instruct_content` value update by reference
|
|
ic = message_cp.instruct_content
|
|
if ic:
|
|
# model create by pydantic create_model like `pydantic.main.prd`, can't pickle.dump directly
|
|
schema = ic.model_json_schema()
|
|
mapping = actionoutout_schema_to_mapping(schema)
|
|
|
|
message_cp.instruct_content = {"class": schema["title"], "mapping": mapping, "value": ic.model_dump()}
|
|
msg_ser = pickle.dumps(message_cp)
|
|
|
|
return msg_ser
|
|
|
|
|
|
def deserialize_message(message_ser: str) -> "Message":
|
|
message = pickle.loads(message_ser)
|
|
if message.instruct_content:
|
|
ic = message.instruct_content
|
|
actionnode_class = import_class("ActionNode", "metagpt.actions.action_node") # avoid circular import
|
|
ic_obj = actionnode_class.create_model_class(class_name=ic["class"], mapping=ic["mapping"])
|
|
ic_new = ic_obj(**ic["value"])
|
|
message.instruct_content = ic_new
|
|
|
|
return message
|