CAMEL 多智能体通信指南使用 MessageAgentTool 与 AgentCommunicationToolkit 实现 Agent 间消息协作【免费下载链接】camel CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel导读本文围绕 CAMEL 仓库中MessageAgentTool当前实现为AgentCommunicationToolkit展开讲解如何让多个ChatAgent通过注册表与消息收发工具进行直接对话、广播通知与历史追溯从而在多智能体系统中构建轻量级的点对点协作通道。读完本文你将掌握该 Toolkit 的完整 API、与 Workforce 体系的集成方式以及源码级的设计细节与测试验证方法。一、背景多智能体系统中的直接通信需求在 CAMEL 的多智能体编排场景中任务通常由多个角色分工完成研究员负责检索、写手负责成稿、审稿人负责反馈。经典的做法是通过一个中心化的调度器如 Workforce分发任务但很多时候智能体之间需要即时的、点对点的直接沟通——例如写手把草稿发给审稿人等待反馈而不是每次都绕道调度层。MessageAgentTool正是为这种场景设计的一个面向智能体间通信的工具包toolkit允许 Agent 通过对方的 ID 向其发送消息并接收回复从而实现多智能体协作。它由 CAMEL 的 BaseToolkit 派生天然继承了统一的超时控制、get_tools()约定与 MCP Server 运行能力。二、MessageAgentTool 类与构造参数依据参考文档MessageAgentTool的定义如下class MessageAgentTool(BaseToolkit):构造函数的两个核心参数参数类型默认值说明agentsOptional[Dict[str, ChatAgent]]None以「Agent ID → ChatAgent 实例」为映射的初始注册表可在构造时预置一批已就绪的 AgenttimeoutOptional[float]None所有工具操作允许的最大执行时间秒None表示不设超时timeout参数的语义与 base.py 中BaseToolkit的超时机制一致构造时若传入非正数会抛出ValueError(Timeout must be a positive number.)而每个暴露给 LLM 的 callable 方法都会在子类初始化阶段被自动包装上with_timeout装饰器方法签名中自带timeout参数或被manual_timeout标记的方法除外。这意味着timeout会实际作用于message_agent等每一次工具调用防止某个 Agent 长时间无响应导致整个协作流程卡死。三、核心方法逐一解析3.1 register_agent注册可通信的 Agentdef register_agent(self, agent_id: str, agent: ChatAgent):将新的 Agent 登记进通信注册表agent_idstrAgent 的唯一标识符agentChatAgent要注册的ChatAgent实例。返回str 类型的确认消息。源码实现中若该 ID 已存在会记录一条 warning 并直接覆盖同时会为新人初始化独立的消息历史列表返回信息形如Agent {agent_id} registered successfully. Total agents: {n}。3.2 message_agent向其他 Agent 发消息并取回响应def message_agent(self, message: str, receiver_agent_id: str):这是整个 Toolkit 的主通道messagestr要发送给目标 Agent 的正文内容receiver_agent_idstr目标 Agent 的唯一标识。返回str即目标 Agent 的响应文本若操作失败则返回错误信息字符串。在消息被送达前_find_agent_id会执行宽松匹配先尝试精确匹配再尝试大小写不敏感匹配最后还会剥离_数字、_agent后缀/前缀做部分匹配例如调用方写writer_01也能命中注册的writer这大幅降低了 LLM 调用工具时拼错 ID 的失败率。消息送达时会为接收方构造一段结构化的上下文标注发送方 ID 与消息内容分隔线随后调用target_agent.step(...)驱动其推理因此接收方收到的不只是字符串而是一次完整的ChatAgent决策。3.3 list_available_agents查看在线 Agent 清单def list_available_agents(self):返回格式化字符串列出所有已注册 Agent 的 ID 与状态若注册表为空则返回No agents registered。源码实现还会附带每个 Agent 当前的历史消息数量并通过hasattr(agent, pause_event)探测其是否兼容 Workforce 体系兼容者会标注Workforce-compatible。3.4 remove_agent移除 Agentdef remove_agent(self, agent_id: str):从通信注册表中移除指定 Agentagent_idstr为要移除的 Agent 唯一标识。返回确认消息若 Agent 不存在则返回错误消息。注意源码实现中移除 Agent 时会保留其历史消息用于审计并返回Agent {agent_id} removed successfully. Remaining agents: {n}。3.5 get_agent_count查询 Agent 总数def get_agent_count(self):返回str当前注册表中的 Agent 数量。这在需要快速校验协作规模或做动态扩容判断时非常有用。3.6 get_tools导出为 LLM 可调用的 FunctionTool 列表def get_tools(self):返回List[FunctionTool]。这是 CAMEL Toolkit 的统一契约——ChatAgent通过agent.add_tool(tool)把返回的工具注入自身后LLM 便可在对话中自主调用message_agent等函数。需要特别说明的是参考文档对应版本的get_tools()只暴露安全且对 LLM 有用的函数register_agent与remove_agent这类管理函数只应被程序化调用不会暴露给 LLM避免模型随意增删协作成员。四、源码演进从 MessageAgentTool 到 AgentCommunicationToolkit当前仓库 camel/toolkits/message_agent_toolkit.py 中的实际实现类名为AgentCommunicationToolkit在 camel/toolkits/init.py 中被导出并在AgentMessage消息载体上做了进一步增强。参考文档中的MessageAgentTool是其早期 API 形态两者功能一脉相承当前实现新增了如下能力4.1 构造函数扩展def __init__( self, agents: Optional[Dict[str, ChatAgent]] None, timeout: Optional[float] None, max_message_history: int 100, get_response: bool False, ) - None:在原有agents、timeout基础上新增参数默认值说明max_message_history100每个 Agent 保留的最大消息条数超出后自动裁剪最旧消息get_responseFalse是否默认要求接收方返回响应。为False时send_message只投递通知为True时投递后立刻取回对方的回复内容AgentMessage继承自BaseMessage除标准消息字段外还携带message_id、sender_id、receiver_id、timestamp、reply_to、metadata六个专用于智能体间通信的属性每条消息在发送时通过uuid.uuid4()生成唯一 ID。4.2 比参考文档更丰富的工具面get_tools()现在返回 5 个FunctionTool对应源码 get_toolssend_message对应旧版message_agent支持reply_to建立对话线索thread支持metadata_json附加任意 JSON 元数据broadcast_message向除发送者及exclude_agents指定的 ID外的所有 Agent 群发公告每个接收方会收到metadata_json{broadcast: true}标记的消息list_available_agents同旧版展示可通信 Agent 清单get_message_history按 Agent ID 拉取历史往来消息limit控制最近条数时间戳格式化为%Y-%m-%d %H:%M:%S每条消息截断为 100 字符展示get_conversation_thread基于任意消息 ID 取出整个会话线程并按时间排序用于复盘一段往返对话的完整脉络。消息历史采用发送方 接收方双份记录的策略_add_to_history并在超过max_message_history时按保留最近 N 条裁剪避免无限增长。测试用例 test/toolkits/test_message_agent_toolkit.py 验证了该行为将上限设为 3 后连发 5 条消息历史最终只保留 3 条。五、实战将通信工具接入 Workforce 协作流仓库提供了完整可运行的示例 examples/toolkits/message_agent_toolkit.py展示了研究员 → 写手 → 审稿人三角色通过消息协作完成一篇文章的流程核心步骤可归纳为from camel.agents import ChatAgent from camel.messages import BaseMessage from camel.societies.workforce import Workforce from camel.tasks import Task from camel.toolkits import AgentCommunicationToolkit # 1. 初始化通信工具包 msg_toolkit AgentCommunicationToolkit(max_message_history100) # 2. 创建角色 Agent并在系统提示词中明确要求其使用消息工具 researcher ChatAgent( BaseMessage.make_assistant_message( role_nameResearcher, content( You are a Research Specialist. ALWAYS use send_message to share your findings with the writer. First use list_available_agents to see who you can message, then send your research findings to the writer. ), ) ) # 3. 注册 AgentID 与角色名保持一致 msg_toolkit.register_agent(Researcher, researcher) # 4. 把 5 个通信工具注入每个 Agent for tool in msg_toolkit.get_tools(): researcher.add_tool(tool) # 5. 组装 Workforce 并派发需要协作的任务 workforce Workforce(Content Creation Team) workforce.add_single_agent_worker(Researcher, researcher) task Task( content( IMPORTANT: You MUST use the messaging tools for coordination! Research and write a brief article about the benefits of AI collaboration. Researcher: list_available_agents() then send_message() to share findings with writer... ), idcollaborative_article_task, ) result await workforce.process_task_async(task)示例运行后还会调用get_message_history打印各角色的往来记录并用get_toolkit_status()输出注册 Agent 数、总消息数与每 Agent 历史上限等运行指标——这套工具注入 系统提示词引导 事后审计的组合是 CAMEL 官方推荐的 Agent 间通信落地模式。更复杂的完整版集成可见 examples/workforce/eigent.py 与 examples/debug/eigent.py。六、错误处理与边界行为源码与测试共同明确了以下行为契约空消息拒绝send_message/message_agent对空白内容直接抛出ValueError(Message content cannot be empty)对应测试 test_send_message_empty_content目标不存在接收方 ID 无法匹配时返回Error: Agent ... not found. Available agents: ...错误字符串同时附带当前可用 Agent 列表帮助 LLM 自我纠正test_send_message_agent_not_found非法 JSON 元数据metadata_json无法解析或非对象时自动降级为空字典并记录 warning不阻断消息投递投递异常兜底接收方step()抛异常时返回Error: Failed to deliver message ...保证工具层不崩溃。七、总结与适用边界MessageAgentTool/AgentCommunicationToolkit为 CAMEL 多智能体系统提供了一条去中心化的点对点消息通道注册表管理成员、宽松 ID 匹配降低 LLM 调用失误、双份历史与会话线程支持事后审计、get_response开关在通知与请求-响应两种模式间灵活切换并能与 Workforce 体系无缝共存。需要注意的适用边界该 Toolkit 的通信是同步驱动的send_message内部直接调用接收方的step()适合成员规模可控、即时性要求高的协作场景若需要异步消息队列、持久化存储或跨进程通信则属于 CAMEL 消息/存储体系中其他组件的职责范畴。相关参考资料见 docs/key_modules/tools.md 与 docs/mintlify/reference/camel.toolkits.message_agent_toolkit.mdx。【免费下载链接】camel CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 CAMEL 多智能体通信指南：使用 MessageAgentTool 与 AgentCommunicationToolkit 实现 Agent 间消息协作
CAMEL 多智能体通信指南使用 MessageAgentTool 与 AgentCommunicationToolkit 实现 Agent 间消息协作【免费下载链接】camel CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel导读本文围绕 CAMEL 仓库中MessageAgentTool当前实现为AgentCommunicationToolkit展开讲解如何让多个ChatAgent通过注册表与消息收发工具进行直接对话、广播通知与历史追溯从而在多智能体系统中构建轻量级的点对点协作通道。读完本文你将掌握该 Toolkit 的完整 API、与 Workforce 体系的集成方式以及源码级的设计细节与测试验证方法。一、背景多智能体系统中的直接通信需求在 CAMEL 的多智能体编排场景中任务通常由多个角色分工完成研究员负责检索、写手负责成稿、审稿人负责反馈。经典的做法是通过一个中心化的调度器如 Workforce分发任务但很多时候智能体之间需要即时的、点对点的直接沟通——例如写手把草稿发给审稿人等待反馈而不是每次都绕道调度层。MessageAgentTool正是为这种场景设计的一个面向智能体间通信的工具包toolkit允许 Agent 通过对方的 ID 向其发送消息并接收回复从而实现多智能体协作。它由 CAMEL 的 BaseToolkit 派生天然继承了统一的超时控制、get_tools()约定与 MCP Server 运行能力。二、MessageAgentTool 类与构造参数依据参考文档MessageAgentTool的定义如下class MessageAgentTool(BaseToolkit):构造函数的两个核心参数参数类型默认值说明agentsOptional[Dict[str, ChatAgent]]None以「Agent ID → ChatAgent 实例」为映射的初始注册表可在构造时预置一批已就绪的 AgenttimeoutOptional[float]None所有工具操作允许的最大执行时间秒None表示不设超时timeout参数的语义与 base.py 中BaseToolkit的超时机制一致构造时若传入非正数会抛出ValueError(Timeout must be a positive number.)而每个暴露给 LLM 的 callable 方法都会在子类初始化阶段被自动包装上with_timeout装饰器方法签名中自带timeout参数或被manual_timeout标记的方法除外。这意味着timeout会实际作用于message_agent等每一次工具调用防止某个 Agent 长时间无响应导致整个协作流程卡死。三、核心方法逐一解析3.1 register_agent注册可通信的 Agentdef register_agent(self, agent_id: str, agent: ChatAgent):将新的 Agent 登记进通信注册表agent_idstrAgent 的唯一标识符agentChatAgent要注册的ChatAgent实例。返回str 类型的确认消息。源码实现中若该 ID 已存在会记录一条 warning 并直接覆盖同时会为新人初始化独立的消息历史列表返回信息形如Agent {agent_id} registered successfully. Total agents: {n}。3.2 message_agent向其他 Agent 发消息并取回响应def message_agent(self, message: str, receiver_agent_id: str):这是整个 Toolkit 的主通道messagestr要发送给目标 Agent 的正文内容receiver_agent_idstr目标 Agent 的唯一标识。返回str即目标 Agent 的响应文本若操作失败则返回错误信息字符串。在消息被送达前_find_agent_id会执行宽松匹配先尝试精确匹配再尝试大小写不敏感匹配最后还会剥离_数字、_agent后缀/前缀做部分匹配例如调用方写writer_01也能命中注册的writer这大幅降低了 LLM 调用工具时拼错 ID 的失败率。消息送达时会为接收方构造一段结构化的上下文标注发送方 ID 与消息内容分隔线随后调用target_agent.step(...)驱动其推理因此接收方收到的不只是字符串而是一次完整的ChatAgent决策。3.3 list_available_agents查看在线 Agent 清单def list_available_agents(self):返回格式化字符串列出所有已注册 Agent 的 ID 与状态若注册表为空则返回No agents registered。源码实现还会附带每个 Agent 当前的历史消息数量并通过hasattr(agent, pause_event)探测其是否兼容 Workforce 体系兼容者会标注Workforce-compatible。3.4 remove_agent移除 Agentdef remove_agent(self, agent_id: str):从通信注册表中移除指定 Agentagent_idstr为要移除的 Agent 唯一标识。返回确认消息若 Agent 不存在则返回错误消息。注意源码实现中移除 Agent 时会保留其历史消息用于审计并返回Agent {agent_id} removed successfully. Remaining agents: {n}。3.5 get_agent_count查询 Agent 总数def get_agent_count(self):返回str当前注册表中的 Agent 数量。这在需要快速校验协作规模或做动态扩容判断时非常有用。3.6 get_tools导出为 LLM 可调用的 FunctionTool 列表def get_tools(self):返回List[FunctionTool]。这是 CAMEL Toolkit 的统一契约——ChatAgent通过agent.add_tool(tool)把返回的工具注入自身后LLM 便可在对话中自主调用message_agent等函数。需要特别说明的是参考文档对应版本的get_tools()只暴露安全且对 LLM 有用的函数register_agent与remove_agent这类管理函数只应被程序化调用不会暴露给 LLM避免模型随意增删协作成员。四、源码演进从 MessageAgentTool 到 AgentCommunicationToolkit当前仓库 camel/toolkits/message_agent_toolkit.py 中的实际实现类名为AgentCommunicationToolkit在 camel/toolkits/init.py 中被导出并在AgentMessage消息载体上做了进一步增强。参考文档中的MessageAgentTool是其早期 API 形态两者功能一脉相承当前实现新增了如下能力4.1 构造函数扩展def __init__( self, agents: Optional[Dict[str, ChatAgent]] None, timeout: Optional[float] None, max_message_history: int 100, get_response: bool False, ) - None:在原有agents、timeout基础上新增参数默认值说明max_message_history100每个 Agent 保留的最大消息条数超出后自动裁剪最旧消息get_responseFalse是否默认要求接收方返回响应。为False时send_message只投递通知为True时投递后立刻取回对方的回复内容AgentMessage继承自BaseMessage除标准消息字段外还携带message_id、sender_id、receiver_id、timestamp、reply_to、metadata六个专用于智能体间通信的属性每条消息在发送时通过uuid.uuid4()生成唯一 ID。4.2 比参考文档更丰富的工具面get_tools()现在返回 5 个FunctionTool对应源码 get_toolssend_message对应旧版message_agent支持reply_to建立对话线索thread支持metadata_json附加任意 JSON 元数据broadcast_message向除发送者及exclude_agents指定的 ID外的所有 Agent 群发公告每个接收方会收到metadata_json{broadcast: true}标记的消息list_available_agents同旧版展示可通信 Agent 清单get_message_history按 Agent ID 拉取历史往来消息limit控制最近条数时间戳格式化为%Y-%m-%d %H:%M:%S每条消息截断为 100 字符展示get_conversation_thread基于任意消息 ID 取出整个会话线程并按时间排序用于复盘一段往返对话的完整脉络。消息历史采用发送方 接收方双份记录的策略_add_to_history并在超过max_message_history时按保留最近 N 条裁剪避免无限增长。测试用例 test/toolkits/test_message_agent_toolkit.py 验证了该行为将上限设为 3 后连发 5 条消息历史最终只保留 3 条。五、实战将通信工具接入 Workforce 协作流仓库提供了完整可运行的示例 examples/toolkits/message_agent_toolkit.py展示了研究员 → 写手 → 审稿人三角色通过消息协作完成一篇文章的流程核心步骤可归纳为from camel.agents import ChatAgent from camel.messages import BaseMessage from camel.societies.workforce import Workforce from camel.tasks import Task from camel.toolkits import AgentCommunicationToolkit # 1. 初始化通信工具包 msg_toolkit AgentCommunicationToolkit(max_message_history100) # 2. 创建角色 Agent并在系统提示词中明确要求其使用消息工具 researcher ChatAgent( BaseMessage.make_assistant_message( role_nameResearcher, content( You are a Research Specialist. ALWAYS use send_message to share your findings with the writer. First use list_available_agents to see who you can message, then send your research findings to the writer. ), ) ) # 3. 注册 AgentID 与角色名保持一致 msg_toolkit.register_agent(Researcher, researcher) # 4. 把 5 个通信工具注入每个 Agent for tool in msg_toolkit.get_tools(): researcher.add_tool(tool) # 5. 组装 Workforce 并派发需要协作的任务 workforce Workforce(Content Creation Team) workforce.add_single_agent_worker(Researcher, researcher) task Task( content( IMPORTANT: You MUST use the messaging tools for coordination! Research and write a brief article about the benefits of AI collaboration. Researcher: list_available_agents() then send_message() to share findings with writer... ), idcollaborative_article_task, ) result await workforce.process_task_async(task)示例运行后还会调用get_message_history打印各角色的往来记录并用get_toolkit_status()输出注册 Agent 数、总消息数与每 Agent 历史上限等运行指标——这套工具注入 系统提示词引导 事后审计的组合是 CAMEL 官方推荐的 Agent 间通信落地模式。更复杂的完整版集成可见 examples/workforce/eigent.py 与 examples/debug/eigent.py。六、错误处理与边界行为源码与测试共同明确了以下行为契约空消息拒绝send_message/message_agent对空白内容直接抛出ValueError(Message content cannot be empty)对应测试 test_send_message_empty_content目标不存在接收方 ID 无法匹配时返回Error: Agent ... not found. Available agents: ...错误字符串同时附带当前可用 Agent 列表帮助 LLM 自我纠正test_send_message_agent_not_found非法 JSON 元数据metadata_json无法解析或非对象时自动降级为空字典并记录 warning不阻断消息投递投递异常兜底接收方step()抛异常时返回Error: Failed to deliver message ...保证工具层不崩溃。七、总结与适用边界MessageAgentTool/AgentCommunicationToolkit为 CAMEL 多智能体系统提供了一条去中心化的点对点消息通道注册表管理成员、宽松 ID 匹配降低 LLM 调用失误、双份历史与会话线程支持事后审计、get_response开关在通知与请求-响应两种模式间灵活切换并能与 Workforce 体系无缝共存。需要注意的适用边界该 Toolkit 的通信是同步驱动的send_message内部直接调用接收方的step()适合成员规模可控、即时性要求高的协作场景若需要异步消息队列、持久化存储或跨进程通信则属于 CAMEL 消息/存储体系中其他组件的职责范畴。相关参考资料见 docs/key_modules/tools.md 与 docs/mintlify/reference/camel.toolkits.message_agent_toolkit.mdx。【免费下载链接】camel CAMEL: The first and the best multi-agent framework. Finding the Scaling Law of Agents. https://www.camel-ai.org项目地址: https://gitcode.com/GitHub_Trending/ca/camel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考