序列化、状态恢复与成本追踪

学习目标

理解 MetaGPT 的序列化机制、中断恢复流程和成本追踪系统。

---

项目实践

序列化基础

MetaGPT 使用 Pydantic model_dump() + JSON 文件实现序列化：

class BaseSerialization(BaseModel):
    """所有可序列化类的基类"""
    pass

class SerializationMixin(BaseSerialization):
    def serialize(self, file_path: str = None) -> str:
        """将实例序列化到 JSON 文件"""
        file_path = file_path or self.get_serialization_path()
        data = self.model_dump()
        write_json_file(file_path, data, use_fallback=True)
        return file_path

    @classmethod
    def deserialize(cls, file_path: str = None) -> BaseModel:
        """从 JSON 文件反序列化"""
        data = read_json_file(file_path)
        return cls(**data)

    @classmethod
    def get_serialization_path(cls) -> str:
        """默认路径：./workspace/storage/ClassName.json"""
        return str(SERDESER_PATH / f"{cls.__qualname__}.json")

Team 序列化

Team 的序列化需要保存团队信息和上下文配置：

def serialize(self, stg_path: Path = None):
    stg_path = SERDESER_PATH.joinpath("team") if stg_path is None else stg_path
    team_info_path = stg_path.joinpath("team.json")

    data = self.model_dump()
    data["context"] = self.env.context.serialize()  # 序列化 Context
    write_json_file(team_info_path, data)

反序列化：

@classmethod
def deserialize(cls, stg_path: Path, context: Context = None) -> "Team":
    team_info_path = stg_path.joinpath("team.json")
    team_info = read_json_file(team_info_path)

    ctx = context or Context()
    ctx.deserialize(team_info.pop("context", None))  # 恢复 Context
    team = Team(**team_info, context=ctx)
    return team

中断恢复

通过 recover_path 参数从序列化存储恢复：

def generate_repo(..., recover_path=None):
    if not recover_path:
        # 新项目
        company = Team(context=ctx)
        company.hire([TeamLeader(), ProductManager(), Architect(), Engineer2(), DataAnalyst()])
    else:
        # 从存储恢复
        stg_path = Path(recover_path)
        if not stg_path.exists() or not str(stg_path).endswith("team"):
            raise FileNotFoundError(f"{recover_path} not exists or not endswith `team`")
        company = Team.deserialize(stg_path=stg_path, context=ctx)
        idea = company.idea  # 恢复原始需求

CLI 使用：

metagpt "继续开发" --recover-path /workspace/storage/team

恢复时的状态定位

Role 使用 latest_observed_msg 定位最后处理位置：

class Role(...):
    latest_observed_msg: Optional[Message] = None
    recovered: bool = False

    def _process_role_extra(self):
        if self.latest_observed_msg:
            self.recovered = True  # 标记为恢复状态

    async def _observe(self) -> int:
        if self.recovered and self.latest_observed_msg:
            # 从 memory 中查找接近 last_observed_msg 的消息
            news = self.rc.memory.find_news(observed=[self.latest_observed_msg], k=10)
        # ...

恢复流程：

1. deserialize 重建 Role 对象，包括 latest_observed_msg
2. 下次 run() 时 _observe() 检测到 recovered=True
3. 从 memory 中查找 latest_observed_msg 附近的新消息
4. 继续处理，recovered 重置为 False

CostManager 成本追踪

class CostManager:
    total_cost: float = 0.0
    max_budget: float = 100.0
    costs: dict[str, float] = {}  # 按模型统计

    def record_cost(self, model: str, prompt_tokens: int, completion_tokens: int):
        cost = self.calculate_cost(model, prompt_tokens, completion_tokens)
        self.total_cost += cost
        self.costs[model] = self.costs.get(model, 0) + cost

    def get_total_cost(self) -> float:
        return self.total_cost

每个 Provider 在 LLM 调用后记录成本：

# BaseLLM
async def aask(self, msg, system_msgs=None, ...):
    response = await self._achat_completion(messages, ...)
    usage = self.get_usage(messages, response)
    self.cost_manager.record_cost(
        model=self.config.model,
        prompt_tokens=usage["prompt_tokens"],
        completion_tokens=usage["completion_tokens"],
    )
    return self.get_choice_text(response)

预算检查

Team.run() 在每轮前检查预算：

async def run(self, n_round=3, idea="", ...):
    while n_round > 0:
        if self.env.is_idle:
            break
        n_round -= 1
        self._check_balance()   # 检查预算
        await self.env.run()

def _check_balance(self):
    if self.cost_manager.total_cost >= self.cost_manager.max_budget:
        raise NoMoneyException(
            self.cost_manager.total_cost,
            f"Insufficient funds: {self.cost_manager.max_budget}"
        )

异常处理

NoMoneyException 是自定义异常：

class NoMoneyException(Exception):
    def __init__(self, cost: float, message: str):
        self.cost = cost
        self.message = message
        super().__init__(message)

抛出后可以捕获并优雅处理：

try:
    await company.run(n_round=5)
except NoMoneyException as e:
    logger.warning(f"Budget exhausted after ${e.cost}: {e.message}")
    # 当前轮次的结果已保存，可以从 team.json 恢复后追加预算

---

问题与规避

序列化不包含 LLM client 状态

• Pydantic model_dump() 排除 exclude=True 的字段（如 aclient 连接池）
• 反序列化后 LLM client 需要重新初始化
• 对策：Context 的 deserialize() 方法重建 LLM 实例

恢复时的消息顺序

• 如果恢复后有大量新消息，find_news(k=10) 只查找最近的 10 条
• 可能遗漏重要的中间消息
• 对策：增大 k 值或在恢复后手动检查 rc.memory

序列化文件大小

• model_dump() 会序列化所有嵌套对象，包括大量 Message
• 对于长对话，序列化文件可能很大（>10MB）
• 对策：限制 memory.storage 的大小，或使用增量序列化

CostManager 不持久化

• CostManager 的数据在序列化时不包含在 team.json 中
• 恢复后 total_cost 从零开始，可能重复消耗预算
• 对策：在序列化时手动保存 cost_manager 状态，恢复时恢复

---

设计取舍

JSON vs Protocol Buffers

• JSON 人类可读、易调试，但序列化/反序列化较慢
• Protocol Buffers 更快更紧凑，但需要 schema 定义
• MetaGPT 选择 JSON 因为 Agent 场景对延迟不敏感，可读性更重要

完整序列化 vs 增量序列化

• 当前实现是全量序列化，每次保存所有状态
• 增量的好处：节省 I/O 和存储空间
• 代价：实现复杂度增加（需要追踪变更）
• 当前选择：全量序列化简单可靠，适合中小规模项目

Budget 按美元计费

• investment 参数以美元为单位，直观易懂
• 代价：需要根据模型定价手动计算 token 预算
• 替代方案：直接使用 token 上限，但不同模型的 token 价值不同

---

参考来源

• 源码验证: metagpt/base/base_serialization.py — BaseSerialization 基类
• 源码验证: metagpt/schema.py:72 — SerializationMixin
• 源码验证: metagpt/team.py:59 — Team.serialize() / deserialize()
• 源码验证: metagpt/software_company.py:63 — recover_path 恢复流程
• 源码验证: metagpt/roles/role.py:156 — latest_observed_msg 恢复定位
• 源码验证: metagpt/utils/cost_manager.py — CostManager 成本追踪
• 源码验证: metagpt/team.py:98 — _check_balance() 预算检查