1. 从单体到集群为什么我们需要重新思考 Agent 的架构过去一年里我几乎把市面上能叫得出名字的 Agent 框架都跑了一遍。从最早的 ReAct 循环到后来的 Plan-and-Execute再到各种带记忆、带工具调用的编排方案踩过的坑能写满一个笔记本。但真正让我意识到“单体 Agent 已经走到瓶颈”的是去年底做的一个企业知识库问答项目——单个 Agent 挂载了十几个工具上下文窗口被工具描述和中间结果塞得满满当当稍微复杂一点的多跳查询就开始胡言乱语工具调用成功率从 90% 一路掉到 60% 以下。这不是模型能力的问题而是架构的问题。一个 Agent 既要做意图识别又要做任务规划还要负责工具选择、参数填充、结果校验、异常重试最后还要组织语言输出——这就像让一个人同时当项目经理、程序员、测试和客服短期能扛长期必崩。DeepAgents MCP A2A Skills 这套组合本质上是在回答一个问题如何把“一个全能 Agent”拆成“一群专业 Agent”并且让它们像一支训练有素的团队一样协作。这里面四个关键词各司其职DeepAgents 负责深度任务分解与编排MCP 解决 Agent 与外部工具/数据源的标准化连接A2A 打通 Agent 与 Agent 之间的通信协议Skills 则是把可复用的能力封装成即插即用的模块。这套架构适合谁如果你正在做以下任何一件事它都值得你花时间研究需要处理多步骤复杂业务流程的自动化系统、需要接入大量异构工具的企业级 Agent 平台、需要多个专业 Agent 协同完成任务的场景比如代码生成 代码审查 测试用例生成、以及希望把 Agent 能力模块化复用的团队。我接下来会按照“设计思路 → 核心组件拆解 → 实操搭建 → 问题排查”的顺序把这套架构从里到外讲清楚。每个部分都会附带我在实际项目中验证过的配置和参数能直接抄作业的地方我会明确标出来。2. 整体架构设计四层解耦的 Agent 集群2.1 为什么是“四层”而不是“三层”或“五层”在动手之前先想清楚分层逻辑。我见过不少团队一上来就把所有东西塞进一个编排层结果就是改一处崩三处。这套架构我最终定型为四层每一层有明确的职责边界层级组件核心职责不负责什么编排层DeepAgents任务分解、Agent 调度、状态管理不直接调用外部工具通信层A2AAgent 间消息传递、任务委派、结果回传不关心任务内容能力层Skills封装可复用能力单元不处理跨 Agent 协调连接层MCP标准化接入外部工具和数据源不做任务规划这个分法的核心原则是单一职责。DeepAgents 只管“谁来做、按什么顺序做”A2A 只管“怎么把消息传过去”Skills 只管“这个能力怎么实现”MCP 只管“怎么连上外部系统”。任何一层的变化不会波及其他层。举个实际例子我们有个客户需要 Agent 集群处理合同审核流程。编排层定义“先提取条款 → 再比对模板 → 再生成风险报告 → 最后人工复核”这个流程通信层负责把“提取条款”的任务发给文档处理 Agent能力层的 Skills 里有一个“PDF 条款抽取”技能连接层通过 MCP 连上企业的文档管理系统。后来客户换了文档管理系统只需要改 MCP 的连接配置上面三层完全不动。2.2 DeepAgents 的编排哲学不是工作流是任务图很多人第一反应是把 DeepAgents 当成工作流引擎来用定义好节点和边然后按顺序执行。这么用不是不行但浪费了它最大的价值——动态任务分解。传统工作流是“你告诉它每一步做什么”DeepAgents 是“你告诉它目标是什么它自己决定分几步、每步谁来做”。这背后的机制是编排 Agent 拿到高层目标后先做一次任务分解生成一个任务图Task Graph每个节点是一个子任务边是依赖关系。然后它根据每个子任务的性质决定是分配给某个专业 Agent还是自己直接处理。我实测下来任务分解的质量高度依赖两个东西分解提示词的设计和可用 Agent/Skill 的描述质量。前者决定了它能不能把任务拆得粒度合适后者决定了它能不能把子任务匹配到正确的执行者。提示任务分解的粒度控制在 3-7 个子任务比较合适。太少说明没拆开太多说明拆过头了协调成本会超过收益。2.3 A2A 协议Agent 之间的“普通话”A2AAgent-to-Agent要解决的核心问题是当 Agent A 需要 Agent B 帮忙做一件事时它们怎么对话没有标准协议的时候每个团队自己定一套 JSON 格式结果就是 Agent 之间没法互通每接一个新 Agent 就要写一个适配器。A2A 定义了一套标准的消息格式和交互模式核心包括Agent Card每个 Agent 对外暴露自己的能力描述包括名称、功能、输入输出格式、调用示例。这相当于 Agent 的“名片”。Task 消息任务委派的标准格式包含任务 ID、目标描述、输入参数、期望输出格式、超时设置。Artifact 消息任务结果的标准格式包含任务 ID、状态、输出内容、错误信息。流式更新长任务支持中间状态推送避免调用方干等。这套协议最大的价值在于解耦。Agent B 的内部实现怎么改都行只要它的 Agent Card 不变Agent A 就不需要动。我们有个项目里文档处理 Agent 从基于规则的方法换成了微调模型但因为 Agent Card 的输入输出格式没变编排层和其他 Agent 完全无感知。2.4 Skills 与 MCP 的分工能力封装 vs 连接标准化这两个概念经常被混淆我用一个类比说清楚Skills 是菜谱MCP 是厨房设备的统一接口。Skills 封装的是“怎么做一件事”的知识和步骤。比如“生成周报”这个 Skill里面包含了数据从哪里取、怎么汇总、用什么格式输出、有哪些注意事项。它是一个可复用的能力单元可以在不同 Agent 之间共享。MCP 解决的是“怎么连上外部系统”的问题。数据库、API、文件系统、消息队列每种系统的连接方式都不一样。MCP 提供了一套标准的连接器协议让 Agent 不需要为每个外部系统写一套适配代码。两者的关系是一个 Skill 在执行过程中可能需要通过多个 MCP 连接器来获取数据和执行操作。比如“生成周报”这个 Skill可能通过 MCP 连上数据库取数据通过 MCP 连上邮件系统发报告。3. 核心组件深度拆解与配置要点3.1 DeepAgents 编排器任务分解的提示词工程编排器的核心是一个任务分解提示词这个提示词的质量直接决定了整个集群的效率。我经过十几轮迭代最终定型的模板结构如下ORCHESTRATOR_PROMPT 你是一个任务编排器。你的职责是将用户的高层目标分解为可执行的子任务并分配给合适的执行者。 ## 可用执行者 {agent_cards} ## 可用技能 {skill_descriptions} ## 分解规则 1. 每个子任务必须明确目标、输入、期望输出、依赖关系 2. 子任务粒度单个子任务应能在 1-3 步内完成 3. 优先使用已有技能避免重复造轮子 4. 如果某个子任务没有合适的执行者标记为 NEEDS_HUMAN ## 输出格式 以 JSON 格式输出任务图 { tasks: [ { id: task_1, goal: 任务目标描述, assignee: agent_name 或 skill_name, inputs: {}, expected_output: 输出格式描述, depends_on: [] } ] } 这个模板里有几个关键设计点值得展开说第一把 Agent Card 和 Skill 描述动态注入提示词。这意味着当你新增一个 Agent 或 Skill 时编排器会自动“知道”它的存在不需要改提示词。这是可扩展性的关键。第二强制要求输出依赖关系。没有依赖关系的任务图就是一堆散点编排器无法判断执行顺序。我试过让模型自己推断依赖结果经常出现循环依赖或者遗漏关键依赖后来改成强制显式声明。第三NEEDS_HUMAN 标记。这是实际项目中非常重要的一个设计。不是所有任务都能自动化当编排器发现某个子任务没有合适的执行者时标记出来让人工介入比强行分配一个不合适的 Agent 要好得多。3.2 A2A 通信层Agent Card 的设计规范Agent Card 是 A2A 协议的核心它决定了其他 Agent 能否正确调用这个 Agent。我总结了一个好用的 Agent Card 应该包含的字段{ name: document_processor, version: 1.2.0, description: 处理各类文档的提取、转换和格式化任务, capabilities: [ { name: extract_clauses, description: 从合同中提取关键条款, input_schema: { type: object, properties: { document_url: {type: string}, clause_types: {type: array, items: {type: string}} }, required: [document_url] }, output_schema: { type: object, properties: { clauses: {type: array}, confidence: {type: number} } }, examples: [ { input: {document_url: s3://bucket/contract.pdf, clause_types: [payment, termination]}, output: {clauses: [...], confidence: 0.95} } ] } ], endpoint: http://agent-service:8080/a2a, health_check: http://agent-service:8080/health }这里面的examples字段经常被忽略但它极其重要。编排器在决定把任务分配给谁的时候示例的质量直接影响匹配准确率。我实测过加上高质量示例后任务分配准确率从 72% 提升到了 91%。3.3 Skills 封装从“能跑”到“可复用”的关键步骤把一个能力封装成 Skill不是简单地把代码包一层就完事了。一个合格的 Skill 需要满足输入输出明确、错误处理完善、有使用文档、有测试用例、版本可管理。我以“数据汇总”这个 Skill 为例展示完整的封装结构class DataAggregationSkill: 数据汇总技能从多个数据源获取数据并生成汇总报告 name data_aggregation version 1.0.0 description 从多个数据源获取数据按指定维度汇总输出结构化报告 input_schema { sources: {type: array, description: 数据源列表每个包含 type 和 connection_info}, dimensions: {type: array, description: 汇总维度}, metrics: {type: array, description: 汇总指标}, time_range: {type: object, description: 时间范围} } output_schema { report: {type: object, description: 汇总报告}, data_quality: {type: object, description: 数据质量指标} } async def execute(self, inputs, context): # 1. 参数校验 self._validate_inputs(inputs) # 2. 通过 MCP 连接器获取数据 raw_data [] for source in inputs[sources]: connector context.mcp.get_connector(source[type]) data await connector.fetch(source[connection_info], inputs[time_range]) raw_data.append(data) # 3. 数据清洗和汇总 cleaned self._clean(raw_data) aggregated self._aggregate(cleaned, inputs[dimensions], inputs[metrics]) # 4. 质量检查 quality self._check_quality(aggregated) return {report: aggregated, data_quality: quality} def _validate_inputs(self, inputs): # 详细的参数校验逻辑 pass这个结构里有几个设计决策值得说明为什么输入输出用 JSON Schema 而不是 Python 类型注解因为 Skill 可能被不同语言的 Agent 调用JSON Schema 是跨语言的通用描述。而且编排器需要读取这个 Schema 来做参数填充JSON Schema 更容易解析。为什么 execute 方法接收 context 参数context 里包含了 MCP 连接器、日志器、配置等运行时依赖。这样 Skill 本身不需要关心这些依赖怎么初始化只需要从 context 里取。为什么要有版本号当 Skill 的行为发生变化时版本号让调用方可以明确知道自己依赖的是哪个版本。我们有过一次教训一个 Skill 的输出格式悄悄改了导致下游三个 Agent 全部出错排查了半天才发现是版本问题。3.4 MCP 连接器标准化接入的实操细节MCP 连接器的核心是定义一套标准的接口让不同的外部系统都能以统一的方式被访问。一个 MCP 连接器需要实现以下方法class MCPConnector: MCP 连接器基类 async def connect(self, connection_info): 建立连接 raise NotImplementedError async def fetch(self, query, paramsNone): 获取数据 raise NotImplementedError async def execute(self, operation, paramsNone): 执行操作 raise NotImplementedError async def health_check(self): 健康检查 raise NotImplementedError async def close(self): 关闭连接 raise NotImplementedError以数据库连接器为例实际实现时需要注意几个点连接池管理。不要每次请求都新建连接用连接池复用。我一般设置最小连接数 2、最大连接数 10具体根据并发量调整。超时设置。数据库查询必须有超时否则一个慢查询会拖垮整个 Agent 集群。我通常设置查询超时 30 秒连接超时 5 秒。错误分类。把错误分成可重试如连接超时和不可重试如语法错误两类可重试的自动重试 2-3 次不可重试的直接返回错误。结果集大小限制。默认限制返回 1000 行超过的部分要么分页要么截断。这是防止一个查询把内存打爆。4. 从零搭建一个可运行的多智能体集群4.1 环境准备与依赖安装先把基础环境搭起来。我用的技术栈是 Python 3.11 FastAPI Redis做状态存储 PostgreSQL做持久化。以下是核心依赖pip install fastapi uvicorn redis asyncpg pydantic httpx pip install deepagents # 编排框架 pip install a2a-sdk # A2A 协议实现 pip install mcp-client # MCP 连接器Redis 用来存任务状态和 Agent 注册信息PostgreSQL 用来存任务执行历史和审计日志。这两个不是必须的但生产环境强烈建议加上。4.2 定义第一个 Agent 和它的 Agent Card我们从最简单的开始一个“文本摘要 Agent”。先定义它的 Agent Cardfrom a2a_sdk import AgentCard, Capability summarizer_card AgentCard( nametext_summarizer, version1.0.0, description对长文本进行摘要支持指定摘要长度和风格, capabilities[ Capability( namesummarize, description生成文本摘要, input_schema{ type: object, properties: { text: {type: string, description: 待摘要的文本}, max_length: {type: integer, default: 200}, style: {type: string, enum: [concise, detailed, bullet], default: concise} }, required: [text] }, output_schema{ type: object, properties: { summary: {type: string}, original_length: {type: integer}, summary_length: {type: integer} } } ) ], endpointhttp://localhost:8001/a2a )然后实现这个 Agent 的服务端from fastapi import FastAPI from a2a_sdk import A2AServer, Task, Artifact app FastAPI() server A2AServer(cardsummarizer_card) server.handler(summarize) async def handle_summarize(task: Task) - Artifact: text task.inputs[text] max_length task.inputs.get(max_length, 200) style task.inputs.get(style, concise) # 调用 LLM 生成摘要 summary await llm_summarize(text, max_length, style) return Artifact( task_idtask.id, statuscompleted, output{ summary: summary, original_length: len(text), summary_length: len(summary) } ) app.include_router(server.router)启动这个服务后它就注册到了 A2A 网络中其他 Agent 可以通过 Agent Card 发现它并调用。4.3 编排器接入与任务分发编排器需要知道有哪些 Agent 可用。我实现了一个简单的 Agent 注册中心class AgentRegistry: def __init__(self, redis_client): self.redis redis_client async def register(self, card: AgentCard): await self.redis.hset(agents, card.name, card.json()) async def discover(self, capability: str None) - list[AgentCard]: all_cards await self.redis.hgetall(agents) cards [AgentCard.parse_raw(v) for v in all_cards.values()] if capability: cards [c for c in cards if any(cap.name capability for cap in c.capabilities)] return cards async def get_card(self, name: str) - AgentCard: data await self.redis.hget(agents, name) return AgentCard.parse_raw(data) if data else None编排器在分解任务时从注册中心拉取所有 Agent Card注入到提示词中。任务分解完成后对于每个子任务编排器通过 A2A 协议把任务发给对应的 Agent。这里有个细节任务分发要支持并行。如果两个子任务没有依赖关系应该同时发给两个 Agent而不是串行等待。我用 asyncio.gather 实现并行分发async def dispatch_tasks(self, task_graph): # 按依赖关系分层 layers self._topological_sort(task_graph) results {} for layer in layers: # 同层任务并行执行 layer_results await asyncio.gather(*[ self._dispatch_single(task, results) for task in layer ]) for task, result in zip(layer, layer_results): results[task.id] result return results4.4 完整流程串联一个合同审核的实战案例把上面所有组件串起来跑一个完整的合同审核流程。用户输入“审核这份合同提取关键条款比对标准模板生成风险报告。”第一步编排器分解任务。编排器拿到目标后结合可用的 Agent Card 和 Skill 描述生成任务图{ tasks: [ { id: task_1, goal: 从合同文档中提取所有条款, assignee: document_processor, inputs: {document_url: s3://contracts/2024-001.pdf}, expected_output: 条款列表每条包含条款类型和内容, depends_on: [] }, { id: task_2, goal: 将提取的条款与标准模板比对, assignee: template_comparator, inputs: {clauses: $task_1.output.clauses}, expected_output: 差异列表, depends_on: [task_1] }, { id: task_3, goal: 基于差异生成风险报告, assignee: risk_analyzer, inputs: {differences: $task_2.output.differences}, expected_output: 风险报告包含风险等级和建议, depends_on: [task_2] } ] }第二步并行执行无依赖任务。task_1 没有依赖直接执行。task_2 依赖 task_1等 task_1 完成后执行。task_3 依赖 task_2最后执行。第三步结果汇总。编排器收集所有任务的输出组织成最终报告返回给用户。整个流程中每个 Agent 只负责自己擅长的部分通过 A2A 协议通信通过 MCP 连接外部系统通过 Skills 复用能力。这就是“可编排、可互通、可扩展”的实际含义。5. 常见问题与排查技巧实录5.1 任务分解粒度失控现象编排器把一个简单的“发邮件”任务拆成了“打开邮件客户端 → 填写收件人 → 填写主题 → 填写正文 → 点击发送”五个子任务。原因提示词里没有明确粒度约束模型倾向于过度分解。解决在提示词中加入明确的粒度规则并给出正反示例。我用的规则是“如果一个子任务可以由单个 Skill 在一次调用中完成就不要继续拆分。”同时加入 few-shot 示例展示合适的粒度。5.2 Agent 之间消息丢失现象Agent A 给 Agent B 发了任务但 B 一直没收到A 超时失败。排查步骤检查 B 的健康检查端点是否正常检查 A2A 消息队列是否有积压检查 B 的 Agent Card 中的 endpoint 地址是否正确检查网络策略是否允许 A 访问 B常见原因Agent Card 注册后 endpoint 变了但没更新。我遇到过好几次Agent 重启后端口变了但注册中心里还是旧地址。解决方案是 Agent 启动时强制重新注册并加上心跳机制定期刷新。5.3 Skill 版本冲突现象同一个 Skill 被两个 Agent 依赖但两个 Agent 需要不同版本的行为。解决Skill 注册时带上版本号Agent 调用时指定版本。注册中心支持同一个 Skill 的多个版本共存。调用方不指定版本时默认使用最新版本。问题类型典型现象排查方向解决方案任务分解过细子任务数量异常多检查提示词粒度规则加入粒度约束和示例消息丢失任务超时无响应检查健康检查和 endpoint强制重新注册 心跳版本冲突行为不一致检查 Skill 版本版本号 显式指定上下文溢出编排器输出截断检查注入的 Agent Card 数量按需注入 摘要压缩循环依赖任务图无法执行检查 depends_on拓扑排序 循环检测5.4 编排器上下文溢出现象当可用 Agent 和 Skill 数量超过 20 个时编排器的提示词变得极长导致输出质量下降甚至截断。解决不要把所有 Agent Card 都注入提示词。先做一个粗筛根据任务关键词匹配相关的 Agent只注入匹配到的。我实现了一个简单的关键词匹配 向量相似度混合检索把注入的 Agent 数量控制在 5-8 个。提示Agent Card 的 description 字段要写得精准这是检索匹配的主要依据。不要写“处理文档”要写“从 PDF/Word 合同中提取条款、甲乙方信息、金额、期限”。5.5 实操心得三个让我少走弯路的经验第一先跑通两个 Agent 的协作再扩展到集群。我一开始就想搭一个五 Agent 的系统结果调试了两周都没跑通。后来退回到两个 Agent一天就通了然后再逐个增加。复杂度是线性增加的但调试难度是指数增加的。第二给每个 Agent 加详细的日志。多 Agent 系统出问题时最难的是定位是哪个环节出的错。我在每个 Agent 的输入输出、每次 A2A 消息、每次 MCP 调用都加了结构化日志排查效率提升了至少三倍。第三Skill 的测试用例要覆盖边界情况。空输入、超长输入、格式错误的输入、并发调用——这些在单体 Agent 里可能不会遇到但在集群里因为调用方多样出现概率大大增加。我现在的习惯是每个 Skill 至少写 10 个测试用例其中 5 个是边界情况。6. 扩展方向这套架构还能怎么用这套架构的扩展性体现在两个维度横向扩展是增加更多专业 Agent纵向扩展是让每个 Agent 内部也采用类似的架构。横向扩展的例子在合同审核流程中可以增加一个“法规合规检查 Agent”专门负责比对最新法规增加一个“历史案例检索 Agent”从过往案例中找相似情况。这些 Agent 只需要实现自己的 Agent Card 和业务逻辑注册到中心后编排器自动发现。纵向扩展的例子文档处理 Agent 内部也可以拆成“格式识别 → 内容提取 → 结构化输出”三个子 Agent通过内部的 A2A 通信协作。这样整个系统就变成了一个多层级的 Agent 集群。我目前正在探索的一个方向是动态 Agent 生成当编排器发现某个子任务没有合适的执行者时不是标记 NEEDS_HUMAN而是根据任务描述自动生成一个临时的 Agent Card 和对应的 Skill 实现。这个方向还在实验阶段但初步结果挺有意思的。最后分享一个我在实际项目中验证过的参数配置可以直接抄编排器超时设 120 秒单个 Agent 任务超时设 60 秒A2A 消息重试 3 次MCP 连接池大小 10任务图最大深度 5 层。这套参数在日均 10 万次调用的场景下跑得很稳。