在实际 AI 应用开发中构建一个能够独立完成复杂任务的智能体Agent已经不再是难题。然而当任务链条变长、需要多步骤协作或涉及不同专业领域时单个智能体往往力不从心。这时一个自然的想法是能否让多个智能体像团队成员一样围绕一个共同目标进行对话、分工与协作这正是多智能体协作Multi-Agent Collaboration领域的核心课题。AgentCouch 这一概念形象地描绘了让智能体们“坐在沙发上聊天”的场景其背后指向的是构建一个支持智能体间高效、结构化通信与协作的框架或平台。这种协作并非简单的消息转发而是需要解决智能体如何发现彼此、如何理解对方的能力、如何协商任务、如何传递复杂的上下文如代码、设计稿、数据等一系列工程挑战。对于希望将 AI 能力从单点工具升级为自动化工作流的开发者而言理解并实践多智能体协作是必经之路。本文将围绕如何构建一个支持智能体间对话的协作系统展开。我们将从核心协议与概念入手逐步搭建一个最小化的多智能体通信骨架并深入探讨智能体能力描述、任务编排、上下文管理等关键问题。无论你是希望集成现有 AI 服务还是从零开始设计专属的智能体团队本文提供的思路和代码示例都将为你提供一个坚实的起点。1. 理解多智能体协作的核心协议与通信模型在让智能体们“聊天”之前必须先为它们定义一套共同的语言和交互规则。这类似于人类团队中的沟通协议和会议流程。1.1 为什么需要专门的通信协议单个智能体通常通过自然语言与用户交互其内部状态和决策过程对外是不可见的。当多个智能体协作时如果仅靠转发用户的自然语言指令会导致严重的信息损失和歧义。例如智能体A生成了一段代码智能体B需要审查它。如果只传递代码文本B可能不知道这段代码的用途、所属模块、或需要满足的测试用例。因此协作协议需要能封装丰富的结构化信息。目前业界逐渐形成了一些共识和雏形标准例如MCPModel Context Protocol的概念。虽然 MCP 的具体定义可能因项目而异但其核心思想是提供一种标准化的方式让不同的“模型”或智能体能够访问和操作共享的“上下文”Context。这个上下文可以包括文档、代码库、数据库 schema、API 文档、会话历史等。在多智能体场景中MCP 或其类似协议可以演化为智能体间交换上下文信息的载体。1.2 智能体协作的基本通信模式智能体间的对话通常不是随意的闲聊而是围绕任务的、有结构的交互。主要模式包括请求-响应模式这是最基础的同步模式。智能体 A 向智能体 B 发送一个明确的请求如“请分析这段日志”B 处理并返回结果。这类似于 HTTP 或 RPC 调用。发布-订阅模式适用于事件驱动的场景。当某个智能体完成一项工作如“数据库备份完成”或检测到一个状态如“系统负载过高”时它会向一个消息通道发布事件。其他关心此事件的智能体订阅该通道并作出响应。广播与协商模式当一个任务需要多个智能体共同决策时发起者可以向所有相关智能体广播任务信息收集大家的“意见”能力声明、预估耗时、所需资源等然后进行协商或指派。链式或工作流模式智能体按照预定义的流程依次执行上一个的输出是下一个的输入。这需要工作流引擎来编排顺序和处理分支。在实现层面这些模式可以通过消息队列如 RabbitMQ、Kafka、WebSocket、或简单的 HTTP 服务来实现。选择哪种模式取决于智能体协作的实时性、耦合度和复杂度要求。1.3 定义智能体的“技能”与“身份”为了让智能体能有效地找到协作者每个智能体需要清晰地声明自己的能力。这通常通过一个“技能描述”文件来实现。这个描述文件应该包含智能体 ID唯一标识符。能力列表该智能体擅长处理的任务类型例如code_review、sql_generation、ui_design_critique。输入/输出格式它能接受什么格式的输入如 JSON Schema、文本、特定类型的文件以及它返回数据的格式。调用端点其他智能体如何调用它如 REST API 地址、消息队列的主题名。元数据版本、作者、所需资源等。# 示例一个代码审查智能体的技能描述 (agent_skill.yaml) agent_id: code_reviewer_v1 name: Python 代码审查专家 description: 专注于审查 Python 代码的语法、风格和潜在 bug。 capabilities: - name: review_python_code description: 审查给定的 Python 代码片段。 input_schema: type: object properties: code: type: string description: 待审查的 Python 代码 context: type: string description: 代码的上下文或需求描述可选 required: [code] output_schema: type: object properties: issues: type: array items: type: object properties: line: type: integer severity: type: string enum: [ERROR, WARNING, INFO] message: type: string suggestion: type: string summary: type: string endpoint: type: http url: http://localhost:8081/review method: POST metadata: version: 1.0.0 language: python有了这样的描述一个协调者智能体或服务注册中心就能知道系统中有哪些可用的“专家”并能根据任务类型进行匹配和路由。2. 搭建多智能体通信的基础骨架我们将从一个最简单的场景开始两个智能体通过 HTTP 进行直接的请求-响应式对话。我们将创建两个简单的 Python 服务来模拟智能体。2.1 环境准备与项目结构首先确保你的开发环境已安装 Python 3.8。我们将使用FastAPI来快速构建 HTTP 服务因为它轻量且易于定义 API。创建一个项目目录并初始化虚拟环境mkdir agent-couch-demo cd agent-couch-demo python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate pip install fastapi uvicorn requests pydantic项目结构如下agent-couch-demo/ ├── coordinator.py # 协调者服务可选用于路由 ├── agents/ │ ├── __init__.py │ ├── code_reviewer.py # 代码审查智能体 │ └── sql_expert.py # SQL 生成智能体 ├── shared/ │ ├── __init__.py │ └── models.py # 共享的数据模型消息格式 └── requirements.txt2.2 定义智能体间消息协议在shared/models.py中我们定义智能体间通信的基本消息格式。这是实现“共同语言”的关键。# shared/models.py from pydantic import BaseModel, Field from typing import Any, Dict, List, Optional from enum import Enum class MessageType(str, Enum): 消息类型枚举 TASK_REQUEST task_request # 任务请求 TASK_RESPONSE task_response # 任务响应 BROADCAST broadcast # 广播消息 ERROR error # 错误消息 class AgentMessage(BaseModel): 智能体间通信的基础消息模型 msg_id: str Field(..., description消息唯一ID) sender_id: str Field(..., description发送者智能体ID) receiver_id: Optional[str] Field(None, description接收者智能体IDNone表示广播) msg_type: MessageType Field(..., description消息类型) content: Dict[str, Any] Field(..., description消息内容结构由msg_type决定) timestamp: str Field(..., description消息发送时间戳) # 上下文信息用于追踪任务链 context_id: Optional[str] Field(None, description关联的上下文或会话ID) parent_msg_id: Optional[str] Field(None, description父消息ID用于回复链) class TaskRequestContent(BaseModel): TASK_REQUEST 类型的消息内容结构 task_type: str Field(..., description任务类型如 code_review, generate_sql) task_params: Dict[str, Any] Field(..., description任务参数) required_capabilities: Optional[List[str]] Field(None, description完成任务所需的能力列表) class TaskResponseContent(BaseModel): TASK_RESPONSE 类型的消息内容结构 task_id: str Field(..., description对应的任务请求ID) status: str Field(..., description任务状态如 success, failed, partial) result: Optional[Dict[str, Any]] Field(None, description任务执行结果) error_info: Optional[str] Field(None, description错误信息)这个模型确保了消息的规范性。AgentMessage是信封content字段内的具体结构由msg_type决定。2.3 实现第一个智能体代码审查者在agents/code_reviewer.py中我们实现一个简单的代码审查智能体。它提供一个 HTTP 端点接收代码并返回审查意见。# agents/code_reviewer.py from fastapi import FastAPI, HTTPException import uvicorn from shared.models import AgentMessage, MessageType, TaskRequestContent, TaskResponseContent from pydantic import BaseModel import uuid from datetime import datetime app FastAPI(titleCode Reviewer Agent) # 模拟一个简单的代码审查逻辑 def review_python_code(code: str, context: str ) - dict: issues [] lines code.split(\n) for i, line in enumerate(lines, start1): line line.rstrip() # 一些简单的规则检查 if len(line) 100: issues.append({ line: i, severity: WARNING, message: f行 {i} 超过 100 字符, suggestion: 考虑拆分行或简化表达式 }) if print( in line and TODO not in context.upper(): # 假设在生产代码中不鼓励直接使用print issues.append({ line: i, severity: INFO, message: f行 {i} 使用了 print 语句, suggestion: 考虑使用 logging 模块以便于控制输出级别 }) # 可以在这里添加更多检查如导入风格、命名规范等 return { issues: issues, summary: f发现 {len(issues)} 个潜在问题。, reviewed_at: datetime.utcnow().isoformat() } class ReviewRequest(BaseModel): code: str context: str app.post(/review) async def do_review(request: ReviewRequest): 对外提供的 API 端点 try: result review_python_code(request.code, request.context) return {status: success, data: result} except Exception as e: raise HTTPException(status_code500, detailf审查过程出错: {str(e)}) app.post(/agent_message) async def handle_agent_message(message: AgentMessage): 处理来自其他智能体的标准化消息 if message.msg_type ! MessageType.TASK_REQUEST: return AgentMessage( msg_idstr(uuid.uuid4()), sender_idcode_reviewer_v1, receiver_idmessage.sender_id, msg_typeMessageType.ERROR, content{error: f不支持的消息类型: {message.msg_type}}, timestampdatetime.utcnow().isoformat(), context_idmessage.context_id, parent_msg_idmessage.msg_id ) try: # 解析任务请求内容 task_content TaskRequestContent(**message.content) if task_content.task_type ! code_review: raise ValueError(f本智能体不支持的任务类型: {task_content.task_type}) code task_content.task_params.get(code) if not code: raise ValueError(任务参数中缺少 code 字段) context task_content.task_params.get(context, ) # 执行核心审查逻辑 review_result review_python_code(code, context) # 构建响应消息 response_content TaskResponseContent( task_idmessage.msg_id, statussuccess, resultreview_result ) return AgentMessage( msg_idstr(uuid.uuid4()), sender_idcode_reviewer_v1, receiver_idmessage.sender_id, msg_typeMessageType.TASK_RESPONSE, contentresponse_content.dict(), timestampdatetime.utcnow().isoformat(), context_idmessage.context_id, parent_msg_idmessage.msg_id ) except Exception as e: # 返回错误响应 error_response TaskResponseContent( task_idmessage.msg_id, statusfailed, error_infostr(e) ) return AgentMessage( msg_idstr(uuid.uuid4()), sender_idcode_reviewer_v1, receiver_idmessage.sender_id, msg_typeMessageType.TASK_RESPONSE, contenterror_response.dict(), timestampdatetime.utcnow().isoformat(), context_idmessage.context_id, parent_msg_idmessage.msg_id ) if __name__ __main__: # 启动服务在 8081 端口 uvicorn.run(app, host0.0.0.0, port8081)这个智能体提供了两个端点/review: 一个简单的 REST API供外部直接调用。/agent_message: 专用于智能体间通信的端点接收和返回标准化的AgentMessage。这体现了“协议”的重要性——智能体间使用一种更丰富、更结构化的方式交流。2.4 实现第二个智能体与协调者我们再创建一个 SQL 专家智能体 (agents/sql_expert.py)其结构与审查者类似但能力是生成 SQL。为了演示智能体间的对话我们还需要一个简单的协调者 (coordinator.py)。协调者的作用是接收用户或系统的原始任务将其分解并路由给合适的智能体。# coordinator.py from fastapi import FastAPI, HTTPException import uvicorn import requests import uuid from datetime import datetime from shared.models import AgentMessage, MessageType, TaskRequestContent from pydantic import BaseModel from typing import Dict app FastAPI(titleAgent Coordinator) # 简单的智能体注册表实际项目中可能使用服务发现如 Consul AGENT_REGISTRY { code_review: {agent_id: code_reviewer_v1, endpoint: http://localhost:8081/agent_message}, generate_sql: {agent_id: sql_expert_v1, endpoint: http://localhost:8082/agent_message}, } class UserRequest(BaseModel): task: str # 如 review_code_and_generate_sql parameters: Dict[str, any] def send_to_agent(agent_info: dict, message: AgentMessage) - AgentMessage: 向指定智能体发送消息并获取响应 try: resp requests.post(agent_info[endpoint], jsonmessage.dict(), timeout30) resp.raise_for_status() return AgentMessage(**resp.json()) except requests.exceptions.RequestException as e: # 构建一个本地的错误响应消息 error_content { task_id: message.msg_id, status: failed, error_info: f无法连接到智能体 {agent_info[agent_id]}: {str(e)} } return AgentMessage( msg_idstr(uuid.uuid4()), sender_idcoordinator, receiver_idmessage.sender_id, msg_typeMessageType.TASK_RESPONSE, contenterror_content, timestampdatetime.utcnow().isoformat(), context_idmessage.context_id, parent_msg_idmessage.msg_id ) app.post(/orchestrate) async def orchestrate_task(request: UserRequest): 协调任务这是一个简单的顺序工作流示例 context_id str(uuid.uuid4()) # 为本次用户请求创建唯一上下文ID if request.task review_code_and_generate_sql: # 步骤1将代码发送给审查者 review_agent AGENT_REGISTRY[code_review] review_msg AgentMessage( msg_idstr(uuid.uuid4()), sender_idcoordinator, receiver_idreview_agent[agent_id], msg_typeMessageType.TASK_REQUEST, contentTaskRequestContent( task_typecode_review, task_paramsrequest.parameters ).dict(), timestampdatetime.utcnow().isoformat(), context_idcontext_id ) review_response send_to_agent(review_agent, review_msg) # 步骤2根据审查结果这里简单处理再调用 SQL 专家 # 注意实际逻辑可能需要解析 review_response 的内容来决定下一步 sql_agent AGENT_REGISTRY[generate_sql] # 假设我们从用户参数中提取一个需求描述来生成 SQL sql_task_params {description: request.parameters.get(description, 生成查询用户表的SQL)} sql_msg AgentMessage( msg_idstr(uuid.uuid4()), sender_idcoordinator, receiver_idsql_agent[agent_id], msg_typeMessageType.TASK_REQUEST, contentTaskRequestContent( task_typegenerate_sql, task_paramssql_task_params ).dict(), timestampdatetime.utcnow().isoformat(), context_idcontext_id, parent_msg_idreview_response.msg_id # 关联到上一步的消息 ) sql_response send_to_agent(sql_agent, sql_msg) # 汇总结果返回给用户 return { context_id: context_id, workflow: request.task, steps: [ {agent: code_reviewer, response: review_response.dict()}, {agent: sql_expert, response: sql_response.dict()} ] } else: raise HTTPException(status_code400, detailf不支持的任务类型: {request.task}) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8080)这个协调者实现了一个最简单的顺序工作流。它展示了如何维护一个智能体注册表。将用户请求转换为标准化的智能体间消息。按顺序调用多个智能体并传递上下文。汇总结果。3. 运行与验证多智能体对话现在让我们启动这个微型的多智能体系统并验证它们是否能成功“对话”。3.1 启动所有服务打开三个终端窗口分别运行# 终端1启动代码审查智能体 cd agent-couch-demo python -m agents.code_reviewer # 服务启动在 http://localhost:8081 # 终端2启动 SQL 专家智能体假设已实现结构类似 # python -m agents.sql_expert # 服务启动在 http://localhost:8082 为了演示我们可以先跳过用模拟响应 # 终端3启动协调者 python coordinator.py # 服务启动在 http://localhost:8080为了简化我们可以先不实现完整的 SQL 专家而是修改协调者在第二步模拟一个成功的 SQL 生成响应。3.2 测试智能体间直接对话首先测试代码审查智能体是否能正确处理来自协调者的标准化消息。我们可以使用curl或 Python 的requests库来模拟协调者发送消息。# test_direct_message.py import requests import json from datetime import datetime import uuid # 构建一个标准的 AgentMessage message { msg_id: str(uuid.uuid4()), sender_id: test_coordinator, receiver_id: code_reviewer_v1, msg_type: task_request, content: { task_type: code_review, task_params: { code: def calculate_sum(a, b):\n result a b\n print(fThe sum is {result})\n return result, context: 这是一个计算两数之和的函数请审查。 } }, timestamp: datetime.utcnow().isoformat(), context_id: str(uuid.uuid4()) } url http://localhost:8081/agent_message response requests.post(url, jsonmessage) print(Status Code:, response.status_code) print(Response Body:, json.dumps(response.json(), indent2, ensure_asciiFalse))运行这个测试脚本你应该会收到一个结构化的响应其中包含审查结果。这证明了两个服务之间能够使用我们定义的协议进行通信。3.3 测试完整的工作流现在通过协调者的接口来测试整个工作流。curl -X POST http://localhost:8080/orchestrate \ -H Content-Type: application/json \ -d { task: review_code_and_generate_sql, parameters: { code: for i in range(10):\n print(i), description: 需要一个查询所有活跃用户的SQL } }协调者会依次调用代码审查智能体和 SQL 专家智能体或模拟并将两个步骤的结果汇总返回。查看响应你应该能看到一个包含context_id和两个步骤详细响应的 JSON 对象。3.4 验证关键点通过以上测试我们验证了多智能体协作系统的几个核心能力协议通信智能体间使用AgentMessage进行结构化数据交换而不是纯文本。服务发现协调者通过注册表知道每个智能体的能力和地址。任务路由协调者根据任务类型 (task_type) 将请求路由到正确的智能体。上下文关联通过context_id和parent_msg_id可以追踪一个任务链中的所有消息。错误处理消息处理过程中有基本的异常捕获和错误响应格式。4. 深入探讨从基础对话到复杂协作上面的例子是一个高度简化的模型。要让智能体们真正高效地“坐在沙发上聊天”还需要解决更多工程问题。4.1 智能体能力发现与动态注册硬编码的注册表 (AGENT_REGISTRY) 不适合生产环境。我们需要一个服务发现机制。常见的模式是注册中心每个智能体启动时向一个中心服务如 Consul、Etcd、或一个简单的注册服务注册自己的技能描述见 1.3 节。心跳与健康检查注册中心定期检查智能体是否存活将不可用的智能体从可用列表中移除。能力查询协调者或需要帮助的智能体可以向注册中心查询“谁有能力做code_review”。# 伪代码智能体启动时自动注册 def register_agent(agent_skill_yaml_path, registry_url): with open(agent_skill_yaml_path, r) as f: skill_info yaml.safe_load(f) requests.post(f{registry_url}/register, jsonskill_info) # 伪代码协调者动态查找智能体 def find_agent_for_task(task_type, registry_url): response requests.get(f{registry_url}/discover, params{capability: task_type}) agents response.json() if agents: # 可以在这里实现负载均衡或选择策略 return agents[0] else: raise Exception(fNo agent found for task: {task_type})4.2 对话管理与上下文持久化一次复杂的协作可能涉及多轮对话。例如审查者可能要求生成者提供更多上下文生成者可能需要向用户确认。这需要维护一个“会话”状态。会话服务提供一个中心化的服务来管理会话。每个会话有一个唯一 ID关联所有相关的消息、智能体状态和最终结果。消息持久化将所有AgentMessage存储到数据库如 MongoDB、PostgreSQL便于调试、审计和实现“断点续聊”。上下文注入智能体在处理消息时可以从会话服务中获取整个对话历史从而做出更连贯的决策。4.3 编排引擎与工作流定义我们的协调者只是一个简单的顺序执行器。复杂的任务可能需要并行执行、条件分支、循环、错误重试等。此时需要引入工作流引擎如Apache Airflow、Prefect、或 Temporal。你可以用 YAML 或 Python DSL 定义工作流# workflow_definition.yaml name: data_processing_and_review tasks: - id: fetch_data type: http_request agent: data_fetcher params: {url: {{input.data_url}}} - id: clean_data type: data_transform agent: data_cleaner params: {raw_data: {{tasks.fetch_data.output}}} depends_on: [fetch_data] - id: review_cleaning_code type: code_review agent: code_reviewer params: {code: {{tasks.clean_data.metadata.code_snippet}}} depends_on: [clean_data] - id: generate_report type: report_generation agent: report_generator params: data: {{tasks.clean_data.output}} review_comments: {{tasks.review_cleaning_code.output}} depends_on: [clean_data, review_cleaning_code]编排引擎负责解析这个定义创建任务实例管理依赖调用相应的智能体并处理失败和重试。4.4 共享上下文与 MCP 思想这是实现深度协作的关键。智能体 A 生成的图表、智能体 B 编写的代码、智能体 C 查询的数据如何让智能体 D 无缝使用这就是MCPModel Context Protocol类协议要解决的问题。其核心是定义一个标准的、工具可读的“上下文”格式。一个简单的共享上下文服务可以这样设计# shared_context.py from typing import Dict, Any import json class SharedContextService: def __init__(self): self._context_store {} # context_id - {resources: {...}} def put_resource(self, context_id: str, resource_type: str, resource_id: str, content: Dict[str, Any]): 将一个资源放入共享上下文 if context_id not in self._context_store: self._context_store[context_id] {resources: {}} key f{resource_type}:{resource_id} self._context_store[context_id][resources][key] { content: content, created_by: some_agent_id, created_at: timestamp } def get_resource(self, context_id: str, resource_type: str, resource_id: str): 从共享上下文获取一个资源 key f{resource_type}:{resource_id} return self._context_store.get(context_id, {}).get(resources, {}).get(key) def list_resources(self, context_id: str, resource_type: str None): 列出共享上下文中的所有资源或某类资源 resources self._context_store.get(context_id, {}).get(resources, {}) if resource_type: return {k: v for k, v in resources.items() if k.startswith(f{resource_type}:)} return resources # 智能体在生成代码后可以将其存入共享上下文 context_service.put_resource( context_idsession_123, resource_typecode, resource_iddata_cleaner_module_v1, content{language: python, code: def clean(x): ..., dependencies: [pandas]} ) # 另一个智能体可以获取并使用这段代码 code_obj context_service.get_resource(session_123, code, data_cleaner_module_v1)这样智能体间的协作就不再是孤立的请求-响应而是围绕一个不断丰富的共享上下文进行建设。5. 生产环境考量与常见问题排查将多智能体系统投入生产会面临比 demo 复杂得多的问题。5.1 安全与权限认证与授权智能体间的调用必须有身份验证。可以使用 API 密钥、JWT 或 mTLS。输入验证与净化每个智能体必须严格验证输入防止注入攻击。输出过滤智能体生成的内容尤其是 LLM 驱动的可能需要经过安全过滤后才能传递给下一个智能体或用户。5.2 性能与可靠性超时与重试网络调用必须设置合理的超时并实现重试机制最好有退避策略。限流与熔断防止一个慢速或故障的智能体拖垮整个系统。为每个智能体设置并发限制和熔断器。异步通信对于耗时任务应采用异步模式。协调者发送请求后立即返回智能体通过回调或消息队列通知结果。监控与日志所有消息的流入流出、每个智能体的处理耗时和结果状态都必须有详细的日志和指标便于监控和排错。5.3 常见问题排查表问题现象可能原因检查方式处理建议协调者返回“No agent found”1. 注册中心未运行或不可达。2. 智能体未成功注册。3. 任务类型与智能体声明的能力不匹配。1. 检查注册中心服务状态和日志。2. 检查智能体启动日志确认注册请求是否成功。3. 查询注册中心查看当前已注册的智能体及其能力列表。确保注册中心先于智能体启动。检查智能体技能描述文件中的capabilities字段是否包含协调者请求的任务类型。智能体间调用超时1. 网络问题或防火墙规则。2. 目标智能体进程崩溃或负载过高。3. 目标智能体处理逻辑存在死循环或长时间阻塞。1. 使用ping、telnet或curl测试网络连通性。2. 检查目标智能体的进程状态、CPU/内存使用率和日志。3. 在目标智能体代码中添加超时和日志定位慢速操作。实现调用端的超时设置和重试机制。在目标智能体端实现健康检查接口并在负载高时返回 503 状态码。优化处理逻辑。消息格式解析错误1. 发送方和接收方使用的AgentMessage模型版本不一致。2. 消息内容不符合content字段定义的 schema。3. 序列化/反序列化库如 Pydantic配置不同。1. 对比双方shared/models.py的版本和字段定义。2. 在接收方日志中打印原始消息内容验证其结构。3. 检查是否有字段类型不匹配如字符串传成了数字。使用共享的、版本化的模型定义库。在消息处理入口添加严格的 schema 验证和详细的错误日志。考虑使用 Protocol Buffers 或 Avro 等更严格的序列化方案。上下文丢失或错乱1.context_id或parent_msg_id传递错误或丢失。2. 共享上下文服务如 Redis数据过期或丢失。3. 多个会话的context_id发生冲突。1. 在每条消息的日志中记录其context_id和parent_msg_id追踪传递链。2. 检查共享上下文服务的存储状态和过期策略。3. 确保context_id生成算法如 UUID的全局唯一性。将context_id作为必需字段在所有消息中传递。为共享上下文实现持久化存储和备份。使用分布式锁来管理对同一上下文的并发修改。工作流卡在某个步骤1. 该步骤的智能体失败且未正确返回错误响应。2. 工作流引擎的状态机出现错误。3. 任务依赖条件未满足如等待超时。1. 检查卡住步骤对应的智能体日志和返回消息。2. 检查工作流引擎的持久化状态如数据库中的任务状态。3. 检查前置任务是否成功完成输出是否符合预期。为工作流中的每个任务设置明确的超时和失败处理策略如重试、跳过、终止整个工作流。增强工作流引擎的监控和告警能力。5.4 扩展方向与最佳实践标准化协议深入研究和采纳社区正在形成的标准如 MCP 的演进版本可以减少自研协议的维护成本。智能体自治性考虑让智能体具备一定的自主决策能力例如在无法完成任务时能主动在注册中心寻找其他能帮忙的智能体。可观测性建立完善的可观测性体系包括链路追踪为每个context_id生成 Trace、指标监控QPS、耗时、错误率和集中式日志。这对于调试复杂的多智能体交互至关重要。测试策略为每个智能体编写单元测试和集成测试。模拟其他智能体的请求验证其协议兼容性和功能正确性。对整个工作流进行端到端测试。版本管理智能体的技能描述、通信协议、API 端点都可能演进。需要设计清晰的版本管理策略支持向后兼容或平滑升级。构建一个让智能体有效协作的系统其复杂度不亚于构建一个微服务架构。核心在于定义清晰的边界、稳定的通信契约和可靠的协调机制。从本文的最小可行系统出发逐步引入服务发现、工作流引擎、共享上下文和监控告警你就能搭建起属于你自己的、能够应对复杂现实任务的“AgentCouch”。 构建多智能体协作系统：从协议设计到工程实践
在实际 AI 应用开发中构建一个能够独立完成复杂任务的智能体Agent已经不再是难题。然而当任务链条变长、需要多步骤协作或涉及不同专业领域时单个智能体往往力不从心。这时一个自然的想法是能否让多个智能体像团队成员一样围绕一个共同目标进行对话、分工与协作这正是多智能体协作Multi-Agent Collaboration领域的核心课题。AgentCouch 这一概念形象地描绘了让智能体们“坐在沙发上聊天”的场景其背后指向的是构建一个支持智能体间高效、结构化通信与协作的框架或平台。这种协作并非简单的消息转发而是需要解决智能体如何发现彼此、如何理解对方的能力、如何协商任务、如何传递复杂的上下文如代码、设计稿、数据等一系列工程挑战。对于希望将 AI 能力从单点工具升级为自动化工作流的开发者而言理解并实践多智能体协作是必经之路。本文将围绕如何构建一个支持智能体间对话的协作系统展开。我们将从核心协议与概念入手逐步搭建一个最小化的多智能体通信骨架并深入探讨智能体能力描述、任务编排、上下文管理等关键问题。无论你是希望集成现有 AI 服务还是从零开始设计专属的智能体团队本文提供的思路和代码示例都将为你提供一个坚实的起点。1. 理解多智能体协作的核心协议与通信模型在让智能体们“聊天”之前必须先为它们定义一套共同的语言和交互规则。这类似于人类团队中的沟通协议和会议流程。1.1 为什么需要专门的通信协议单个智能体通常通过自然语言与用户交互其内部状态和决策过程对外是不可见的。当多个智能体协作时如果仅靠转发用户的自然语言指令会导致严重的信息损失和歧义。例如智能体A生成了一段代码智能体B需要审查它。如果只传递代码文本B可能不知道这段代码的用途、所属模块、或需要满足的测试用例。因此协作协议需要能封装丰富的结构化信息。目前业界逐渐形成了一些共识和雏形标准例如MCPModel Context Protocol的概念。虽然 MCP 的具体定义可能因项目而异但其核心思想是提供一种标准化的方式让不同的“模型”或智能体能够访问和操作共享的“上下文”Context。这个上下文可以包括文档、代码库、数据库 schema、API 文档、会话历史等。在多智能体场景中MCP 或其类似协议可以演化为智能体间交换上下文信息的载体。1.2 智能体协作的基本通信模式智能体间的对话通常不是随意的闲聊而是围绕任务的、有结构的交互。主要模式包括请求-响应模式这是最基础的同步模式。智能体 A 向智能体 B 发送一个明确的请求如“请分析这段日志”B 处理并返回结果。这类似于 HTTP 或 RPC 调用。发布-订阅模式适用于事件驱动的场景。当某个智能体完成一项工作如“数据库备份完成”或检测到一个状态如“系统负载过高”时它会向一个消息通道发布事件。其他关心此事件的智能体订阅该通道并作出响应。广播与协商模式当一个任务需要多个智能体共同决策时发起者可以向所有相关智能体广播任务信息收集大家的“意见”能力声明、预估耗时、所需资源等然后进行协商或指派。链式或工作流模式智能体按照预定义的流程依次执行上一个的输出是下一个的输入。这需要工作流引擎来编排顺序和处理分支。在实现层面这些模式可以通过消息队列如 RabbitMQ、Kafka、WebSocket、或简单的 HTTP 服务来实现。选择哪种模式取决于智能体协作的实时性、耦合度和复杂度要求。1.3 定义智能体的“技能”与“身份”为了让智能体能有效地找到协作者每个智能体需要清晰地声明自己的能力。这通常通过一个“技能描述”文件来实现。这个描述文件应该包含智能体 ID唯一标识符。能力列表该智能体擅长处理的任务类型例如code_review、sql_generation、ui_design_critique。输入/输出格式它能接受什么格式的输入如 JSON Schema、文本、特定类型的文件以及它返回数据的格式。调用端点其他智能体如何调用它如 REST API 地址、消息队列的主题名。元数据版本、作者、所需资源等。# 示例一个代码审查智能体的技能描述 (agent_skill.yaml) agent_id: code_reviewer_v1 name: Python 代码审查专家 description: 专注于审查 Python 代码的语法、风格和潜在 bug。 capabilities: - name: review_python_code description: 审查给定的 Python 代码片段。 input_schema: type: object properties: code: type: string description: 待审查的 Python 代码 context: type: string description: 代码的上下文或需求描述可选 required: [code] output_schema: type: object properties: issues: type: array items: type: object properties: line: type: integer severity: type: string enum: [ERROR, WARNING, INFO] message: type: string suggestion: type: string summary: type: string endpoint: type: http url: http://localhost:8081/review method: POST metadata: version: 1.0.0 language: python有了这样的描述一个协调者智能体或服务注册中心就能知道系统中有哪些可用的“专家”并能根据任务类型进行匹配和路由。2. 搭建多智能体通信的基础骨架我们将从一个最简单的场景开始两个智能体通过 HTTP 进行直接的请求-响应式对话。我们将创建两个简单的 Python 服务来模拟智能体。2.1 环境准备与项目结构首先确保你的开发环境已安装 Python 3.8。我们将使用FastAPI来快速构建 HTTP 服务因为它轻量且易于定义 API。创建一个项目目录并初始化虚拟环境mkdir agent-couch-demo cd agent-couch-demo python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate pip install fastapi uvicorn requests pydantic项目结构如下agent-couch-demo/ ├── coordinator.py # 协调者服务可选用于路由 ├── agents/ │ ├── __init__.py │ ├── code_reviewer.py # 代码审查智能体 │ └── sql_expert.py # SQL 生成智能体 ├── shared/ │ ├── __init__.py │ └── models.py # 共享的数据模型消息格式 └── requirements.txt2.2 定义智能体间消息协议在shared/models.py中我们定义智能体间通信的基本消息格式。这是实现“共同语言”的关键。# shared/models.py from pydantic import BaseModel, Field from typing import Any, Dict, List, Optional from enum import Enum class MessageType(str, Enum): 消息类型枚举 TASK_REQUEST task_request # 任务请求 TASK_RESPONSE task_response # 任务响应 BROADCAST broadcast # 广播消息 ERROR error # 错误消息 class AgentMessage(BaseModel): 智能体间通信的基础消息模型 msg_id: str Field(..., description消息唯一ID) sender_id: str Field(..., description发送者智能体ID) receiver_id: Optional[str] Field(None, description接收者智能体IDNone表示广播) msg_type: MessageType Field(..., description消息类型) content: Dict[str, Any] Field(..., description消息内容结构由msg_type决定) timestamp: str Field(..., description消息发送时间戳) # 上下文信息用于追踪任务链 context_id: Optional[str] Field(None, description关联的上下文或会话ID) parent_msg_id: Optional[str] Field(None, description父消息ID用于回复链) class TaskRequestContent(BaseModel): TASK_REQUEST 类型的消息内容结构 task_type: str Field(..., description任务类型如 code_review, generate_sql) task_params: Dict[str, Any] Field(..., description任务参数) required_capabilities: Optional[List[str]] Field(None, description完成任务所需的能力列表) class TaskResponseContent(BaseModel): TASK_RESPONSE 类型的消息内容结构 task_id: str Field(..., description对应的任务请求ID) status: str Field(..., description任务状态如 success, failed, partial) result: Optional[Dict[str, Any]] Field(None, description任务执行结果) error_info: Optional[str] Field(None, description错误信息)这个模型确保了消息的规范性。AgentMessage是信封content字段内的具体结构由msg_type决定。2.3 实现第一个智能体代码审查者在agents/code_reviewer.py中我们实现一个简单的代码审查智能体。它提供一个 HTTP 端点接收代码并返回审查意见。# agents/code_reviewer.py from fastapi import FastAPI, HTTPException import uvicorn from shared.models import AgentMessage, MessageType, TaskRequestContent, TaskResponseContent from pydantic import BaseModel import uuid from datetime import datetime app FastAPI(titleCode Reviewer Agent) # 模拟一个简单的代码审查逻辑 def review_python_code(code: str, context: str ) - dict: issues [] lines code.split(\n) for i, line in enumerate(lines, start1): line line.rstrip() # 一些简单的规则检查 if len(line) 100: issues.append({ line: i, severity: WARNING, message: f行 {i} 超过 100 字符, suggestion: 考虑拆分行或简化表达式 }) if print( in line and TODO not in context.upper(): # 假设在生产代码中不鼓励直接使用print issues.append({ line: i, severity: INFO, message: f行 {i} 使用了 print 语句, suggestion: 考虑使用 logging 模块以便于控制输出级别 }) # 可以在这里添加更多检查如导入风格、命名规范等 return { issues: issues, summary: f发现 {len(issues)} 个潜在问题。, reviewed_at: datetime.utcnow().isoformat() } class ReviewRequest(BaseModel): code: str context: str app.post(/review) async def do_review(request: ReviewRequest): 对外提供的 API 端点 try: result review_python_code(request.code, request.context) return {status: success, data: result} except Exception as e: raise HTTPException(status_code500, detailf审查过程出错: {str(e)}) app.post(/agent_message) async def handle_agent_message(message: AgentMessage): 处理来自其他智能体的标准化消息 if message.msg_type ! MessageType.TASK_REQUEST: return AgentMessage( msg_idstr(uuid.uuid4()), sender_idcode_reviewer_v1, receiver_idmessage.sender_id, msg_typeMessageType.ERROR, content{error: f不支持的消息类型: {message.msg_type}}, timestampdatetime.utcnow().isoformat(), context_idmessage.context_id, parent_msg_idmessage.msg_id ) try: # 解析任务请求内容 task_content TaskRequestContent(**message.content) if task_content.task_type ! code_review: raise ValueError(f本智能体不支持的任务类型: {task_content.task_type}) code task_content.task_params.get(code) if not code: raise ValueError(任务参数中缺少 code 字段) context task_content.task_params.get(context, ) # 执行核心审查逻辑 review_result review_python_code(code, context) # 构建响应消息 response_content TaskResponseContent( task_idmessage.msg_id, statussuccess, resultreview_result ) return AgentMessage( msg_idstr(uuid.uuid4()), sender_idcode_reviewer_v1, receiver_idmessage.sender_id, msg_typeMessageType.TASK_RESPONSE, contentresponse_content.dict(), timestampdatetime.utcnow().isoformat(), context_idmessage.context_id, parent_msg_idmessage.msg_id ) except Exception as e: # 返回错误响应 error_response TaskResponseContent( task_idmessage.msg_id, statusfailed, error_infostr(e) ) return AgentMessage( msg_idstr(uuid.uuid4()), sender_idcode_reviewer_v1, receiver_idmessage.sender_id, msg_typeMessageType.TASK_RESPONSE, contenterror_response.dict(), timestampdatetime.utcnow().isoformat(), context_idmessage.context_id, parent_msg_idmessage.msg_id ) if __name__ __main__: # 启动服务在 8081 端口 uvicorn.run(app, host0.0.0.0, port8081)这个智能体提供了两个端点/review: 一个简单的 REST API供外部直接调用。/agent_message: 专用于智能体间通信的端点接收和返回标准化的AgentMessage。这体现了“协议”的重要性——智能体间使用一种更丰富、更结构化的方式交流。2.4 实现第二个智能体与协调者我们再创建一个 SQL 专家智能体 (agents/sql_expert.py)其结构与审查者类似但能力是生成 SQL。为了演示智能体间的对话我们还需要一个简单的协调者 (coordinator.py)。协调者的作用是接收用户或系统的原始任务将其分解并路由给合适的智能体。# coordinator.py from fastapi import FastAPI, HTTPException import uvicorn import requests import uuid from datetime import datetime from shared.models import AgentMessage, MessageType, TaskRequestContent from pydantic import BaseModel from typing import Dict app FastAPI(titleAgent Coordinator) # 简单的智能体注册表实际项目中可能使用服务发现如 Consul AGENT_REGISTRY { code_review: {agent_id: code_reviewer_v1, endpoint: http://localhost:8081/agent_message}, generate_sql: {agent_id: sql_expert_v1, endpoint: http://localhost:8082/agent_message}, } class UserRequest(BaseModel): task: str # 如 review_code_and_generate_sql parameters: Dict[str, any] def send_to_agent(agent_info: dict, message: AgentMessage) - AgentMessage: 向指定智能体发送消息并获取响应 try: resp requests.post(agent_info[endpoint], jsonmessage.dict(), timeout30) resp.raise_for_status() return AgentMessage(**resp.json()) except requests.exceptions.RequestException as e: # 构建一个本地的错误响应消息 error_content { task_id: message.msg_id, status: failed, error_info: f无法连接到智能体 {agent_info[agent_id]}: {str(e)} } return AgentMessage( msg_idstr(uuid.uuid4()), sender_idcoordinator, receiver_idmessage.sender_id, msg_typeMessageType.TASK_RESPONSE, contenterror_content, timestampdatetime.utcnow().isoformat(), context_idmessage.context_id, parent_msg_idmessage.msg_id ) app.post(/orchestrate) async def orchestrate_task(request: UserRequest): 协调任务这是一个简单的顺序工作流示例 context_id str(uuid.uuid4()) # 为本次用户请求创建唯一上下文ID if request.task review_code_and_generate_sql: # 步骤1将代码发送给审查者 review_agent AGENT_REGISTRY[code_review] review_msg AgentMessage( msg_idstr(uuid.uuid4()), sender_idcoordinator, receiver_idreview_agent[agent_id], msg_typeMessageType.TASK_REQUEST, contentTaskRequestContent( task_typecode_review, task_paramsrequest.parameters ).dict(), timestampdatetime.utcnow().isoformat(), context_idcontext_id ) review_response send_to_agent(review_agent, review_msg) # 步骤2根据审查结果这里简单处理再调用 SQL 专家 # 注意实际逻辑可能需要解析 review_response 的内容来决定下一步 sql_agent AGENT_REGISTRY[generate_sql] # 假设我们从用户参数中提取一个需求描述来生成 SQL sql_task_params {description: request.parameters.get(description, 生成查询用户表的SQL)} sql_msg AgentMessage( msg_idstr(uuid.uuid4()), sender_idcoordinator, receiver_idsql_agent[agent_id], msg_typeMessageType.TASK_REQUEST, contentTaskRequestContent( task_typegenerate_sql, task_paramssql_task_params ).dict(), timestampdatetime.utcnow().isoformat(), context_idcontext_id, parent_msg_idreview_response.msg_id # 关联到上一步的消息 ) sql_response send_to_agent(sql_agent, sql_msg) # 汇总结果返回给用户 return { context_id: context_id, workflow: request.task, steps: [ {agent: code_reviewer, response: review_response.dict()}, {agent: sql_expert, response: sql_response.dict()} ] } else: raise HTTPException(status_code400, detailf不支持的任务类型: {request.task}) if __name__ __main__: uvicorn.run(app, host0.0.0.0, port8080)这个协调者实现了一个最简单的顺序工作流。它展示了如何维护一个智能体注册表。将用户请求转换为标准化的智能体间消息。按顺序调用多个智能体并传递上下文。汇总结果。3. 运行与验证多智能体对话现在让我们启动这个微型的多智能体系统并验证它们是否能成功“对话”。3.1 启动所有服务打开三个终端窗口分别运行# 终端1启动代码审查智能体 cd agent-couch-demo python -m agents.code_reviewer # 服务启动在 http://localhost:8081 # 终端2启动 SQL 专家智能体假设已实现结构类似 # python -m agents.sql_expert # 服务启动在 http://localhost:8082 为了演示我们可以先跳过用模拟响应 # 终端3启动协调者 python coordinator.py # 服务启动在 http://localhost:8080为了简化我们可以先不实现完整的 SQL 专家而是修改协调者在第二步模拟一个成功的 SQL 生成响应。3.2 测试智能体间直接对话首先测试代码审查智能体是否能正确处理来自协调者的标准化消息。我们可以使用curl或 Python 的requests库来模拟协调者发送消息。# test_direct_message.py import requests import json from datetime import datetime import uuid # 构建一个标准的 AgentMessage message { msg_id: str(uuid.uuid4()), sender_id: test_coordinator, receiver_id: code_reviewer_v1, msg_type: task_request, content: { task_type: code_review, task_params: { code: def calculate_sum(a, b):\n result a b\n print(fThe sum is {result})\n return result, context: 这是一个计算两数之和的函数请审查。 } }, timestamp: datetime.utcnow().isoformat(), context_id: str(uuid.uuid4()) } url http://localhost:8081/agent_message response requests.post(url, jsonmessage) print(Status Code:, response.status_code) print(Response Body:, json.dumps(response.json(), indent2, ensure_asciiFalse))运行这个测试脚本你应该会收到一个结构化的响应其中包含审查结果。这证明了两个服务之间能够使用我们定义的协议进行通信。3.3 测试完整的工作流现在通过协调者的接口来测试整个工作流。curl -X POST http://localhost:8080/orchestrate \ -H Content-Type: application/json \ -d { task: review_code_and_generate_sql, parameters: { code: for i in range(10):\n print(i), description: 需要一个查询所有活跃用户的SQL } }协调者会依次调用代码审查智能体和 SQL 专家智能体或模拟并将两个步骤的结果汇总返回。查看响应你应该能看到一个包含context_id和两个步骤详细响应的 JSON 对象。3.4 验证关键点通过以上测试我们验证了多智能体协作系统的几个核心能力协议通信智能体间使用AgentMessage进行结构化数据交换而不是纯文本。服务发现协调者通过注册表知道每个智能体的能力和地址。任务路由协调者根据任务类型 (task_type) 将请求路由到正确的智能体。上下文关联通过context_id和parent_msg_id可以追踪一个任务链中的所有消息。错误处理消息处理过程中有基本的异常捕获和错误响应格式。4. 深入探讨从基础对话到复杂协作上面的例子是一个高度简化的模型。要让智能体们真正高效地“坐在沙发上聊天”还需要解决更多工程问题。4.1 智能体能力发现与动态注册硬编码的注册表 (AGENT_REGISTRY) 不适合生产环境。我们需要一个服务发现机制。常见的模式是注册中心每个智能体启动时向一个中心服务如 Consul、Etcd、或一个简单的注册服务注册自己的技能描述见 1.3 节。心跳与健康检查注册中心定期检查智能体是否存活将不可用的智能体从可用列表中移除。能力查询协调者或需要帮助的智能体可以向注册中心查询“谁有能力做code_review”。# 伪代码智能体启动时自动注册 def register_agent(agent_skill_yaml_path, registry_url): with open(agent_skill_yaml_path, r) as f: skill_info yaml.safe_load(f) requests.post(f{registry_url}/register, jsonskill_info) # 伪代码协调者动态查找智能体 def find_agent_for_task(task_type, registry_url): response requests.get(f{registry_url}/discover, params{capability: task_type}) agents response.json() if agents: # 可以在这里实现负载均衡或选择策略 return agents[0] else: raise Exception(fNo agent found for task: {task_type})4.2 对话管理与上下文持久化一次复杂的协作可能涉及多轮对话。例如审查者可能要求生成者提供更多上下文生成者可能需要向用户确认。这需要维护一个“会话”状态。会话服务提供一个中心化的服务来管理会话。每个会话有一个唯一 ID关联所有相关的消息、智能体状态和最终结果。消息持久化将所有AgentMessage存储到数据库如 MongoDB、PostgreSQL便于调试、审计和实现“断点续聊”。上下文注入智能体在处理消息时可以从会话服务中获取整个对话历史从而做出更连贯的决策。4.3 编排引擎与工作流定义我们的协调者只是一个简单的顺序执行器。复杂的任务可能需要并行执行、条件分支、循环、错误重试等。此时需要引入工作流引擎如Apache Airflow、Prefect、或 Temporal。你可以用 YAML 或 Python DSL 定义工作流# workflow_definition.yaml name: data_processing_and_review tasks: - id: fetch_data type: http_request agent: data_fetcher params: {url: {{input.data_url}}} - id: clean_data type: data_transform agent: data_cleaner params: {raw_data: {{tasks.fetch_data.output}}} depends_on: [fetch_data] - id: review_cleaning_code type: code_review agent: code_reviewer params: {code: {{tasks.clean_data.metadata.code_snippet}}} depends_on: [clean_data] - id: generate_report type: report_generation agent: report_generator params: data: {{tasks.clean_data.output}} review_comments: {{tasks.review_cleaning_code.output}} depends_on: [clean_data, review_cleaning_code]编排引擎负责解析这个定义创建任务实例管理依赖调用相应的智能体并处理失败和重试。4.4 共享上下文与 MCP 思想这是实现深度协作的关键。智能体 A 生成的图表、智能体 B 编写的代码、智能体 C 查询的数据如何让智能体 D 无缝使用这就是MCPModel Context Protocol类协议要解决的问题。其核心是定义一个标准的、工具可读的“上下文”格式。一个简单的共享上下文服务可以这样设计# shared_context.py from typing import Dict, Any import json class SharedContextService: def __init__(self): self._context_store {} # context_id - {resources: {...}} def put_resource(self, context_id: str, resource_type: str, resource_id: str, content: Dict[str, Any]): 将一个资源放入共享上下文 if context_id not in self._context_store: self._context_store[context_id] {resources: {}} key f{resource_type}:{resource_id} self._context_store[context_id][resources][key] { content: content, created_by: some_agent_id, created_at: timestamp } def get_resource(self, context_id: str, resource_type: str, resource_id: str): 从共享上下文获取一个资源 key f{resource_type}:{resource_id} return self._context_store.get(context_id, {}).get(resources, {}).get(key) def list_resources(self, context_id: str, resource_type: str None): 列出共享上下文中的所有资源或某类资源 resources self._context_store.get(context_id, {}).get(resources, {}) if resource_type: return {k: v for k, v in resources.items() if k.startswith(f{resource_type}:)} return resources # 智能体在生成代码后可以将其存入共享上下文 context_service.put_resource( context_idsession_123, resource_typecode, resource_iddata_cleaner_module_v1, content{language: python, code: def clean(x): ..., dependencies: [pandas]} ) # 另一个智能体可以获取并使用这段代码 code_obj context_service.get_resource(session_123, code, data_cleaner_module_v1)这样智能体间的协作就不再是孤立的请求-响应而是围绕一个不断丰富的共享上下文进行建设。5. 生产环境考量与常见问题排查将多智能体系统投入生产会面临比 demo 复杂得多的问题。5.1 安全与权限认证与授权智能体间的调用必须有身份验证。可以使用 API 密钥、JWT 或 mTLS。输入验证与净化每个智能体必须严格验证输入防止注入攻击。输出过滤智能体生成的内容尤其是 LLM 驱动的可能需要经过安全过滤后才能传递给下一个智能体或用户。5.2 性能与可靠性超时与重试网络调用必须设置合理的超时并实现重试机制最好有退避策略。限流与熔断防止一个慢速或故障的智能体拖垮整个系统。为每个智能体设置并发限制和熔断器。异步通信对于耗时任务应采用异步模式。协调者发送请求后立即返回智能体通过回调或消息队列通知结果。监控与日志所有消息的流入流出、每个智能体的处理耗时和结果状态都必须有详细的日志和指标便于监控和排错。5.3 常见问题排查表问题现象可能原因检查方式处理建议协调者返回“No agent found”1. 注册中心未运行或不可达。2. 智能体未成功注册。3. 任务类型与智能体声明的能力不匹配。1. 检查注册中心服务状态和日志。2. 检查智能体启动日志确认注册请求是否成功。3. 查询注册中心查看当前已注册的智能体及其能力列表。确保注册中心先于智能体启动。检查智能体技能描述文件中的capabilities字段是否包含协调者请求的任务类型。智能体间调用超时1. 网络问题或防火墙规则。2. 目标智能体进程崩溃或负载过高。3. 目标智能体处理逻辑存在死循环或长时间阻塞。1. 使用ping、telnet或curl测试网络连通性。2. 检查目标智能体的进程状态、CPU/内存使用率和日志。3. 在目标智能体代码中添加超时和日志定位慢速操作。实现调用端的超时设置和重试机制。在目标智能体端实现健康检查接口并在负载高时返回 503 状态码。优化处理逻辑。消息格式解析错误1. 发送方和接收方使用的AgentMessage模型版本不一致。2. 消息内容不符合content字段定义的 schema。3. 序列化/反序列化库如 Pydantic配置不同。1. 对比双方shared/models.py的版本和字段定义。2. 在接收方日志中打印原始消息内容验证其结构。3. 检查是否有字段类型不匹配如字符串传成了数字。使用共享的、版本化的模型定义库。在消息处理入口添加严格的 schema 验证和详细的错误日志。考虑使用 Protocol Buffers 或 Avro 等更严格的序列化方案。上下文丢失或错乱1.context_id或parent_msg_id传递错误或丢失。2. 共享上下文服务如 Redis数据过期或丢失。3. 多个会话的context_id发生冲突。1. 在每条消息的日志中记录其context_id和parent_msg_id追踪传递链。2. 检查共享上下文服务的存储状态和过期策略。3. 确保context_id生成算法如 UUID的全局唯一性。将context_id作为必需字段在所有消息中传递。为共享上下文实现持久化存储和备份。使用分布式锁来管理对同一上下文的并发修改。工作流卡在某个步骤1. 该步骤的智能体失败且未正确返回错误响应。2. 工作流引擎的状态机出现错误。3. 任务依赖条件未满足如等待超时。1. 检查卡住步骤对应的智能体日志和返回消息。2. 检查工作流引擎的持久化状态如数据库中的任务状态。3. 检查前置任务是否成功完成输出是否符合预期。为工作流中的每个任务设置明确的超时和失败处理策略如重试、跳过、终止整个工作流。增强工作流引擎的监控和告警能力。5.4 扩展方向与最佳实践标准化协议深入研究和采纳社区正在形成的标准如 MCP 的演进版本可以减少自研协议的维护成本。智能体自治性考虑让智能体具备一定的自主决策能力例如在无法完成任务时能主动在注册中心寻找其他能帮忙的智能体。可观测性建立完善的可观测性体系包括链路追踪为每个context_id生成 Trace、指标监控QPS、耗时、错误率和集中式日志。这对于调试复杂的多智能体交互至关重要。测试策略为每个智能体编写单元测试和集成测试。模拟其他智能体的请求验证其协议兼容性和功能正确性。对整个工作流进行端到端测试。版本管理智能体的技能描述、通信协议、API 端点都可能演进。需要设计清晰的版本管理策略支持向后兼容或平滑升级。构建一个让智能体有效协作的系统其复杂度不亚于构建一个微服务架构。核心在于定义清晰的边界、稳定的通信契约和可靠的协调机制。从本文的最小可行系统出发逐步引入服务发现、工作流引擎、共享上下文和监控告警你就能搭建起属于你自己的、能够应对复杂现实任务的“AgentCouch”。