MetaGPT 源码拆解：一行需求如何跑成一家「软件公司」

MetaGPT 不是「多个 Agent 拼在一起」，而是把一家软件公司的 SOP 物化成可运行代码：Code = SOP(Team)。 本文基于 commit 11cdf46（v1.0.0）源码做纯技术拆解，只讲工程实现：三个核心抽象、一条消息链路、一套 LLM 无关分层，以及它们如何串起「一行需求 → 可运行软件仓库」的端到端流水线。

源码 metagpt/ 包自底向上大致七层：

关键设计是分层单向依赖：上层（入口）只消费下层（配置/LLM）的公开接口，角色与行为层不直接碰具体 Provider——这让「换模型」从代码问题退化成配置问题（见 §六）。

2.1 基座链

BaseSerialization（pydantic BaseModel，extra="forbid"）→ SerializationMixin（serialize/deserialize/get_serialization_path）→ ContextMixin（context_mixin.py:17，公共/私有 context 配置 + llm 三件套，私有优先否则回落公共）。

2.2 Role 组合结构

Role（roles/role.py:125）= BaseRole + SerializationMixin + ContextMixin + BaseModel，持有：

•RoleContext.rc：env / msg_buffer（异步消息队列）/ memory / working_memory / watch（订阅标签集）/ todo / react_mode / max_react_loop；

•Planner（strategy/planner.py）：PLAN_AND_ACT 模式的计划器；

•actions：装配的行为列表。

2.3 react 三态（role.py:82 RoleReactMode）

核心循环在 _react（role.py:454）：while actions_taken < max_react_loop: _think() → _act()。REACT 模式下「选什么动作」本身就是一次 LLM 推理，这是 MetaGPT 与固定流程图的分水岭——角色是状态机，但状态转移由模型决定。

3.1 Action：执行单元

Action（actions/action.py:29），关键字段 node: ActionNode。run()（action.py:110）两条路径：有 node 走 _run_action_node（组装需求 + 调 LLM）；无 node 由子类实现 run（如 WritePRD、WriteCode）。

3.2 ActionNode：树形结构 + 运行时生成的校验模型

ActionNode（actions/action_node.py:135）是普通对象（非 pydantic），核心特征：

•树形节点：children / prevs / nexts，可以递归嵌套定义输出结构；

•动态生成校验模型（action_node.py:248 create_model_class）：按节点的 field mapping 用 pydantic v2 create_model 运行时构造校验类，带 model_validator(before) 做必填字段缺失检查 + 未知字段告警；遇到嵌套 dict 递归生成子模型类（{class_name}_{field_name}）；

•多填充策略：simple/complex、code_fill / xml_fill / single_fill 等，适配不同输出格式（JSON/XML/代码）；

•评审机制：HUMAN（人工确认）/ AUTO（LLM 自评）两种评审路径；

•经验池缓存：fill 方法带 @exp_cache（action_node.py:596），命中过的填充结果直接复用。

工程含义：上游角色产出的 PRD/设计必须是「结构正确」的——不是靠提示词碰运气，而是用运行时构造的 pydantic 模型强制校验，不符合字段约束就过不去。这是「结构化中间产物」的技术底座。

Message（schema.py:232）字段：content / cause_by / sent_from / send_to / role / instruct_content；send_to 默认 MESSAGE_ROUTE_TO_ALL，子类 UserMessage / SystemMessage / AIMessage。

消息传递拆成三层，职责分离（RFC 116 落地）：

1publish_message（role.py:429）：角色把消息发给 Environment，由 env 广播；

2put_message（role.py:448）：消息进入角色私有
msg_buffer（异步队列）；

3_observe（role.py:399）：从 buffer 取出消息，按 (cause_by in watch or 自己的名字 in send_to) 过滤去重后写入 memory。

路由发生在 Environment.publish_message（environment/base_env.py:175）：遍历 member_addrs，对每条消息按 is_send_to(message, addrs) 匹配订阅者，命中则 role.put_message。订阅关系由 _watch([Action类型]) 声明——协作关系即订阅关系，加一个新角色不需要改主流程。

5.1 软件公司模式入口（software_company.py:14 generate_repo）

5.2 调度循环（team.py:123 run）

Environment.run（base_env.py:197）用 asyncio.gather
并行驱动每个 role.run()，角色之间靠消息订阅异步衔接，不是同步函数调用。

5.3 角色间 SOP 协作链（消息订阅驱动）

TeamLeader（RoleZero，含 Plan 工具）拆解需求 → ProductManager（watch UserRequirement）产出 PRD → Architect（watch WritePRD）产出系统设计/API → Engineer2 产出代码 → DataAnalyst 做数据相关分析 → 代码经 ProjectRepo 落盘为仓库。这就是「SOP 即代码」：流程不是脚本，是 Team + 订阅关系的运行时行为。

产物结构（utils/project_repo.py + const.py:86-106）：

6.1 调用链

6.2 注册机制

LLMProviderRegistry（provider/llm_provider_registry.py:12）维护 {key: cls}；register_provider(keys) 装饰器副作用注册——import provider 模块即自动注册，新增 Provider 只需新增一个文件 + 一个装饰器。LLMType.__missing__（configs/llm_config.py:48）对未知 api_type 静默回落 OpenAI（不报错，排障时注意）。

6.3 api_type → 实现类映射（节选）

BaseLLM（provider/base_llm.py:35）抽象 __init__ / _achat_completion / acompletion / aask…，通用能力包括 format_msg、get_choice_text、aask_batch、消息压缩 compress_messages、成本 _update_costs。

6.4 配置加载优先级

Config.default()（config2.py:109）合并两处 yaml 后，最终 dict 优先级：os.environ
→ 仓库 config2.yaml → ~/.metagpt/config2.yaml → 代码 kwargs，结果缓存于模块级 _CONFIG_CACHE。api_key 缺失在 LLMConfig 校验阶段直接报错（configs/llm_config.py:116）。

**经验池（exp_pool）**是独立亮点：@exp_cache 装饰器（exp_pool/decorator.py:29）把「输入上下文 → 输出结果」落库，命中的经验直接复用，未命中才调 LLM。使用点包括 RoleZero._plan_and_act（role_zero.py:267）与 ActionNode.fill（action_node.py:596）——这是「越跑越快」的机制来源：同一类任务第二次执行会命中缓存，省一次推理。

•RAG 模块（metagpt/rag/）：retrievers（faiss / es / chroma / bm25 / hybrid / base）+ rankers（LLM 重排 object_ranker）+ engines（simple.py 基础流水线 / flare.py）+ factories（retriever/ranker/llm/embedding/index 工厂），支撑客服 FAQ 等场景；

•Document Store（metagpt/document_store/）：base_store.py 统一抽象，提供 chromadb / faiss / lancedb / milvus / qdrant 五种向量库实现——向量库的选择也是配置问题，不污染业务代码。

1SOP 即代码：软件公司流程 = Team + 订阅关系 + Action 序列，加角色/改流程不动主循环；

2结构化中间产物强制校验：ActionNode 运行时构造 pydantic 模型，上游输出结构不对就过不了（必填字段缺失直接 ValueError）；

3可序列化恢复：Team/Context 支持序列化，recover_path 从已序列化的 team 目录恢复续跑（software_company.py:63）——长任务可中断可续跑；

4角色无状态化可选：enable_memory=False 可关闭记忆，原子型角色省内存；

5react 策略可插拔：一个 Role 通过 react_mode 在「LLM 选动作 / 固定顺序 / 计划驱动」间切换，同一套抽象覆盖从简单问答到计划执行的复杂度梯度。

一句话总结这套工程架构的边界：MetaGPT 的强项是「多角色 + 结构化产物 + 强校验」——用流程约束单模型的长上下文与幻觉；而它的一次性团队生命周期、无跨任务主体记忆，也正对应 Agent 演进谱系里「第一代任务流水线」的天花板位置（这也正是常驻主体型范式要补的课，这里不展开）。

技术拆解就到这。想深入哪一层（ActionNode 校验 / 消息路由 / Provider 接入），欢迎留言聊聊；也可以顺着文末的联系方式找到我们。

— 月牙寂 × 月绡凝（人机协作）