1. 从单体到集群为什么我们需要超级多智能体1.1 一个真实的需求场景去年下半年我接手了一个企业内部知识助手的项目需求听起来不复杂帮员工查制度文档、走审批流程、生成周报。一开始我用的是单体 Agent 方案一个模型加一堆工具函数跑起来确实快。但上线两周后问题就来了——查文档的 Agent 经常被审批流程的上下文污染生成周报时又因为要调用太多工具导致响应时间飙到十几秒。最要命的是我想给文档检索单独换一个更便宜的模型结果发现整个系统耦合在一起改一处动全身。这个经历让我彻底转向了多智能体架构。所谓超级多智能体不是简单地把几个 Agent 拼在一起而是要解决三个核心问题可编排谁先谁后、谁调用谁、可互通Agent 之间怎么传消息、怎么共享状态、可扩展新增一个 Agent 要不要改老代码。这三个问题对应到技术选型上就是 DeepAgents 负责编排、MCP 负责工具接入、A2A 负责 Agent 间通信、Skills 负责能力封装。如果你正在做 Agent 项目或者被单体 Agent 的耦合问题折磨过这套组合拳值得花时间研究。它不依赖某个特定厂商核心思路可以迁移到任何 Agent 框架上。1.2 四个关键词到底在说什么先把概念理清楚不然后面容易绕晕。DeepAgents在这里指的是一种深度编排的 Agent 架构模式核心思想是把复杂任务拆成有向图每个节点是一个 Agent 或一个工具调用边代表数据流和控制流。它和简单的链式调用Chain最大的区别在于支持条件分支、循环、并行和人工介入节点。MCPModel Context Protocol是一个开放协议解决的是Agent 怎么标准化地调用外部工具和数据源的问题。你可以把它理解成 Agent 世界的 USB 接口——不管你是数据库、文件系统还是第三方 API只要实现 MCP 协议任何支持 MCP 的 Agent 都能直接接入。热搜里有人问mcp 是软件协议还是硬件协议答案是软件协议它定义的是通信格式和调用规范跟硬件层面的协议比如 USB、I2C完全是两回事。A2AAgent to Agent解决的是 Agent 之间的通信问题。MCP 管的是 Agent 调工具A2A 管的是 Agent 调 Agent。它定义了 Agent 之间怎么发现对方、怎么发消息、怎么处理异步响应。热搜里出现的a2a spring说明已经有人在 Spring 生态里做 A2A 的集成了。Skills是能力的封装单元。一个 Skill 可以是一个提示词模板、一段代码逻辑、一组工具调用的组合。它的价值在于复用——你写好一个生成周报的 Skill可以在多个 Agent 里直接引用不用重复实现。热搜里claude agent skills: a first principles deep dive和codex skills都指向同一个趋势Skills 正在成为 Agent 能力复用的标准方式。1.3 这套架构适合谁不是所有项目都需要上多智能体。如果你的需求是单轮问答、简单工具调用单体 Agent 完全够用硬上多智能体只会增加复杂度。但如果你遇到以下情况这套架构就值得考虑任务流程超过 5 个步骤且步骤之间有依赖关系需要多个专业领域的 Agent 协作比如一个查数据、一个做分析、一个写报告工具数量超过 10 个单体 Agent 的提示词已经塞不下需要给不同 Agent 配不同模型来控制成本系统需要频繁新增能力不想每次改核心代码下面我会按照实际搭建的顺序从架构设计到落地实现把每个环节讲透。2. 架构设计编排层、通信层、能力层怎么分2.1 三层架构的职责边界我在实际项目中把系统分成三层每层职责清晰层与层之间通过标准协议通信。编排层DeepAgents负责决策任务怎么拆、下一步走哪个节点、什么时候结束。这一层不关心具体工具怎么调只关心流程控制。我通常用一个有向无环图DAG来描述流程节点类型包括Agent 节点、工具节点、条件判断节点、人工审核节点、并行分支节点。通信层A2A MCP负责传输Agent 之间怎么传消息用 A2AAgent 调工具用 MCP。这一层的关键是协议标准化只要协议统一底层换实现不影响上层。能力层Skills负责执行每个 Skill 是一个独立的能力单元包含提示词、工具绑定、输入输出定义。Skill 可以注册到多个 Agent 上实现复用。这样分层的最大好处是新增一个 Agent 只需要在编排层加节点、在能力层加 Skill通信层完全不用动。我试过在一个已有 8 个 Agent 的系统里新增一个合同审查Agent从写 Skill 到接入流程只花了半天。2.2 为什么选 DAG 而不是状态机编排层可以用状态机也可以用 DAG。我最终选 DAG 的原因是状态机适合线性流程但多智能体场景下经常需要并行和条件分支。比如同时查三个数据源然后汇总这种需求用状态机写起来很别扭DAG 天然支持并行节点。DAG 的另一个好处是可观测。每个节点的输入输出都能记录下来出问题时可以精确定位是哪个节点出了错。我在生产环境里给每个节点都加了日志埋点排查问题时直接看节点级别的 trace比看一整条链路的日志高效得多。2.3 通信协议选型MCP 和 A2A 的分工很多人搞不清 MCP 和 A2A 的区别我用一个类比说明MCP 像是人用工具A2A 像是人跟人说话。你用手电筒工具不需要跟手电筒商量直接按开关就行这是 MCP。但你要跟同事协作就得说话、等回复、确认理解这是 A2A。具体到实现上MCP 的调用是同步或异步的请求-响应模式Agent 发一个调用请求工具返回结果。A2A 则更复杂支持流式响应、多轮对话、任务委派。我在项目里用 MCP 接入了文件系统、数据库、搜索引擎三类工具用 A2A 实现了研究 Agent和写作 Agent之间的协作。注意MCP 和 A2A 不是互斥的。一个 Agent 可以同时是 MCP 的客户端调工具和 A2A 的服务端被其他 Agent 调用。这种双重身份在多智能体系统里很常见。2.4 Skills 的粒度怎么把握Skills 的粒度设计是个经验活。太粗复用性差太细组合起来麻烦。我的原则是一个 Skill 对应一个完整的业务动作。比如查询员工考勤是一个 Skill它内部可能调了三个 MCP 工具查数据库、格式化数据、计算统计但对外只暴露一个接口。这样其他 Agent 引用时不用关心内部实现。反过来调用数据库这种太底层的操作不适合做成 Skill它应该是 MCP 工具。Skill 的抽象层级要高于工具低于完整的 Agent。3. 核心细节MCP 工具接入与 A2A 通信实现3.1 MCP 工具接入的完整流程MCP 的核心是标准化。不管你的工具是 Python 函数、HTTP API 还是数据库查询都要包装成 MCP 协议规定的格式。我以接入一个文档检索工具为例走一遍完整流程。首先定义工具的元数据包括名称、描述、输入参数 schema。这部分用 JSON Schema 描述目的是让 Agent 知道这个工具能干什么、需要什么参数。{ name: search_documents, description: 根据关键词检索企业制度文档, inputSchema: { type: object, properties: { query: {type: string, description: 检索关键词}, top_k: {type: integer, default: 5, description: 返回结果数量} }, required: [query] } }然后实现工具的执行逻辑。这部分可以用任何语言写只要最终暴露成 MCP 服务端即可。我用 Python 写了一个简单的实现def search_documents(query: str, top_k: int 5): # 实际检索逻辑 results vector_store.search(query, top_k) return [{title: r.title, content: r.content} for r in results]最后把工具注册到 MCP 服务端Agent 通过 MCP 客户端连接后就能自动发现并调用这个工具。这里有个坑要注意工具描述的质量直接决定 Agent 会不会正确调用。我一开始写的描述很简略结果 Agent 经常在不需要检索的时候也去调这个工具。后来我把描述改得更具体加上仅在用户询问制度、流程、规定时使用这样的约束误调用率明显下降。3.2 A2A 通信的消息格式设计A2A 通信的关键是消息格式要统一。我定义了一个基础消息结构包含发送方、接收方、消息类型、负载和上下文 ID。{ from: research_agent, to: writing_agent, type: task_delegation, context_id: task_20240115_001, payload: { task: 根据以下研究结果撰写报告, data: {...}, requirements: 字数 800 字包含数据引用 } }context_id很重要它让多个 Agent 之间的多轮交互能关联起来。比如研究 Agent 发了任务后写作 Agent 可能中途回来问问题这时候靠 context_id 就能找到原始任务上下文。A2A 的响应支持流式和非流式两种。流式适合长任务写作 Agent 可以边写边把内容推给调用方非流式适合短任务一次性返回结果。我在项目里默认用流式因为用户体验更好而且能提前发现生成方向跑偏的问题。3.3 Skills 的注册与发现机制Skills 要能被复用就需要一个注册中心。我实现了一个简单的 Skill Registry每个 Skill 注册时提供名称、描述、输入输出 schema 和执行入口。class SkillRegistry: def __init__(self): self.skills {} def register(self, name, description, schema, handler): self.skills[name] { description: description, schema: schema, handler: handler } def get(self, name): return self.skills.get(name) def list_all(self): return list(self.skills.keys())Agent 在初始化时从 Registry 拉取可用的 Skill 列表根据任务需要动态绑定。这样新增 Skill 不用改 Agent 代码只要注册进去就行。热搜里提到的find skills和skills 推荐反映了大家对这个机制的关注。我的经验是Skill 的发现不能只靠名称匹配还要有语义检索。我后来加了一个基于向量相似度的 Skill 检索Agent 用自然语言描述需求就能找到合适的 Skill比硬编码名称灵活得多。3.4 状态管理与上下文传递多智能体系统里状态管理是最容易出问题的地方。我的做法是全局状态用共享存储局部状态用消息传递。全局状态包括任务 ID、用户信息、会话历史存在 Redis 里所有 Agent 都能读写。局部状态包括当前节点的中间结果通过 A2A 消息在 Agent 之间传递不落盘。这样设计的好处是避免了状态不一致。我踩过的坑是早期把所有状态都放在消息里传结果一个 Agent 改了状态但没传给下一个导致数据丢失。后来把关键状态抽到共享存储问题就解决了。提示共享存储要加版本号或乐观锁防止多个 Agent 并发写同一份数据时互相覆盖。4. 实操落地从零搭建一个多智能体系统4.1 环境准备与依赖安装我以 Python 技术栈为例列出核心依赖。如果你用其他语言思路是一样的只是库不同。pip install fastapi uvicorn redis pydantic pip install mcp-client mcp-server pip install a2a-sdk pip install langgraph # 用于 DAG 编排MCP 和 A2A 的 SDK 目前还在快速迭代建议锁定版本号避免升级导致接口不兼容。我在项目里用 requirements.txt 固定了版本每次升级前先在测试环境验证。Redis 用于共享状态存储FastAPI 用于暴露 HTTP 接口Pydantic 用于数据校验。LangGraph 是我常用的 DAG 编排库它原生支持条件分支和循环比手写状态机省事。4.2 定义第一个 Agent 和 Skill先定义一个最简单的文档检索 Agent它绑定一个检索文档的 Skill。from pydantic import BaseModel class SearchInput(BaseModel): query: str top_k: int 5 class SearchOutput(BaseModel): results: list total: int def search_handler(input: SearchInput) - SearchOutput: results vector_store.search(input.query, input.top_k) return SearchOutput(resultsresults, totallen(results)) # 注册 Skill registry.register( namesearch_documents, description检索企业制度文档适用于查询规定、流程、政策, schema{input: SearchInput, output: SearchOutput}, handlersearch_handler )然后定义 Agent它从 Registry 获取 Skill通过 MCP 调用工具通过 A2A 接收任务。class DocumentAgent: def __init__(self, registry, mcp_client, a2a_client): self.registry registry self.mcp mcp_client self.a2a a2a_client self.skill registry.get(search_documents) async def handle(self, message): input_data SearchInput(**message.payload) result self.skill[handler](input_data) return {status: success, data: result.dict()}这个 Agent 的结构很清晰接收消息、解析输入、执行 Skill、返回结果。新增能力只需要注册新 Skill 并绑定到 Agent 上。4.3 编排多个 Agent 的 DAG有了单个 Agent 后用 LangGraph 把它们编排成 DAG。我以一个员工咨询场景为例用户提问后先判断问题类型如果是制度类问题走文档 Agent如果是数据类问题走数据 Agent最后统一由汇总 Agent 生成回答。from langgraph.graph import StateGraph, END def route_question(state): if 制度 in state[question] or 流程 in state[question]: return document_agent elif 数据 in state[question] or 统计 in state[question]: return data_agent else: return general_agent graph StateGraph(AgentState) graph.add_node(router, route_question) graph.add_node(document_agent, document_node) graph.add_node(data_agent, data_node) graph.add_node(general_agent, general_node) graph.add_node(summarizer, summarize_node) graph.set_entry_point(router) graph.add_conditional_edges(router, route_question, { document_agent: document_agent, data_agent: data_agent, general_agent: general_agent }) graph.add_edge(document_agent, summarizer) graph.add_edge(data_agent, summarizer) graph.add_edge(general_agent, summarizer) graph.add_edge(summarizer, END) app graph.compile()这个 DAG 的关键点是路由节点。路由可以用规则实现也可以用一个小模型做意图分类。我在项目里用的是规则加模型兜底常见问题走规则规则匹配不到时调模型判断兼顾速度和准确率。4.4 参数计算与性能调优多智能体系统的性能瓶颈通常在两个地方Agent 间的通信延迟和模型调用次数。我做过一组实测数据供参考。场景Agent 数量平均响应时间模型调用次数单体 Agent12.1s1-2双 Agent 串行23.8s2-3三 Agent 并行34.2s3-4五 Agent 混合56.5s5-7从数据看Agent 数量增加会带来明显的延迟增长。优化方向有三个一是并行化能并行的节点不要串行二是缓存相同输入的结果缓存起来三是模型分级简单任务用便宜的小模型。我在项目里给每个 Agent 配了不同的模型路由用 7B 小模型文档检索用 13B 模型汇总生成用 70B 大模型。这样整体成本降了约 60%响应时间只增加了 15%。4.5 部署与监控部署上我推荐容器化每个 Agent 一个容器通过消息队列或 HTTP 通信。这样单个 Agent 出问题不影响整体也方便水平扩展。监控要覆盖三个层面节点级别的执行时间、Agent 级别的调用成功率、系统级别的吞吐量。我用 Prometheus 采集指标Grafana 做可视化。关键指标包括每个节点的 P99 延迟、A2A 消息的丢失率、MCP 工具调用的错误率。注意多智能体系统的调试比单体复杂得多。建议在开发阶段开启全链路日志记录每个节点的输入输出和 Agent 间的消息。生产环境可以采样记录避免日志量过大。5. 常见问题与排查技巧实录5.1 Agent 之间消息丢失或乱序这是 A2A 通信最常见的问题。表现是下游 Agent 收不到消息或者收到的消息顺序不对。排查思路先看消息队列的积压情况如果积压严重说明消费能力不足再看消息的 context_id 是否一致不一致说明发送方生成逻辑有问题最后看网络层是否有丢包。我的解决方案是给消息加序号和确认机制。发送方给每条消息编号接收方收到后回 ACK超时未收到 ACK 就重发。这样能保证消息不丢但要注意去重避免重发导致重复处理。5.2 MCP 工具调用超时MCP 工具调用超时通常有三个原因工具本身执行慢、网络延迟、Agent 等待策略不合理。我遇到过一次数据库查询工具超时排查发现是查询没加索引全表扫描导致。加了索引后从 8 秒降到 200 毫秒。所以工具本身的性能要先保证。网络延迟方面如果 MCP 服务端和 Agent 不在同一台机器建议加连接池和重试机制。Agent 侧的等待策略要设置合理的超时时间我一般设 30 秒超过就返回降级结果。5.3 Skill 冲突与优先级当多个 Skill 功能重叠时Agent 可能选错。比如同时有查询文档和搜索知识库两个 SkillAgent 不知道该用哪个。解决办法是给 Skill 加优先级和适用场景标签。Agent 选择时先按场景过滤再按优先级排序。另外Skill 的描述要写清楚边界避免功能重叠。我在项目里维护了一个 Skill 冲突检查清单新增 Skill 时先检查是否与已有 Skill 重叠重叠的要么合并要么明确分工。5.4 常见问题速查表问题现象可能原因排查方法解决方案Agent 不响应消息未送达检查消息队列和 ACK加重试和确认机制工具调用失败参数格式错误查看 MCP 调用日志校验输入 schema响应时间过长串行节点过多分析 DAG 执行路径并行化可并行节点结果不一致状态竞争检查共享存储读写加锁或版本号Skill 选错描述不清晰查看 Agent 决策日志优化 Skill 描述和优先级模型成本过高大模型调用过多统计各 Agent 模型用量分级用模型加缓存5.5 几个踩过的坑第一个坑是过度设计。我一开始给每个 Agent 都配了完整的 MCP 和 A2A 能力结果系统复杂度飙升调试困难。后来发现很多 Agent 只需要 MCP 调工具不需要 A2A砍掉后系统清爽很多。第二个坑是忽略幂等性。A2A 消息重发时如果接收方没有幂等处理会导致重复执行。比如重复扣款、重复发邮件。我的做法是给每个任务加唯一 ID接收方处理前先检查是否已处理过。第三个坑是日志太多。全链路日志在调试时很有用但生产环境日志量太大影响性能。后来改成采样记录只对错误和慢请求记录完整日志。第四个坑是Skill 版本管理。Skill 更新后正在运行的 Agent 可能还在用旧版本。我的解决方案是给 Skill 加版本号Agent 绑定具体版本更新时灰度切换。5.6 性能优化的几个实用技巧缓存是最有效的优化手段。我在 MCP 工具层加了结果缓存相同参数的调用直接返回缓存结果。对于文档检索这类读多写少的场景缓存命中率能到 40% 以上。批处理也很重要。如果多个 Agent 需要调用同一个工具可以合并成一次批量调用。比如三个 Agent 都要查用户信息合并成一次查询比三次单独查询快得多。异步化是另一个方向。A2A 通信默认是异步的但很多实现里 Agent 处理消息是同步的。改成异步后Agent 可以在等待下游响应时处理其他任务吞吐量能提升 2-3 倍。最后是模型分级。不是所有任务都需要大模型。路由、分类、简单抽取用 7B 模型就够只有生成和复杂推理才需要大模型。我在项目里做了模型分级后成本降了 60%效果几乎没影响。这套架构我用了大半年从最初的 3 个 Agent 扩展到现在的 12 个新增能力基本不用改核心代码。如果你也在做类似的项目建议先从两三个 Agent 的小系统开始跑通了再逐步扩展。一上来就搞十几个 Agent调试成本会让你怀疑人生。
1. 从单体到集群为什么我们需要超级多智能体1.1 一个真实的需求场景去年下半年我接手了一个企业内部知识助手的项目需求听起来不复杂帮员工查制度文档、走审批流程、生成周报。一开始我用的是单体 Agent 方案一个模型加一堆工具函数跑起来确实快。但上线两周后问题就来了——查文档的 Agent 经常被审批流程的上下文污染生成周报时又因为要调用太多工具导致响应时间飙到十几秒。最要命的是我想给文档检索单独换一个更便宜的模型结果发现整个系统耦合在一起改一处动全身。这个经历让我彻底转向了多智能体架构。所谓超级多智能体不是简单地把几个 Agent 拼在一起而是要解决三个核心问题可编排谁先谁后、谁调用谁、可互通Agent 之间怎么传消息、怎么共享状态、可扩展新增一个 Agent 要不要改老代码。这三个问题对应到技术选型上就是 DeepAgents 负责编排、MCP 负责工具接入、A2A 负责 Agent 间通信、Skills 负责能力封装。如果你正在做 Agent 项目或者被单体 Agent 的耦合问题折磨过这套组合拳值得花时间研究。它不依赖某个特定厂商核心思路可以迁移到任何 Agent 框架上。1.2 四个关键词到底在说什么先把概念理清楚不然后面容易绕晕。DeepAgents在这里指的是一种深度编排的 Agent 架构模式核心思想是把复杂任务拆成有向图每个节点是一个 Agent 或一个工具调用边代表数据流和控制流。它和简单的链式调用Chain最大的区别在于支持条件分支、循环、并行和人工介入节点。MCPModel Context Protocol是一个开放协议解决的是Agent 怎么标准化地调用外部工具和数据源的问题。你可以把它理解成 Agent 世界的 USB 接口——不管你是数据库、文件系统还是第三方 API只要实现 MCP 协议任何支持 MCP 的 Agent 都能直接接入。热搜里有人问mcp 是软件协议还是硬件协议答案是软件协议它定义的是通信格式和调用规范跟硬件层面的协议比如 USB、I2C完全是两回事。A2AAgent to Agent解决的是 Agent 之间的通信问题。MCP 管的是 Agent 调工具A2A 管的是 Agent 调 Agent。它定义了 Agent 之间怎么发现对方、怎么发消息、怎么处理异步响应。热搜里出现的a2a spring说明已经有人在 Spring 生态里做 A2A 的集成了。Skills是能力的封装单元。一个 Skill 可以是一个提示词模板、一段代码逻辑、一组工具调用的组合。它的价值在于复用——你写好一个生成周报的 Skill可以在多个 Agent 里直接引用不用重复实现。热搜里claude agent skills: a first principles deep dive和codex skills都指向同一个趋势Skills 正在成为 Agent 能力复用的标准方式。1.3 这套架构适合谁不是所有项目都需要上多智能体。如果你的需求是单轮问答、简单工具调用单体 Agent 完全够用硬上多智能体只会增加复杂度。但如果你遇到以下情况这套架构就值得考虑任务流程超过 5 个步骤且步骤之间有依赖关系需要多个专业领域的 Agent 协作比如一个查数据、一个做分析、一个写报告工具数量超过 10 个单体 Agent 的提示词已经塞不下需要给不同 Agent 配不同模型来控制成本系统需要频繁新增能力不想每次改核心代码下面我会按照实际搭建的顺序从架构设计到落地实现把每个环节讲透。2. 架构设计编排层、通信层、能力层怎么分2.1 三层架构的职责边界我在实际项目中把系统分成三层每层职责清晰层与层之间通过标准协议通信。编排层DeepAgents负责决策任务怎么拆、下一步走哪个节点、什么时候结束。这一层不关心具体工具怎么调只关心流程控制。我通常用一个有向无环图DAG来描述流程节点类型包括Agent 节点、工具节点、条件判断节点、人工审核节点、并行分支节点。通信层A2A MCP负责传输Agent 之间怎么传消息用 A2AAgent 调工具用 MCP。这一层的关键是协议标准化只要协议统一底层换实现不影响上层。能力层Skills负责执行每个 Skill 是一个独立的能力单元包含提示词、工具绑定、输入输出定义。Skill 可以注册到多个 Agent 上实现复用。这样分层的最大好处是新增一个 Agent 只需要在编排层加节点、在能力层加 Skill通信层完全不用动。我试过在一个已有 8 个 Agent 的系统里新增一个合同审查Agent从写 Skill 到接入流程只花了半天。2.2 为什么选 DAG 而不是状态机编排层可以用状态机也可以用 DAG。我最终选 DAG 的原因是状态机适合线性流程但多智能体场景下经常需要并行和条件分支。比如同时查三个数据源然后汇总这种需求用状态机写起来很别扭DAG 天然支持并行节点。DAG 的另一个好处是可观测。每个节点的输入输出都能记录下来出问题时可以精确定位是哪个节点出了错。我在生产环境里给每个节点都加了日志埋点排查问题时直接看节点级别的 trace比看一整条链路的日志高效得多。2.3 通信协议选型MCP 和 A2A 的分工很多人搞不清 MCP 和 A2A 的区别我用一个类比说明MCP 像是人用工具A2A 像是人跟人说话。你用手电筒工具不需要跟手电筒商量直接按开关就行这是 MCP。但你要跟同事协作就得说话、等回复、确认理解这是 A2A。具体到实现上MCP 的调用是同步或异步的请求-响应模式Agent 发一个调用请求工具返回结果。A2A 则更复杂支持流式响应、多轮对话、任务委派。我在项目里用 MCP 接入了文件系统、数据库、搜索引擎三类工具用 A2A 实现了研究 Agent和写作 Agent之间的协作。注意MCP 和 A2A 不是互斥的。一个 Agent 可以同时是 MCP 的客户端调工具和 A2A 的服务端被其他 Agent 调用。这种双重身份在多智能体系统里很常见。2.4 Skills 的粒度怎么把握Skills 的粒度设计是个经验活。太粗复用性差太细组合起来麻烦。我的原则是一个 Skill 对应一个完整的业务动作。比如查询员工考勤是一个 Skill它内部可能调了三个 MCP 工具查数据库、格式化数据、计算统计但对外只暴露一个接口。这样其他 Agent 引用时不用关心内部实现。反过来调用数据库这种太底层的操作不适合做成 Skill它应该是 MCP 工具。Skill 的抽象层级要高于工具低于完整的 Agent。3. 核心细节MCP 工具接入与 A2A 通信实现3.1 MCP 工具接入的完整流程MCP 的核心是标准化。不管你的工具是 Python 函数、HTTP API 还是数据库查询都要包装成 MCP 协议规定的格式。我以接入一个文档检索工具为例走一遍完整流程。首先定义工具的元数据包括名称、描述、输入参数 schema。这部分用 JSON Schema 描述目的是让 Agent 知道这个工具能干什么、需要什么参数。{ name: search_documents, description: 根据关键词检索企业制度文档, inputSchema: { type: object, properties: { query: {type: string, description: 检索关键词}, top_k: {type: integer, default: 5, description: 返回结果数量} }, required: [query] } }然后实现工具的执行逻辑。这部分可以用任何语言写只要最终暴露成 MCP 服务端即可。我用 Python 写了一个简单的实现def search_documents(query: str, top_k: int 5): # 实际检索逻辑 results vector_store.search(query, top_k) return [{title: r.title, content: r.content} for r in results]最后把工具注册到 MCP 服务端Agent 通过 MCP 客户端连接后就能自动发现并调用这个工具。这里有个坑要注意工具描述的质量直接决定 Agent 会不会正确调用。我一开始写的描述很简略结果 Agent 经常在不需要检索的时候也去调这个工具。后来我把描述改得更具体加上仅在用户询问制度、流程、规定时使用这样的约束误调用率明显下降。3.2 A2A 通信的消息格式设计A2A 通信的关键是消息格式要统一。我定义了一个基础消息结构包含发送方、接收方、消息类型、负载和上下文 ID。{ from: research_agent, to: writing_agent, type: task_delegation, context_id: task_20240115_001, payload: { task: 根据以下研究结果撰写报告, data: {...}, requirements: 字数 800 字包含数据引用 } }context_id很重要它让多个 Agent 之间的多轮交互能关联起来。比如研究 Agent 发了任务后写作 Agent 可能中途回来问问题这时候靠 context_id 就能找到原始任务上下文。A2A 的响应支持流式和非流式两种。流式适合长任务写作 Agent 可以边写边把内容推给调用方非流式适合短任务一次性返回结果。我在项目里默认用流式因为用户体验更好而且能提前发现生成方向跑偏的问题。3.3 Skills 的注册与发现机制Skills 要能被复用就需要一个注册中心。我实现了一个简单的 Skill Registry每个 Skill 注册时提供名称、描述、输入输出 schema 和执行入口。class SkillRegistry: def __init__(self): self.skills {} def register(self, name, description, schema, handler): self.skills[name] { description: description, schema: schema, handler: handler } def get(self, name): return self.skills.get(name) def list_all(self): return list(self.skills.keys())Agent 在初始化时从 Registry 拉取可用的 Skill 列表根据任务需要动态绑定。这样新增 Skill 不用改 Agent 代码只要注册进去就行。热搜里提到的find skills和skills 推荐反映了大家对这个机制的关注。我的经验是Skill 的发现不能只靠名称匹配还要有语义检索。我后来加了一个基于向量相似度的 Skill 检索Agent 用自然语言描述需求就能找到合适的 Skill比硬编码名称灵活得多。3.4 状态管理与上下文传递多智能体系统里状态管理是最容易出问题的地方。我的做法是全局状态用共享存储局部状态用消息传递。全局状态包括任务 ID、用户信息、会话历史存在 Redis 里所有 Agent 都能读写。局部状态包括当前节点的中间结果通过 A2A 消息在 Agent 之间传递不落盘。这样设计的好处是避免了状态不一致。我踩过的坑是早期把所有状态都放在消息里传结果一个 Agent 改了状态但没传给下一个导致数据丢失。后来把关键状态抽到共享存储问题就解决了。提示共享存储要加版本号或乐观锁防止多个 Agent 并发写同一份数据时互相覆盖。4. 实操落地从零搭建一个多智能体系统4.1 环境准备与依赖安装我以 Python 技术栈为例列出核心依赖。如果你用其他语言思路是一样的只是库不同。pip install fastapi uvicorn redis pydantic pip install mcp-client mcp-server pip install a2a-sdk pip install langgraph # 用于 DAG 编排MCP 和 A2A 的 SDK 目前还在快速迭代建议锁定版本号避免升级导致接口不兼容。我在项目里用 requirements.txt 固定了版本每次升级前先在测试环境验证。Redis 用于共享状态存储FastAPI 用于暴露 HTTP 接口Pydantic 用于数据校验。LangGraph 是我常用的 DAG 编排库它原生支持条件分支和循环比手写状态机省事。4.2 定义第一个 Agent 和 Skill先定义一个最简单的文档检索 Agent它绑定一个检索文档的 Skill。from pydantic import BaseModel class SearchInput(BaseModel): query: str top_k: int 5 class SearchOutput(BaseModel): results: list total: int def search_handler(input: SearchInput) - SearchOutput: results vector_store.search(input.query, input.top_k) return SearchOutput(resultsresults, totallen(results)) # 注册 Skill registry.register( namesearch_documents, description检索企业制度文档适用于查询规定、流程、政策, schema{input: SearchInput, output: SearchOutput}, handlersearch_handler )然后定义 Agent它从 Registry 获取 Skill通过 MCP 调用工具通过 A2A 接收任务。class DocumentAgent: def __init__(self, registry, mcp_client, a2a_client): self.registry registry self.mcp mcp_client self.a2a a2a_client self.skill registry.get(search_documents) async def handle(self, message): input_data SearchInput(**message.payload) result self.skill[handler](input_data) return {status: success, data: result.dict()}这个 Agent 的结构很清晰接收消息、解析输入、执行 Skill、返回结果。新增能力只需要注册新 Skill 并绑定到 Agent 上。4.3 编排多个 Agent 的 DAG有了单个 Agent 后用 LangGraph 把它们编排成 DAG。我以一个员工咨询场景为例用户提问后先判断问题类型如果是制度类问题走文档 Agent如果是数据类问题走数据 Agent最后统一由汇总 Agent 生成回答。from langgraph.graph import StateGraph, END def route_question(state): if 制度 in state[question] or 流程 in state[question]: return document_agent elif 数据 in state[question] or 统计 in state[question]: return data_agent else: return general_agent graph StateGraph(AgentState) graph.add_node(router, route_question) graph.add_node(document_agent, document_node) graph.add_node(data_agent, data_node) graph.add_node(general_agent, general_node) graph.add_node(summarizer, summarize_node) graph.set_entry_point(router) graph.add_conditional_edges(router, route_question, { document_agent: document_agent, data_agent: data_agent, general_agent: general_agent }) graph.add_edge(document_agent, summarizer) graph.add_edge(data_agent, summarizer) graph.add_edge(general_agent, summarizer) graph.add_edge(summarizer, END) app graph.compile()这个 DAG 的关键点是路由节点。路由可以用规则实现也可以用一个小模型做意图分类。我在项目里用的是规则加模型兜底常见问题走规则规则匹配不到时调模型判断兼顾速度和准确率。4.4 参数计算与性能调优多智能体系统的性能瓶颈通常在两个地方Agent 间的通信延迟和模型调用次数。我做过一组实测数据供参考。场景Agent 数量平均响应时间模型调用次数单体 Agent12.1s1-2双 Agent 串行23.8s2-3三 Agent 并行34.2s3-4五 Agent 混合56.5s5-7从数据看Agent 数量增加会带来明显的延迟增长。优化方向有三个一是并行化能并行的节点不要串行二是缓存相同输入的结果缓存起来三是模型分级简单任务用便宜的小模型。我在项目里给每个 Agent 配了不同的模型路由用 7B 小模型文档检索用 13B 模型汇总生成用 70B 大模型。这样整体成本降了约 60%响应时间只增加了 15%。4.5 部署与监控部署上我推荐容器化每个 Agent 一个容器通过消息队列或 HTTP 通信。这样单个 Agent 出问题不影响整体也方便水平扩展。监控要覆盖三个层面节点级别的执行时间、Agent 级别的调用成功率、系统级别的吞吐量。我用 Prometheus 采集指标Grafana 做可视化。关键指标包括每个节点的 P99 延迟、A2A 消息的丢失率、MCP 工具调用的错误率。注意多智能体系统的调试比单体复杂得多。建议在开发阶段开启全链路日志记录每个节点的输入输出和 Agent 间的消息。生产环境可以采样记录避免日志量过大。5. 常见问题与排查技巧实录5.1 Agent 之间消息丢失或乱序这是 A2A 通信最常见的问题。表现是下游 Agent 收不到消息或者收到的消息顺序不对。排查思路先看消息队列的积压情况如果积压严重说明消费能力不足再看消息的 context_id 是否一致不一致说明发送方生成逻辑有问题最后看网络层是否有丢包。我的解决方案是给消息加序号和确认机制。发送方给每条消息编号接收方收到后回 ACK超时未收到 ACK 就重发。这样能保证消息不丢但要注意去重避免重发导致重复处理。5.2 MCP 工具调用超时MCP 工具调用超时通常有三个原因工具本身执行慢、网络延迟、Agent 等待策略不合理。我遇到过一次数据库查询工具超时排查发现是查询没加索引全表扫描导致。加了索引后从 8 秒降到 200 毫秒。所以工具本身的性能要先保证。网络延迟方面如果 MCP 服务端和 Agent 不在同一台机器建议加连接池和重试机制。Agent 侧的等待策略要设置合理的超时时间我一般设 30 秒超过就返回降级结果。5.3 Skill 冲突与优先级当多个 Skill 功能重叠时Agent 可能选错。比如同时有查询文档和搜索知识库两个 SkillAgent 不知道该用哪个。解决办法是给 Skill 加优先级和适用场景标签。Agent 选择时先按场景过滤再按优先级排序。另外Skill 的描述要写清楚边界避免功能重叠。我在项目里维护了一个 Skill 冲突检查清单新增 Skill 时先检查是否与已有 Skill 重叠重叠的要么合并要么明确分工。5.4 常见问题速查表问题现象可能原因排查方法解决方案Agent 不响应消息未送达检查消息队列和 ACK加重试和确认机制工具调用失败参数格式错误查看 MCP 调用日志校验输入 schema响应时间过长串行节点过多分析 DAG 执行路径并行化可并行节点结果不一致状态竞争检查共享存储读写加锁或版本号Skill 选错描述不清晰查看 Agent 决策日志优化 Skill 描述和优先级模型成本过高大模型调用过多统计各 Agent 模型用量分级用模型加缓存5.5 几个踩过的坑第一个坑是过度设计。我一开始给每个 Agent 都配了完整的 MCP 和 A2A 能力结果系统复杂度飙升调试困难。后来发现很多 Agent 只需要 MCP 调工具不需要 A2A砍掉后系统清爽很多。第二个坑是忽略幂等性。A2A 消息重发时如果接收方没有幂等处理会导致重复执行。比如重复扣款、重复发邮件。我的做法是给每个任务加唯一 ID接收方处理前先检查是否已处理过。第三个坑是日志太多。全链路日志在调试时很有用但生产环境日志量太大影响性能。后来改成采样记录只对错误和慢请求记录完整日志。第四个坑是Skill 版本管理。Skill 更新后正在运行的 Agent 可能还在用旧版本。我的解决方案是给 Skill 加版本号Agent 绑定具体版本更新时灰度切换。5.6 性能优化的几个实用技巧缓存是最有效的优化手段。我在 MCP 工具层加了结果缓存相同参数的调用直接返回缓存结果。对于文档检索这类读多写少的场景缓存命中率能到 40% 以上。批处理也很重要。如果多个 Agent 需要调用同一个工具可以合并成一次批量调用。比如三个 Agent 都要查用户信息合并成一次查询比三次单独查询快得多。异步化是另一个方向。A2A 通信默认是异步的但很多实现里 Agent 处理消息是同步的。改成异步后Agent 可以在等待下游响应时处理其他任务吞吐量能提升 2-3 倍。最后是模型分级。不是所有任务都需要大模型。路由、分类、简单抽取用 7B 模型就够只有生成和复杂推理才需要大模型。我在项目里做了模型分级后成本降了 60%效果几乎没影响。这套架构我用了大半年从最初的 3 个 Agent 扩展到现在的 12 个新增能力基本不用改核心代码。如果你也在做类似的项目建议先从两三个 Agent 的小系统开始跑通了再逐步扩展。一上来就搞十几个 Agent调试成本会让你怀疑人生。
：先进工艺芯片可靠性signoff的破局之道)
