消息传递系统

相关源文件

• metagpt/actions/action.py
• metagpt/actions/design_api.py
• metagpt/actions/project_management.py
• metagpt/actions/summarize_code.py
• metagpt/actions/write_code.py
• metagpt/actions/write_code_review.py
• metagpt/actions/write_prd.py
• metagpt/const.py
• metagpt/memory/memory.py
• metagpt/roles/architect.py
• metagpt/roles/engineer.py
• metagpt/roles/product_manager.py
• metagpt/roles/project_manager.py
• metagpt/roles/qa_engineer.py
• metagpt/roles/role.py
• metagpt/schema.py
• metagpt/team.py
• metagpt/utils/common.py

消息传递系统是使 MetaGPT 的多智能体架构成为可能的核心通信基础设施。该系统允许不同的角色（智能体）通过结构化的消息交换来交换信息、协调动作并集体解决复杂的任务。本文档解释了 MetaGPT 中的消息结构、消息路由和整体消息处理工作流程。

有关使用这些消息的角色系统信息，请参阅角色系统。

消息结构

MetaGPT 中的消息由Message类表示，该类是智能体之间通信的基本单元。

来源：metagpt/schema.py232-417

关键消息属性

属性	类型	描述
id	str	消息的唯一标识符，如果未提供，则生成为 UUID
content	str	消息的实际文本内容
instruct_content	Optional[BaseModel]	结构化数据内容（用于传递复杂对象）
role	str	消息发送者的角色（“user”、“system”或“assistant”）
cause_by	str	导致此消息的原因的引用（通常是 Action 类的名称）
sent_from	str	消息的发送者（通常是 Role 的名称）
send_to	set[str]	目标接收者的集合（角色名称或特殊常量）
metadata	Dict[str, Any]	消息的附加元数据

来源：metagpt/schema.py232-284

消息类型

MetaGPT 提供专门的消息类以简化消息创建

1. UserMessage：用户发送的消息，role="user"
2. SystemMessage：系统指令，role="system"
3. AIMessage：AI 智能体发送的消息，role="assistant" 以及额外的智能体元数据

来源：metagpt/schema.py419-455

消息队列

MessageQueue 类提供了一个异步消息缓冲区供角色发送和接收消息。

来源：metagpt/schema.py713-784

消息路由

消息在 MetaGPT 中可以使用特殊路由常量以不同的方式进行路由。

常量	值	描述
MESSAGE_ROUTE_TO_ALL	"<all>"	广播给所有角色
MESSAGE_ROUTE_TO_NONE	"<none>"	不路由给任何人（用于终止消息链）
MESSAGE_ROUTE_TO_SELF	"<self>"	路由回发送者

来源：metagpt/const.py77-83

消息流

以下序列图说明了消息如何通过环境在角色之间流动

来源：metagpt/roles/role.py523-554 metagpt/roles/role.py429-452

角色中的消息处理

MetaGPT 中的每个角色都有一个消息缓冲区，并遵循观察-思考-行动的循环来处理消息。

来源：metagpt/roles/role.py399-427 metagpt/roles/role.py454-470 metagpt/roles/role.py529-553

消息观察

角色通过_observe()方法观察传入消息

1. 从消息缓冲区获取未处理的消息
2. 根据角色的兴趣（关注列表）过滤消息
3. 将相关消息存储在角色的内存中
4. 更新latest_observed_msg用于恢复目的

来源：metagpt/roles/role.py399-427

消息创建和发布

角色通过以下方式创建和发送消息

1. publish_message(msg)：通过环境将消息发送给其他角色

   • 如果MESSAGE_ROUTE_TO_SELF在send_to中，则路由到自身
   • 为AIMessage对象添加智能体信息
   • 将消息传递给环境进行分发

2. put_message(message)：将消息直接放入角色的私有缓冲区

来源：metagpt/roles/role.py429-452

消息寻址和路由

MetaGPT 使用基于订阅的寻址系统

来源：metagpt/roles/role.py429-452 metagpt/utils/common.py423-431

角色寻址

角色可以指定两种类型的地址

1. 已发布的地址（role.addresses）：角色监听的地址集合

   • 默认情况下，包括角色的类名和角色名
   • 可以添加额外的地址

2. 消息过滤（role.rc.watch）：角色感兴趣的消息类型集合（基于cause_by）

来源：metagpt/roles/role.py210-214 metagpt/roles/role.py284-291

消息序列化和持久化

消息支持序列化以进行存储和恢复

1. message.dump()：将消息转换为 JSON 字符串
2. Message.load(val)：从 JSON 字符串创建消息
3. MessageQueue.dump()：将队列中的所有消息序列化为 JSON
4. MessageQueue.load(data)：从序列化数据创建消息队列

这使得在中断后能够恢复状态并持久化对话历史。

来源：metagpt/schema.py336-356 metagpt/schema.py748-782

消息使用示例

创建和发布消息

来源：metagpt/roles/engineer.py181-188

使用 instruct_content 处理结构化数据

来源：metagpt/actions/write_prd.py180-188

与 Action 系统的集成

消息传递系统与 Action 系统紧密集成

当一个 Action 完成时，它通常会产生一个结果，该结果被转换为消息并通过 Role 发布

1. Action 执行并返回结果
2. 结果被转换为带有 cause_by 设置为 Action 的消息
3. 消息被发布到环境
4. 环境将消息路由给合适角色的
5. 接收角色观察并处理消息

来源：metagpt/roles/role.py380-396 metagpt/roles/role.py529-553

内存和消息存储

角色会在其内存中存储观察到的消息，这有助于

1. 维护对话历史
2. 从中断中恢复
3. 为未来的行动提供上下文

Memory 类提供根据各种标准检索消息的方法，包括

• get(): 获取所有或最近的消息
• get_by_role(): 获取特定角色的消息
• get_by_action(): 获取由特定动作引起的消息

来源: metagpt/memory/memory.py20-111

异常处理与恢复

消息系统包含处理异常和从故障中恢复的机制

1. role_raise_decorator 会捕获角色执行期间的异常
2. 如果发生异常，将从内存中移除最近观察到的消息
3. 这使得角色在恢复后可以再次观察到该消息
4. recovered 标志被设置为指示恢复状态

来源: metagpt/utils/common.py689-716 metagpt/roles/role.py156-181

实际应用示例

团队协作

消息传递系统支持复杂的团队协作，例如软件开发

这种交互完全通过消息传递系统实现，每个角色都基于其接收到的消息进行观察、思考和行动。

来源: metagpt/team.py123-138

总结

消息传递系统是 MetaGPT 的通信骨干，它支持

1. 结构化通信：具有路由信息的定义明确的消息格式
2. 异步处理：基于队列的消息处理，用于非阻塞操作
3. 灵活寻址：多种寻址机制（广播、直接、自身）
4. 序列化：支持持久化和恢复
5. 集成：与角色和动作系统无缝集成

理解该系统对于扩展 MetaGPT 以实现能够参与协作解决问题过程的新角色、动作和功能至关重要。