🏗️ MetaGPT 深度源码分析

基于源码的 SOP-Driven 多角色协作框架分析 · v0.8+ · 2026-05

📌 项目概览

🏛️ 整体架构

架构核心：四层模型

🤖 Role 核心类

Role 是 MetaGPT 的核心抽象，每个 Role 就是一个自主代理。其核心循环为 _observe → _think → _act → publish_message。

BaseRole (ABC)           # 抽象接口
  └─ Role (Pydantic)     # 核心实现，含 _observe/_think/_act/_react
       └─ RoleZero       # 新一代动态角色 (v0.8+)
            ├─ ProductManager (Alice)
            ├─ Architect (Bob)
            ├─ ProjectManager (Eve)
            ├─ DataAnalyst
            ├─ TeamLeader
            └─ Engineer2

Role 核心属性

class Role(BaseRole, SerializationMixin, ContextMixin, BaseModel):
    name: str = ""                    # 角色名称，如 "Alice"
    profile: str = ""                 # 角色画像，如 "Product Manager"
    goal: str = ""                    # 目标描述
    constraints: str = ""             # 约束条件
    actions: list[Action] = []        # 角色可执行的动作列表
    rc: RoleContext = RoleContext()    # 运行时上下文 (状态机核心)
    planner: Planner = Planner()      # 内置规划器 (PLAN_AND_ACT 模式)
    observe_all_msg_from_buffer: bool = False  # 是否接收所有消息

RoleContext: 运行时状态机

class RoleContext(BaseModel):
    env: BaseEnvironment              # 所属环境引用
    msg_buffer: MessageQueue          # 私有消息缓冲区
    memory: Memory                    # 长期记忆 (按 action 索引)
    working_memory: Memory            # 工作记忆 (PLAN_AND_ACT 用)
    state: int = -1                   # 当前状态 (-1 = 空闲)
    todo: Action = None               # 当前待执行动作
    watch: set[str]                   # 关注的 Action 类型集合
    news: list[Message]               # 本轮新观察到的消息
    react_mode: RoleReactMode         # React 策略模式
    max_react_loop: int = 1           # 最大 react 循环次数

_observe(): 消息接收与过滤

async def _observe(self) -> int:
    # 1. 从 msg_buffer 取出未处理消息
    news = self.rc.msg_buffer.pop_all()
    
    # 2. 按 watch 列表过滤: 只保留 cause_by 在 watch 中的消息
    #    或者 send_to 包含自己名字的消息
    self.rc.news = [
        n for n in news 
        if (n.cause_by in self.rc.watch or self.name in n.send_to) 
           and n not in old_messages
    ]
    
    # 3. 存入 memory (防止重复处理)
    self.rc.memory.add_batch(self.rc.news)
    
    return len(self.rc.news)  # 返回新消息数量

_think(): 状态转移决策

async def _think(self) -> bool:
    # 单一 Action: 直接进入 state 0
    if len(self.actions) == 1:
        self._set_state(0)
        return True
    
    # BY_ORDER 模式: 按顺序切换状态
    if self.rc.react_mode == RoleReactMode.BY_ORDER:
        self._set_state(self.rc.state + 1)
        return self.rc.state >= 0 and self.rc.state < len(self.actions)
    
    # REACT 模式: 用 LLM 选择下一个状态
    prompt = self._get_prefix() + STATE_TEMPLATE.format(
        history=self.rc.history, states="\n".join(self.states),
        n_states=len(self.states) - 1, previous_state=self.rc.state,
    )
    next_state = await self.llm.aask(prompt)
    self._set_state(int(next_state))
    return True

_act(): 执行动作

async def _act(self) -> Message:
    # 执行当前 todo Action
    response = await self.rc.todo.run(self.rc.history)
    
    # 将结果包装为 AIMessage
    if isinstance(response, (ActionOutput, ActionNode)):
        msg = AIMessage(content=response.content, 
                        instruct_content=response.instruct_content,
                        cause_by=self.rc.todo, sent_from=self)
    elif isinstance(response, Message):
        msg = response
    else:
        msg = AIMessage(content=response or "", 
                        cause_by=self.rc.todo, sent_from=self)
    
    self.rc.memory.add(msg)  # 存入记忆
    return msg

run(): 主循环入口

@role_raise_decorator
async def run(self, with_message=None) -> Message | None:
    if with_message:
        self.put_message(msg)       # 放入 buffer
    
    if not await self._observe():   # 观察: 读取并过滤消息
        return                      # 无新消息则等待
    
    rsp = await self.react()        # 反应: _think → _act (可多轮)
    
    self.set_todo(None)             # 重置 todo
    self.publish_message(rsp)       # 发布结果到 Environment
    return rsp

⚡ Action 动作类

Action 是角色能力的原子单位。每个 Action 封装了一次与 LLM 的交互，以及输入输出结构定义。

class Action(SerializationMixin, ContextMixin, BaseModel):
    name: str = ""           # 动作名称 (默认为类名)
    i_context: Union[...]    # 输入上下文 (CodingContext/TestingContext 等)
    prefix: str = ""         # system_message 前缀
    desc: str = ""           # 描述 (给 SkillManager 用)
    node: ActionNode = None  # 结构化输出节点 (如果为 None 则需子类重写 run)
    llm_name_or_type: str    # 指定 LLM 配置

Action.run() 分支逻辑

async def run(self, *args, **kwargs):
    if self.node:
        # 有 ActionNode: 走结构化填充流程
        return await self._run_action_node(*args, **kwargs)
    raise NotImplementedError("子类必须实现 run 方法")

关键 Action 子类

🌐 Environment 环境

Environment 是消息路由中心，承担了"消息总线"的角色。

核心方法

def publish_message(self, message: Message, peekable: bool = True) -> bool:
    """按 RFC 113 路由设计: 消息的 send_to 只指定接收者，
       不关心接收者在哪里 — 路由由 Environment 负责"""
    for role, addrs in self.member_addrs.items():
        if is_send_to(message, addrs):  # 匹配 addresses
            role.put_message(message)    # 推入角色私有 buffer
    self.history.add(message)            # 全局历史 (调试用)
    return True

async def run(self, k=1):
    for _ in range(k):
        futures = []
        for role in self.roles.values():
            if role.is_idle:
                continue
            future = role.run()
            futures.append(future)
        if futures:
            await asyncio.gather(*futures)  # 所有角色并行执行

ExtEnv: 外部环境扩展

MetaGPT 还提供了 ExtEnv 抽象类，用于集成外部环境 (Android/Minecraft/StanfordTown/Werewolf)。它引入了 action_space/observation_space (gymnasium 兼容) 和 read_from_api/write_thru_api 的 API 注册机制。

📨 Message 消息

Message 是 MetaGPT 的核心数据单元，承担了消息路由、上下文传递和序列化的多重职责。

class Message(BaseModel):
    id: str = uuid4().hex                       # 唯一标识
    content: str                                 # 自然语言内容
    instruct_content: Optional[BaseModel] = None  # 结构化内容 (ActionNode 生成的)
    role: str = "user"                          # system / user / assistant
    cause_by: str = "UserRequirement"           # 产生此消息的 Action 类型
    sent_from: str = ""                         # 发送者标识
    send_to: set[str] = {MESSAGE_ROUTE_TO_ALL}  # 接收者集合
    metadata: Dict[str, Any] = {}               # 附加元数据

路由机制: cause_by + send_to 双通道

Message 类型层次

Message
├── UserMessage (role="user")    # 用户输入
├── SystemMessage (role="system") # 系统指令
└── AIMessage (role="assistant")  # Agent 输出
    └── .with_agent(name)         # 附加 agent 元数据
    └── .agent 属性               # 读取 agent 名称

📬 MessageQueue

class MessageQueue(BaseModel):
    _queue: Queue = PrivateAttr(default_factory=Queue)
    
    def pop(self) -> Message | None:     # 取一条
    def pop_all(self) -> List[Message]:  # 取全部
    def push(self, msg: Message):        # 推入一条
    def empty(self) -> bool:             # 判空
    async def dump(self) -> str:         # 序列化 (调试用)

MessageQueue 是基于 asyncio.Queue 的轻量封装，作为每个 Role 的私有消息缓冲区。Environment 的 publish_message 通过 role.put_message(message) 将消息推入此队列，Role 在 _observe 时通过 pop_all() 取出所有未处理消息。

🔄 SOP 编排 (Standard Operating Procedure)

MetaGPT 的核心创新是将软件开发过程固化为 SOP 流水线。每个角色"关注"上游角色的输出，当观察到上游完成时自动开始工作。

SOP 的两种模式

🔀 3 种 React 模式

class RoleReactMode(str, Enum):
    REACT = "react"           # 标准 Think-Act 循环
    BY_ORDER = "by_order"     # 按 Action 顺序执行
    PLAN_AND_ACT = "plan_and_act"  # 先规划再执行

1. REACT 模式

async def _react(self) -> Message:
    """标准 think-act 循环，交替思考和行动"""
    actions_taken = 0
    rsp = AIMessage(content="No actions taken yet", cause_by=Action)
    while actions_taken < self.rc.max_react_loop:
        has_todo = await self._think()  # LLM 选择下一步
        if not has_todo:
            break
        rsp = await self._act()          # 执行选中的 Action
        actions_taken += 1
    return rsp

2. BY_ORDER 模式

# _think() 中:
if self.rc.react_mode == RoleReactMode.BY_ORDER:
    self._set_state(self.rc.state + 1)
    return self.rc.state >= 0 and self.rc.state < len(self.actions)

3. PLAN_AND_ACT 模式

async def _plan_and_act(self) -> Message:
    # 首次: 创建计划
    if not self.planner.plan.goal:
        goal = self.rc.memory.get()[-1].content
        await self.planner.update_plan(goal=goal)
    
    # 循环: 逐任务执行
    while self.planner.current_task:
        task = self.planner.current_task
        task_result = await self._act_on_task(task)  # 子类实现
        await self.planner.process_task_result(task_result)
    
    return rsp

🚀 RoleZero: 新一代动态角色

RoleZero 是 MetaGPT v0.8+ 引入的新一代角色基类，取代了旧式 SOP 角色的大部分功能。它是一个"万能角色"，通过工具调用动态决定下一步操作。

class RoleZero(Role):
    # React 配置
    react_mode: Literal["react"] = "react"  # 强制 react 模式
    max_react_loop: int = 50                # 最多 50 轮
    
    # 工具系统
    tools: list[str] = []                   # 工具列表
    tool_recommender: BM25ToolRecommender   # BM25 工具推荐器
    tool_execution_map: dict[str, Callable] # 工具名 → 函数映射
    
    # 内置三大工具
    editor: Editor = Editor(enable_auto_lint=True)
    browser: Browser = Browser()
    
    # 经验系统
    experience_retriever: ExpRetriever
    
    # 其他
    observe_all_msg_from_buffer: bool = True # 接收所有消息
    memory_k: int = 200                     # 记忆窗口
    use_fixed_sop: bool = False             # 不使用固定 SOP
    use_summary: bool = True                # 结束时总结

RoleZero 的 _think() 流程

async def _think(self) -> bool:
    # 0. 准备: 设置 planner goal，检测语言
    if not self.planner.plan.goal:
        self.planner.plan.goal = self.get_memories()[-1].content
        self.respond_language = await self.llm.aask(DETECT_LANGUAGE_PROMPT)
    
    # 1. 经验检索: 获取相似任务的历史经验
    example = self._retrieve_experience()
    
    # 2. 计划状态: 当前任务和进度
    plan_status, current_task = get_plan_status(planner=self.planner)
    
    # 3. 工具推荐: BM25 从工具列表中筛选相关工具
    tools = await self.tool_recommender.recommend_tools()
    tool_info = json.dumps({tool.name: tool.schemas for tool in tools})
    
    # 4. 组装 System Prompt: 角色 + 工具 + 经验 + 指令
    system_prompt = SYSTEM_PROMPT.format(
        role_info=self._get_prefix(), available_commands=tool_info,
        example=example, instruction=self.instruction)
    
    # 5. 组装 User Prompt: 计划状态 + 当前任务 + 语言
    prompt = CMD_PROMPT.format(
        plan_status=plan_status, current_task=current_task,
        respond_language=self.respond_language)
    
    # 6. 调用 LLM (带经验缓存)
    self.command_rsp = await self.llm_cached_aask(req=req, system_msgs=[system_prompt])
    
    # 7. 去重检查: 避免重复命令
    self.command_rsp = await check_duplicates(...)
    
    return True

RoleZero 的 _act() 流程

async def _act(self) -> Message:
    # 1. 解析 LLM 输出为命令列表
    commands, ok, self.command_rsp = await parse_commands(
        command_rsp=self.command_rsp, llm=self.llm)
    
    # 2. 执行命令
    outputs = await self._run_commands(commands)
    
    # 3. 将结果存入 memory
    self.rc.memory.add(AIMessage(content=self.command_rsp))
    self.rc.memory.add(UserMessage(content=outputs, cause_by=RunCommand))
    
    return AIMessage(content=f"I have finished the task...")

Quick Think: 快速响应机制

RoleZero 实现了"快思考"机制，对简单问题直接回答，不走完整的 think-act 循环:

# 分类: QUICK / SEARCH / TASK / AMBIGUOUS
quick_rsp, _ = await self._quick_think()
if quick_rsp:
    return quick_rsp  # 简单问题直接返回

内置工具执行映射

tool_execution_map = {
    "Plan.append_task": planner.plan.append_task,
    "Plan.reset_task": planner.plan.reset_task,
    "Plan.replace_task": planner.plan.replace_task,
    "RoleZero.ask_human": self.ask_human,
    "RoleZero.reply_to_human": self.reply_to_human,
    "SearchEnhancedQA.run": SearchEnhancedQA().run,
    "Browser.click/close_tab/goto/...": browser.*,
    "Editor.open_file/write/edit_file/...": editor.*,
    "Terminal.run_command": terminal.run_command,
    # 子类可扩展: self._update_tool_execution()
}

🏢 Software Company 多角色协作

Team: 编排层

class Team(BaseModel):
    env: Environment        # 消息环境
    investment: float = 10  # 预算上限
    idea: str = ""          # 项目创意
    
    def hire(self, roles):       # 招聘角色
        self.env.add_roles(roles)
    
    def invest(self, amount):    # 注资 (设置 max_budget)
        self.cost_manager.max_budget = amount
    
    async def run(self, n_round=3, idea=""):
        self.run_project(idea)          # 发布初始需求
        while n_round > 0:
            if self.env.is_idle: break  # 所有角色空闲则结束
            await self.env.run()         # 执行一轮
            self._check_balance()        # 检查预算
            n_round -= 1

典型组建方式 (software_company.py)

company = Team(context=ctx)
company.hire([
    TeamLeader(),
    ProductManager(),
    Architect(),
    Engineer2(),
    DataAnalyst(),
])
company.invest(investment)
asyncio.run(company.run(n_round=n_round, idea=idea))

👩‍💼 ProductManager (Alice)

class ProductManager(RoleZero):
    name: str = "Alice"
    profile: str = "Product Manager"
    goal: str = "Create a PRD or market research/competitive product research."
    tools: list[str] = ["RoleZero", "Browser", "Editor", "SearchEnhancedQA"]
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        if self.use_fixed_sop:
            self.enable_memory = False
            self.set_actions([PrepareDocuments, WritePRD])
            self._watch([UserRequirement, PrepareDocuments])
            self.rc.react_mode = RoleReactMode.BY_ORDER

WritePRD: 核心动作

WritePRD 使用 ActionNode 系统生成结构化 PRD:

WRITE_PRD_NODE = ActionNode(
    key="prd", expected_type=list,
    instruction="输出完整的 PRD 文档，包含: ...",
    # 包含: Language, Project Name, Product Goals, User Stories,
    #       Competitive Analysis, Requirements Pool, UI Design 等
)

ProductManager 的提示词模板分为两种模式:

• PRD 模式: 输出产品需求文档 (含 Mermaid 竞品象限图)
• Market Research 模式: 使用 SearchEnhancedQA 工具进行市场调研

🏗️ Architect (Bob)

class Architect(RoleZero):
    name: str = "Bob"
    profile: str = "Architect"
    goal: str = "design a concise, usable, complete software system"
    tools: list[str] = ["Editor:write,read,similarity_search", 
                        "RoleZero", "Terminal:run_command"]
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([WriteDesign])
        self._watch({WritePRD})  # 关注 PRD 完成

WriteDesign 输出结构

WriteDesign 通过 ActionNode 生成三部分设计:

DESIGN_API_NODE:
  1. DATA_STRUCTURES_AND_INTERFACES  # 数据结构与接口 (类图)
  2. PROGRAM_CALL_FLOW               # 程序调用流程 (序列图)
  3. REFINED_DESIGN_NODE             # 增量更新时的精化设计

📋 ProjectManager (Eve)

class ProjectManager(RoleZero):
    name: str = "Eve"
    profile: str = "Project Manager"
    goal: str = "break down tasks according to PRD/technical design"
    tools: list[str] = ["Editor:write,read,similarity_search", 
                        "RoleZero", "WriteTasks"]
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.set_actions([WriteTasks])
        self._watch([WriteDesign])  # 关注设计完成

👨‍💻 Engineer (Alex)

Engineer 是最复杂的角色，因为它需要处理多种消息来源和内部循环。

class Engineer(Role):  # 注意: 仍用旧式 Role，非 RoleZero
    name: str = "Alex"
    profile: str = "Engineer"
    use_code_review: bool = False
    
    def __init__(self, **kwargs):
        self.set_actions([WriteCode])
        # 关注 6 种消息来源!
        self._watch([WriteTasks, SummarizeCode, WriteCode, 
                     WriteCodeReview, FixBug, WriteCodePlanAndChange])
    
    async def _think(self) -> bool:
        msg = self.rc.news[0]
        # 根据 cause_by 决定下一步:
        # - WriteTasks/FixBug → WriteCodePlanAndChange (增量模式)
        # - WriteTasks/WriteCodePlanAndChange/SummarizeCode → WriteCode
        # - WriteCode/WriteCodeReview → SummarizeCode

Engineer 的内部循环

🧪 QaEngineer (Edward)

class QaEngineer(Role):  # 旧式 Role
    name: str = "Edward"
    profile: str = "QaEngineer"
    test_round_allowed: int = 5
    
    def __init__(self, **kwargs):
        self.set_actions([WriteTest])
        self._watch([SummarizeCode, WriteTest, RunCode, DebugError])

QA 内部循环

🌳 ActionNode: 结构化输出系统

ActionNode 是 MetaGPT 独创的结构化输出机制，比 function calling 更灵活。它支持 Review-Revise 循环来保证输出质量。

class ActionNode:
    key: str                # 节点名称
    expected_type: Type     # 期望类型 (str, list, int, ...)
    instruction: str        # 给 LLM 的指令
    example: str            # 格式示例
    schema: str             # 输出格式: "raw"/"json"/"markdown"
    
    # 子节点 (树形结构)
    children: list[ActionNode] = []
    
    # 填充模式
    fill_mode: FillMode     # CODE_FILL / XML_FILL / SINGLE_FILL
    
    # 审查模式
    review_mode: ReviewMode # HUMAN / AUTO
    revise_mode: ReviseMode # HUMAN / HUMAN_REVIEW / AUTO

ActionNode 填充流程

async def fill(self, req, llm):
    # 1. 组装 Prompt: context + format_example + nodes_instruction
    prompt = SIMPLE_TEMPLATE.format(
        context=req, example=self.example,
        instruction=self._compile_instruction(), 
        constraint=FORMAT_CONSTRAINT)
    
    # 2. 调用 LLM
    output = await llm.aask(prompt)
    
    # 3. 解析输出 (根据 schema)
    # - "raw": 直接提取 [CONTENT]...[/CONTENT] 标签内容
    # - "json": JSON 解析
    # - "markdown": Markdown 解析
    
    # 4. Review (可选)
    if self.review_mode != ReviewMode.HUMAN:
        review_result = await self._review(output)
    
    # 5. Revise (可选，基于 review 结果)
    if self.revise_mode != ReviseMode.HUMAN:
        output = await self._revise(output, review_result)
    
    return ActionOutput(content=output, instruct_content=parsed)

ActionNode 树形示例: WritePRD

📝 提示词模板系统

MetaGPT 的提示词分为多个层次，从通用到角色特定逐级叠加:

层次 1: 角色身份 (PREFIX_TEMPLATE)

PREFIX_TEMPLATE = "You are a {profile}, named {name}, your goal is {goal}. "
CONSTRAINT_TEMPLATE = "the constraint is {constraints}. "
# 如果有环境: "You are in {env.desc} with roles({other_role_names})."

层次 2: 角色指令 (instruction)

每个 RoleZero 子类都有 instruction 属性，定义了角色的工作方式。例如 ProductManager 的指令包含 PRD 和 Market Research 两种模式。

层次 3: 系统提示词 (SYSTEM_PROMPT)

SYSTEM_PROMPT = """
# Basic Info
{role_info}

# Data Structure
class Task(BaseModel): ...

# Available Commands  
{available_commands}

# Example
{example}

# Instruction
{instruction}
"""

层次 4: 命令提示词 (CMD_PROMPT)

CMD_PROMPT = """
# Past Experience
{experience}

# Tool State
{current_state}

# Current Plan
{plan_status}

# Current Task
{current_task}

# Response Language
you must respond in {respond_language}.

# Your commands in a json array:
```json
[{"command_name": "...", "args": {...}}]
```
"""

层次 5: ActionNode 模板

SIMPLE_TEMPLATE = """
## context
{context}

## format example
{example}

## nodes: "<node>: <type>  # <instruction>"
{instruction}

## constraint
{constraint}

## action
Follow instructions of nodes, generate output...
"""

🧠 记忆系统

MetaGPT 实现了三层记忆架构:

Memory 核心方法

class Memory(BaseModel):
    storage: list[Message] = []
    index: DefaultDict[str, list[Message]] = defaultdict(list)
    
    def add(self, message):            # 添加 + 更新索引
    def add_batch(self, messages):     # 批量添加
    def get_by_action(self, action):   # 按 Action 类型查询
    def get_by_actions(self, actions): # 批量查询 (watch 机制核心)
    def find_news(self, observed):     # 找未观察到的消息
    def try_remember(self, keyword):   # 关键词搜索
    def get(self, k=0):               # 取最近 k 条

RoleZeroLongTermMemory 溢出机制

def add(self, message):
    super().add(message)  # 先存短期
    if self.count() > self.memory_k:  # 超过容量
        self._transfer_to_longterm_memory()  # 溢出到长期

def get(self, k=0):
    memories = super().get(k)  # 短期记忆
    if self._should_use_longterm_memory_for_get(k):
        query = self._build_longterm_memory_query()  # 最近用户消息
        related = self._fetch_longterm_memories(query)  # RAG 检索
        memories = related + memories  # 合并: 长期 + 短期
    return memories

💎 Experience Pool: 语义缓存

Experience Pool 是 MetaGPT 独创的 LLM 调用缓存机制，通过语义相似度匹配历史调用，避免重复请求 LLM。

@exp_cache(context_builder=RoleZeroContextBuilder(), 
            serializer=RoleZeroSerializer())
async def llm_cached_aask(self, *, req, system_msgs, **kwargs):
    return await self.llm.aask(req, system_msgs=system_msgs)

Experience Pool 工作流程

Experience 数据模型

class Experience(BaseModel):
    uuid: str
    req: str         # 请求 (序列化后)
    resp: str        # 响应 (序列化后)
    tag: str         # 标签 (如 "RoleZero.llm_cached_aask")
    metric: Metric   # 评分 (score)

⚖️ 与 AutoGen / CrewAI 的对比

🔑 关键发现

📐 设计模式提炼

1. 观察者模式 (Watch-Publish)

角色通过 _watch 订阅感兴趣的消息类型，Environment 充当消息总线。这是 GoF 观察者模式的异步变体。

2. 状态机模式 (RoleContext)

RoleContext 中的 state 和 todo 构成了隐式状态机:state → actions[state] → todo → _act()。

3. 策略模式 (ReactMode)

三种 React 模式通过 RoleReactMode 枚举切换，react() 方法作为策略调度入口。

4. 模板方法模式 (Role.run)

run() 定义了骨架流程 (observe → react → publish)，子类通过重写 _think/_act 定制行为。

5. 装饰器模式 (exp_cache)

exp_cache 装饰器透明地为 LLM 调用添加缓存层，不改变原函数签名。

6. 代理模式 (Environment → Role)

Environment 代理消息路由，Role 无需知道其他 Role 的存在，只通过 Environment 通信。

7. 建造者模式 (ActionNode 树)

ActionNode 通过子节点的级联构建复杂的输出结构，类似建造者模式的逐步组装。