架构总览

相关源文件

本页面内容基于以下源文件生成：

• metagpt/roles/role.py
• metagpt/actions/action.py
• metagpt/team.py
• metagpt/environment/base_env.py
• metagpt/schema.py
• metagpt/context.py
• metagpt/logs.py
• metagpt/utils/common.py
• metagpt/actions/init.py
• metagpt/const.py
• setup.py
• examples/ui_with_chainlit/app.py
• metagpt/ext/spo/app.py
• metagpt/environment/minecraft/mineflayer/index.js
• metagpt/environment/minecraft/mineflayer/mineflayer-collectblock/src/index.ts
• Dockerfile
• requirements.txt
• .devcontainer/docker-compose.yaml
• examples/stanford_town/requirements.txt
• examples/android_assistant/requirements.txt

MetaGPT 采用多智能体协作架构，通过角色、动作、环境和消息的有机组合，实现复杂任务的自动化处理。该框架的核心设计理念是将软件开发的 SOP（标准作业流程）抽象为智能体间的协作流程，每个智能体扮演特定角色（如产品经理、架构师、工程师等），通过消息传递和状态管理完成端到端的任务执行。

核心概念与基类定义

MetaGPT 的架构建立在三个核心抽象之上：Action（动作）、Role（角色）和 Message（消息）。这三个基类构成了整个多智能体系统的基石。

Action 基类

Action 是智能体行为的原子单元，定义了智能体可以执行的具体操作。每个 Action 包含名称、上下文、LLM 实例和可选的 ActionNode，用于结构化输出。

python
1class Action(SerializationMixin, ContextMixin, BaseModel):
2    name: str = ""
3    i_context: Union[dict, CodingContext, ...] = ""
4    prefix: str = ""  # 作为 system_message
5    node: ActionNode = Field(default=None, exclude=True)
6    llm_name_or_type: Optional[str] = None

Action 基类提供了 run() 方法作为执行入口，子类必须实现此方法。当 Action 包含 ActionNode 时，会调用 _run_action_node() 进行结构化处理；否则抛出 NotImplementedError。关键实现细节参见 metagpt/actions/action.py:29-48。

Role 基类

Role 是智能体的核心抽象，封装了身份信息、行为模式、记忆和状态管理。每个 Role 拥有唯一的名称、角色描述、目标、约束条件，以及一系列可执行的 Action。

python
1class Role(BaseRole, SerializationMixin, ContextMixin, BaseModel):
2    name: str = ""
3    profile: str = ""
4    goal: str = ""
5    constraints: str = ""
6    actions: list[SerializeAsAny[Action]] = Field(default=[])
7    rc: RoleContext = Field(default_factory=RoleContext)
8    planner: Planner = Field(default_factory=Planner)

RoleContext 维护了智能体的运行时状态，包括环境引用、消息缓冲区、记忆、工作记忆、当前状态和待执行动作。详细定义参见 metagpt/roles/role.py:125-152。

Message 消息结构

Message 是智能体间通信的基本单元，承载了内容、发送者、接收者、因果关系等元信息。Message 支持路由机制，通过 send_to 字段指定目标角色，通过 cause_by 字段记录产生该消息的 Action 类型。

python
1class SimpleMessage(BaseModel):
2    content: str
3    role: str

完整的 Message 结构继承自 SimpleMessage，增加了路由和元数据能力。参见 metagpt/schema.py:133-135。

系统架构总览

MetaGPT 的系统架构分为四层：应用层、协调层、智能体层和基础设施层。应用层处理用户交互和项目入口；协调层通过 Team 和 Environment 管理多智能体协作；智能体层包含具体的 Role 和 Action 实现；基础设施层提供配置、日志、序列化等支撑能力。

正在加载图表渲染器...

架构要点说明：

1. 分层解耦：应用层不直接依赖智能体实现，通过 Team 和 Environment 进行间接协调，降低了耦合度
2. 消息驱动：智能体间通过 Environment 进行消息传递，不直接调用彼此的方法，支持异步和分布式扩展
3. 上下文共享：所有智能体共享同一个 Context 实例，统一管理 LLM 配置、成本统计和全局状态
4. 可扩展性：新的智能体只需继承 Role 基类并实现特定 Action，无需修改框架核心代码

智能体运行机制

智能体的核心运行遵循 Observe-Think-Act 循环，这是 MetaGPT 实现自主决策的关键机制。

Observe-Think-Act 循环

每个智能体在运行时不断执行三个阶段：

1. Observe（观察）：从消息缓冲区获取新消息，过滤出感兴趣的消息，存入记忆
2. Think（思考）：根据当前状态和历史消息，决定下一步执行哪个 Action
3. Act（行动）：执行选定的 Action，生成结果消息并发布到环境

正在加载图表渲染器...

数据流要点：

1. 消息缓冲：每个 Role 维护独立的 msg_buffer，避免消息丢失和并发冲突（metagpt/roles/role.py:100-102）
2. 记忆过滤：_observe 方法通过 watch 集合和 send_to 字段过滤消息，只处理与当前角色相关的消息（metagpt/roles/role.py:399-427）
3. 状态转换：_think 方法根据 LLM 返回值或预设规则更新 rc.state，决定下一个执行的 Action（metagpt/roles/role.py:340-379）
4. 消息生成：_act 方法执行 Action 后，将结果包装为 AIMessage 并添加到记忆中（metagpt/roles/role.py:381-397）

反应模式

MetaGPT 支持三种智能体反应模式，通过 RoleReactMode 枚举定义（metagpt/roles/role.py:82-89）：

BY_ORDER 模式实现细节：

python
1if self.rc.react_mode == RoleReactMode.BY_ORDER:
2    if self.rc.max_react_loop != len(self.actions):
3        self.rc.max_react_loop = len(self.actions)
4    self._set_state(self.rc.state + 1)
5    return self.rc.state >= 0 and self.rc.state < len(self.actions)

BY_ORDER 模式通过递增 rc.state 实现顺序执行，当 state 超出 actions 列表范围时终止循环。

PLAN_AND_ACT 模式：

该模式首先调用 planner.update_plan() 生成任务计划，然后循环执行 planner.current_task 直到所有任务完成。Planner 维护独立的 working_memory 用于存储计划状态。

记忆管理

智能体的记忆系统分为两层：

1. 长期记忆：存储所有历史消息，支持按 Action 类型检索
2. 工作记忆：存储当前任务相关的临时信息，在 PLAN_AND_ACT 模式下使用

python
1class RoleContext(BaseModel):
2    memory: Memory = Field(default_factory=Memory)
3    working_memory: Memory = Field(default_factory=Memory)
4    
5    @property
6    def important_memory(self) -> list[Message]:
7        """Retrieve information corresponding to the attention action."""
8        return self.memory.get_by_actions(self.watch)

important_memory 属性返回与当前角色关注的 Action 相关的消息，用于构建 LLM 的上下文。

环境与通信机制

Environment 是多智能体协作的核心枢纽，负责管理所有角色的生命周期和消息路由。

Environment 职责

Environment 提供以下核心能力：

1. 角色管理：通过 add_role() 和 add_roles() 注册角色，维护 roles 字典和 member_addrs 映射
2. 消息路由：根据 Message 的 send_to 字段，将消息分发给对应的角色
3. 历史记录：维护 history 列表，记录所有发布的消息用于调试
4. 运行控制：通过 run() 方法驱动所有非空闲角色执行

python
1def publish_message(self, message: Message, peekable: bool = True) -> bool:
2    """Distribute the message to the recipients."""
3    logger.debug(f"publish_message: {message.dump()}")
4    found = False
5    for role, addrs in self.member_addrs.items():
6        if is_send_to(message, addrs):
7            role.put_message(message)
8            found = True
9    if not found:
10        logger.warning(f"Message no recipients: {message.dump()}")
11    self.history.add(message)
12    return True

消息路由逻辑参见 metagpt/environment/base_env.py:175-195。is_send_to() 函数检查 Message 的 send_to 字段是否匹配角色的地址集合。

Team 协调器

Team 是多智能体协作的顶层协调器，封装了 Environment 和项目级别的配置。

python
1class Team(BaseModel):
2    env: Optional[Environment] = None
3    investment: float = Field(default=10.0)
4    idea: str = Field(default="")
5    use_mgx: bool = Field(default=True)

Team 提供以下关键方法：

• hire(roles)：雇佣角色，将角色添加到 Environment
• invest(investment)：设置项目预算，通过 cost_manager.max_budget 控制成本
• run_project(idea)：启动项目，发布用户需求消息到 Environment

Team 定义参见 metagpt/team.py:32-57。

消息发布与订阅

Role 通过 publish_message() 和 put_message() 与 Environment 交互：

python
1def publish_message(self, msg):
2    """If the role belongs to env, then the role's messages will be broadcast to env"""
3    if not msg:
4        return
5    if MESSAGE_ROUTE_TO_SELF in msg.send_to:
6        msg.send_to.add(any_to_str(self))
7        msg.send_to.remove(MESSAGE_ROUTE_TO_SELF)
8    if not msg.sent_from or msg.sent_from == MESSAGE_ROUTE_TO_SELF:
9        msg.sent_from = any_to_str(self)
10    if all(to in {any_to_str(self), self.name} for to in msg.send_to):
11        self.put_message(msg)
12        return
13    if not self.rc.env:
14        return
15    if isinstance(msg, AIMessage) and not msg.agent:
16        msg.with_agent(self._setting)
17    self.rc.env.publish_message(msg)

该方法处理了多种边界情况：

1. 自路由：如果 send_to 包含 MESSAGE_ROUTE_TO_SELF，将消息发送给自己
2. 发送者填充：如果 sent_from 为空，自动填充为当前角色
3. 私有消息：如果所有接收者都是自己，直接放入私有缓冲区，不经过 Environment
4. 环境广播：否则通过 Environment 广播消息

参见 metagpt/roles/role.py:429-452。

上下文与配置管理

Context 是 MetaGPT 的运行时上下文管理中心，统一管理配置、LLM 实例和成本统计。

Context 核心结构

python
1class Context(BaseModel):
2    kwargs: AttrDict = AttrDict()
3    config: Config = Field(default_factory=Config.default)
4    cost_manager: CostManager = CostManager()
5    _llm: Optional[BaseLLM] = None

Context 提供以下能力：

1. 配置访问：通过 config 属性访问全局配置，包括 LLM 配置、项目路径等
2. LLM 实例管理：通过 llm() 方法获取或创建 LLM 实例，支持按配置类型选择不同的 CostManager
3. 成本统计：通过 cost_manager 跟踪所有 LLM 调用的成本
4. 扩展属性：通过 kwargs AttrDict 存储任意运行时数据

Context 定义参见 metagpt/context.py:58-92。

LLM 实例选择

Context 根据配置类型选择不同的 CostManager：

python
1def _select_costmanager(self, llm_config: LLMConfig) -> CostManager:
2    if llm_config.api_type == LLMType.FIREWORKS:
3        return FireworksCostManager()
4    elif llm_config.api_type == LLMType.OPEN_LLM:
5        return TokenCostManager()
6    else:
7        return self.cost_manager

这确保了不同 LLM 提供商的成本计算逻辑可以定制化。

序列化与反序列化

MetaGPT 提供了完整的序列化支持，允许保存和恢复智能体状态。

python
1class SerializationMixin(BaseSerialization):
2    def serialize(self, file_path: str = None) -> str:
3        file_path = file_path or self.get_serialization_path()
4        serialized_data = self.model_dump()
5        write_json_file(file_path, serialized_data, use_fallback=True)
6        return file_path
7
8    @classmethod
9    def deserialize(cls, file_path: str = None) -> BaseModel:
10        file_path = file_path or cls.get_serialization_path()
11        data: dict = read_json_file(file_path)
12        model = cls(**data)
13        return model

SerializationMixin 定义参见 metagpt/schema.py:72-93。Team 类在此基础上提供了项目级别的序列化：

python
1def serialize(self, stg_path: Path = None):
2    stg_path = SERDESER_PATH.joinpath("team") if stg_path is None else stg_path
3    team_info_path = stg_path.joinpath("team.json")
4    serialized_data = self.model_dump()
5    serialized_data["context"] = self.env.context.serialize()
6    write_json_file(team_info_path, serialized_data)

Team 序列化实现参见 metagpt/team.py:59-81。

模块依赖关系

正在加载图表渲染器...

依赖关系说明：

1. Team 依赖 Environment 和 Context：Team 作为顶层协调器，需要 Environment 管理角色，需要 Context 管理配置和成本
2. Environment 依赖 Role 和 Schema：Environment 管理角色生命周期，处理 Message 路由
3. Role 依赖 Action、Context、Memory 和 Schema：Role 执行 Action，维护记忆，访问上下文
4. Action 依赖 Context 和 LLM：Action 需要调用 LLM，访问配置
5. Context 依赖 Config 和 LLM：Context 管理 LLM 实例和全局配置

核心设计决策与取舍

1. 消息驱动架构 vs 直接方法调用

决策：采用消息驱动架构，智能体间通过 Environment 传递消息，不直接调用彼此的方法。

理由：

• 支持异步执行和分布式部署
• 降低智能体间的耦合度
• 便于实现消息过滤和路由
• 支持消息历史记录和调试

代价：

• 增加了消息序列化/反序列化的开销
• 调试链路更长，需要跟踪消息流转

2. Pydantic BaseModel 作为基类

决策：所有核心类（Role、Action、Message）继承自 Pydantic BaseModel。

理由：

• 自动提供数据验证和类型检查
• 原生支持 JSON 序列化/反序列化
• 支持 Field 级别的配置（如 exclude=True 排除敏感字段）
• 便于实现配置热更新

代价：

• Pydantic 的验证逻辑在频繁创建对象时有性能开销
• 部分动态属性需要通过 model_config 配置 extra="allow"

3. 多层记忆系统

决策：区分 memory（长期记忆）和 working_memory（工作记忆）。

理由：

• 长期记忆保留完整历史，支持回溯和审计
• 工作记忆只保留当前任务相关信息，减少 LLM 上下文长度
• 在 PLAN_AND_ACT 模式下，工作记忆用于存储计划状态

代价：

• 增加了记忆管理的复杂度
• 需要开发者明确区分何时使用哪种记忆

4. 三种反应模式的设计

决策：支持 REACT、BY_ORDER、PLAN_AND_ACT 三种反应模式。

理由：

• REACT 适合需要动态决策的复杂任务
• BY_ORDER 适合固定流程的流水线任务
• PLAN_AND_ACT 适合需要分解的大型项目

代价：

• 增加了 _think 方法的复杂度
• 需要开发者根据任务特性选择合适的模式

5. Context 单例共享

决策：所有智能体共享同一个 Context 实例。

理由：

• 统一管理 LLM 配置和成本统计
• 支持全局状态的共享（如项目路径、配置）
• 简化了智能体间的协调

代价：

• 可能导致状态污染，需要谨慎管理
• 在多项目并行场景下需要隔离 Context

6. ActionNode 结构化输出

决策：Action 可选包含 ActionNode，用于结构化 LLM 输出。

理由：

• 提供类型安全的输出格式
• 支持输出验证和后处理
• 便于下游 Action 消费结构化数据

代价：

• 增加了 Action 的定义复杂度
• 需要维护 ActionNode 的 schema 定义

技术选型

关键配置与启动流程

配置加载

MetaGPT 使用 config2.yaml 作为主配置文件，支持以下配置项：

yaml
1llm:
2  api_type: openai
3  model: gpt-4
4  base_url: https://api.openai.com/v1
5  api_key: sk-xxx
6
7project:
8  name: MyProject
9  path: ./workspace
10
11prompt:
12  schema: json  # json 或 markdown

Config 类通过 Config.default() 加载配置，支持环境变量覆盖。

启动流程

典型的 MetaGPT 项目启动流程：

python
1# 1. 创建 Context
2context = Context()
3
4# 2. 创建 Team
5team = Team(context=context)
6
7# 3. 雇佣角色
8team.hire([ProductManager(), Architect(), Engineer()])
9
10# 4. 设置预算
11team.invest(investment=10.0)
12
13# 5. 启动项目
14await team.run_project(idea="Build a todo list app")

流程要点：

1. Context 初始化：加载配置，创建 LLM 实例，初始化 CostManager
2. Team 创建：初始化 Environment（默认使用 MGXEnv）
3. 角色注册：将角色添加到 Environment，设置角色的 context 和 env 引用
4. 预算设置：通过 cost_manager.max_budget 限制总成本
5. 项目启动：发布用户需求消息，触发智能体协作

错误处理与边界条件

1. 预算超限：当 total_cost >= max_budget 时，抛出 NoMoneyException
2. 消息无接收者：当消息的 send_to 不匹配任何角色时，记录警告日志
3. 状态越界：当 _think 返回无效状态时，重置为 -1（终止状态）
4. LLM 调用失败：通过 tenacity 实现自动重试，支持指数退避

python
1# 预算检查实现
2def _check_balance(self):
3    if self.cost_manager.total_cost >= self.cost_manager.max_budget:
4        raise NoMoneyException(self.cost_manager.total_cost, 
5                               f"Insufficient funds: {self.cost_manager.max_budget}")

总结

MetaGPT 的架构设计体现了多智能体系统的核心原则：解耦、消息驱动、状态管理。通过 Role-Action-Environment 的三层抽象，框架实现了智能体行为的灵活组合和协作流程的可配置化。Observe-Think-Act 循环提供了统一的智能体运行模型，而三种反应模式（REACT、BY_ORDER、PLAN_AND_ACT）则覆盖了从简单到复杂的任务场景。

Context 的引入统一了配置管理和成本统计，SerializationMixin 提供了状态持久化能力，这些设计使得 MetaGPT 既能支持快速原型开发，也能满足生产环境的稳定性和可维护性要求。