DeepAgents多Agent集群架构：MCP、A2A与Skills协同实战

1. 从单体到集群为什么我们需要重新思考 Agent 的架构过去一年我一直在折腾各种 Agent 框架从最早的 ReAct 循环手搓到后来用 LangChain 的 AgentExecutor再到各种 AutoGPT 式的自主循环。说实话大部分项目做到最后都会撞上同一堵墙单个 Agent 的能力边界太明显了。你给它挂十个工具它就开始犯迷糊你让它处理一个跨领域的任务它要么在某个环节卡死要么把上下文窗口撑爆。这不是模型不够聪明的问题而是架构本身就不支持复杂任务的分解与协作。DeepAgents 这个方向之所以值得认真聊是因为它把问题从“怎么让一个 Agent 更强”换成了“怎么让一群 Agent 各司其职还能互相通气”。这个思路转变很关键。就像一家公司你不会指望一个员工既做财务又做市场还兼着写代码而是设立不同部门定义清楚接口和协作流程。DeepAgents 提供的是一套构建这种“Agent 公司”的脚手架而 MCP、A2A、Skills 这三个东西分别解决的是工具接入、Agent 间通信、能力封装三个层面的问题。我先把这四个概念的关系理清楚不然后面容易绕晕。DeepAgents是编排层负责定义主 Agent 和子 Agent 的层级结构、任务分发逻辑、状态管理。MCPModel Context Protocol是工具层协议让 Agent 能以统一的方式调用外部工具和数据源不用为每个工具写一套适配代码。A2AAgent-to-Agent是通信层协议解决的是 Agent 之间怎么互相调用、怎么传递上下文、怎么协商任务边界。Skills是能力封装层把一组相关的工具调用、提示词模板、处理逻辑打包成一个可复用的技能单元。这四个东西凑在一起才构成一个真正可编排、可互通、可扩展的 Agent 集群。少了任何一个系统都会在某个维度上退化成“高级一点的单体 Agent”。我见过太多项目只用了其中一两个结果就是要么工具接入乱成一锅粥要么子 Agent 之间各说各话要么技能无法复用每次都要重写。这篇文章适合谁看如果你已经写过至少一个能跑的 Agent demo对 LangChain 或类似框架有基本了解现在想往多 Agent 协作方向走那这篇内容就是为你准备的。我会从架构设计讲到具体实现把每个环节的坑和技巧都摊开说。如果你是完全的新手建议先补一下 Agent 的基础概念不然有些地方可能会觉得跳跃。2. 核心架构拆解四层结构到底怎么协同2.1 DeepAgents 的编排模型主从还是对等DeepAgents 目前主流的编排模型是层级式主从结构。一个主 AgentSupervisor负责接收用户请求、拆解任务、分发给子 AgentSubagent、收集结果、做最终整合。子 Agent 各自负责一个垂直领域比如一个专门查数据库一个专门做数据分析一个专门生成报告。为什么不用对等网络结构理论上对等结构更灵活每个 Agent 都能互相调用。但实际跑下来对等结构有两个致命问题。第一是责任链不清晰出了问题你不知道该找哪个 Agent 负责调试的时候像在抓鬼。第二是上下文爆炸每个 Agent 都要维护和其他所有 Agent 的通信记录token 消耗呈指数级增长。主从结构虽然看起来“中心化”但它把复杂度控制在了可管理的范围内。主 Agent 的核心职责不是“干活”而是“分活”和“收活”。它需要具备几个关键能力任务分解把用户的一句话需求拆成可执行的子任务、路由决策判断每个子任务该交给哪个子 Agent、结果聚合把多个子 Agent 的输出整合成连贯的最终答复。这里有个经验主 Agent 的提示词要写得非常克制不要让它自己去尝试解决问题否则它会忍不住“抢活干”导致子 Agent 被架空。子 Agent 的设计原则是单一职责。一个子 Agent 只做一件事工具集不要超过五个提示词聚焦在一个领域。我试过一个子 Agent 挂八个工具结果它在选择工具时犹豫不决经常选错。后来砍到三个工具准确率立刻上来了。这个规律在多个项目里都验证过子 Agent 的工具数量和处理准确率呈明显的反比关系。2.2 MCP 协议工具接入的标准化答案MCP 解决的是一个很实际的问题你有十个工具每个工具的 API 格式、认证方式、返回结构都不一样难道每个都要写一套适配代码MCP 的思路是定义一个统一的工具描述格式和调用协议工具提供方按照 MCP 标准暴露接口Agent 侧只需要一套 MCP 客户端代码就能调用所有兼容工具。MCP 的核心概念有三个Server、Client、Transport。Server 是工具提供方它声明自己有哪些工具、每个工具接受什么参数、返回什么结构。Client 是 Agent 侧的调用方它连接 Server 并获取工具列表然后按需调用。Transport 是通信层支持 stdio本地进程通信和 HTTP/SSE远程通信两种方式。实际用下来MCP 最大的价值是工具发现。Agent 启动时连接 MCP Server自动获取可用工具列表和参数 schema不需要硬编码。这意味着你新增一个工具只要它符合 MCP 标准Agent 就能自动识别并使用不用改一行 Agent 代码。这个特性在多 Agent 系统里尤其重要因为子 Agent 的工具集经常需要动态调整。不过 MCP 也有它的局限。目前 MCP 的工具描述能力还比较基础复杂的参数校验、条件依赖、错误处理语义表达不够充分。我的做法是在 MCP Server 侧做一层封装把复杂的校验逻辑放在 Server 内部对外暴露简单的参数接口。这样 Agent 侧调用起来清爽复杂的逻辑藏在 Server 里。2.3 A2A 通信Agent 之间怎么“说人话”A2A 要解决的是 Agent 之间的通信问题。听起来简单做起来坑很多。第一个问题是消息格式。两个 Agent 之间传什么纯文本结构化 JSON还是带元数据的消息对象我的经验是任务分发用结构化格式结果回传用自然语言加结构化摘要。主 Agent 给子 Agent 发任务时用 JSON 明确指定任务类型、输入参数、期望输出格式。子 Agent 回传结果时用自然语言描述结论附带一个结构化的数据块供程序解析。第二个问题是上下文传递。子 Agent 需不需要知道完整的对话历史大部分情况下不需要。把完整历史传给每个子 Agent 会导致 token 浪费和注意力分散。我的做法是主 Agent 在分发任务时只传递与当前子任务相关的上下文片段而不是整个对话历史。这需要主 Agent 具备一定的上下文筛选能力提示词里要明确要求它“只传递必要信息”。第三个问题是错误处理。子 Agent 执行失败怎么办重试换一个子 Agent还是把错误抛回给主 Agent我的方案是分级处理可重试的错误如网络超时由子 Agent 内部重试不可重试的错误如参数错误返回给主 Agent由主 Agent 决定是修正参数后重发还是告知用户。这里的关键是错误信息要足够具体不能只返回“执行失败”要说明失败原因和建议的修正方向。2.4 Skills 封装把能力变成可复用的积木Skills 是我认为最被低估的一环。很多人把 Skills 简单理解为“提示词模板”其实它的内涵要丰富得多。一个完整的 Skill 应该包含触发条件什么情况下使用这个技能、工具组合需要调用哪些工具、按什么顺序、处理逻辑中间结果怎么转换、异常怎么处理、输出格式最终产出长什么样。举个例子一个“数据分析”Skill 可能包含先用 MCP 调用数据库查询工具获取原始数据然后用代码执行工具做统计计算最后用图表生成工具产出可视化结果。这三个步骤的编排逻辑、参数传递、错误处理都封装在 Skill 内部主 Agent 只需要说“对这个数据集做分析”Skill 就能自动完成整个流程。Skills 的复用价值在于跨 Agent 共享。同一个“数据分析”Skill可以被财务子 Agent 用来分析报表也可以被运营子 Agent 用来分析用户行为。Skill 定义一次多处使用修改时也只改一个地方。这比每个子 Agent 各自实现一套分析逻辑要高效得多。3. 实操落地从零搭建一个多 Agent 集群3.1 环境准备与依赖安装先把基础环境搭起来。我用的 Python 3.11低于 3.10 的版本有些异步特性支持不好。虚拟环境用 venv 就行没必要上 conda除非你有科学计算的重度需求。python -m venv agent-env source agent-env/bin/activate # Windows 用 agent-env\Scripts\activate pip install deepagents langchain langchain-openai mcpDeepAgents 目前还在快速迭代建议锁定版本号不然今天跑通的代码明天可能就报错。我用的组合是deepagents0.0.8配langchain0.3.x这个组合在我这边跑了两周没出过兼容性问题。MCP 的 Python SDK 是mcp包安装后可以用它来写 MCP Server 和 Client。如果你要连接已有的 MCP Server比如一些社区提供的工具服务只需要 Client 部分就行。环境变量方面至少需要配置模型 API 的 key。我习惯用.env文件管理配合python-dotenv加载。不要把 key 硬编码在代码里这个习惯一定要养成不然代码分享出去就是事故。3.2 定义第一个 MCP Server把工具标准化先写一个最简单的 MCP Server暴露两个工具一个查天气一个算数学。虽然简单但能把 MCP 的完整流程跑通。from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent app Server(demo-tools) app.list_tools() async def list_tools(): return [ Tool( nameget_weather, description查询指定城市的天气, inputSchema{ type: object, properties: { city: {type: string, description: 城市名称} }, required: [city] } ), Tool( namecalculate, description执行数学计算, inputSchema{ type: object, properties: { expression: {type: string, description: 数学表达式} }, required: [expression] } ) ] app.call_tool() async def call_tool(name: str, arguments: dict): if name get_weather: city arguments[city] # 实际项目中这里调用真实天气 API return [TextContent(typetext, textf{city}今天晴25度)] elif name calculate: expr arguments[expression] result eval(expr) # 生产环境不要用 eval这里仅演示 return [TextContent(typetext, textstr(result))] async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ __main__: import asyncio asyncio.run(main())这个 Server 用 stdio 方式通信Agent 侧启动一个子进程来连接它。list_tools返回工具列表和参数 schemacall_tool根据工具名分发到具体实现。注意inputSchema用的是 JSON Schema 格式这是 MCP 的标准Agent 侧会根据这个 schema 来生成工具调用的参数。注意eval在生产环境绝对不能用这里只是为了演示方便。实际项目中数学计算要用安全的表达式解析库比如simpleeval或asteval。3.3 构建主 Agent 与子 Agent 的层级结构接下来定义 Agent 结构。主 Agent 负责路由两个子 Agent 分别处理天气查询和数学计算。from deepagents import create_deep_agent from langchain_openai import ChatOpenAI from langchain_mcp_adapters.client import MultiServerMCPClient llm ChatOpenAI(modelgpt-4o, temperature0) # 连接 MCP Server 获取工具 async def get_tools(): client MultiServerMCPClient({ demo: { command: python, args: [mcp_server.py], transport: stdio } }) return await client.get_tools() # 子 Agent 定义 weather_agent create_deep_agent( nameweather_expert, llmllm, tools[weather_tool], # 从 MCP 获取的天气工具 system_prompt你只负责回答天气相关问题其他问题一律回复请咨询对应专家 ) math_agent create_deep_agent( namemath_expert, llmllm, tools[calculate_tool], system_prompt你只负责数学计算不处理其他类型的问题 ) # 主 Agent supervisor create_deep_agent( namesupervisor, llmllm, subagents[weather_agent, math_agent], system_prompt你是一个任务调度员。分析用户请求判断需要哪个子 Agent 处理。 天气问题交给 weather_expert数学问题交给 math_expert。 你不需要自己回答问题只需要分发任务并整合结果。 )这里的关键是主 Agent 的system_prompt。我反复强调过主 Agent 要“克制”不要让它自己动手。提示词里明确写“你不需要自己回答问题”能有效减少它抢活干的情况。子 Agent 的提示词则要强调“只负责”防止它越界处理不属于自己的任务。3.4 用 A2A 协议打通 Agent 间通信DeepAgents 内部已经封装了 Agent 间的通信机制但如果你要自定义通信逻辑或者连接外部的 A2A 兼容 Agent就需要手动处理消息格式。下面是一个简化的 A2A 消息结构from pydantic import BaseModel from typing import Optional, Any class A2AMessage(BaseModel): sender: str receiver: str task_type: str payload: dict context: Optional[dict] None reply_to: Optional[str] None class A2AResponse(BaseModel): status: str # success | error | partial result: Any error_message: Optional[str] None suggested_fix: Optional[str] None发送任务时主 Agent 构造A2AMessage指定task_type和payload。子 Agent 处理后返回A2AResponse如果出错suggested_fix字段给出修正建议。这个结构看起来简单但实际用起来能省很多事。特别是suggested_fix让主 Agent 知道下一步该怎么调整而不是盲目重试。实操心得context字段只传必要信息。我一开始把完整对话历史塞进去结果子 Agent 的 token 消耗是现在的三倍而且经常被无关信息干扰。后来改成只传当前任务相关的片段效果立竿见影。3.5 Skills 的封装与注册Skill 的封装我用一个类来实现包含元数据、工具列表和执行逻辑。class Skill: def __init__(self, name, description, tools, handler): self.name name self.description description self.tools tools self.handler handler async def execute(self, input_data, contextNone): try: result await self.handler(input_data, self.tools, context) return {status: success, result: result} except Exception as e: return {status: error, error_message: str(e)} # 注册一个数据分析 Skill async def data_analysis_handler(input_data, tools, context): # 步骤1查询数据 raw_data await tools[query_db].ainvoke({sql: input_data[query]}) # 步骤2计算统计量 stats await tools[calculate].ainvoke({expression: ...}) # 步骤3生成图表 chart await tools[plot].ainvoke({data: raw_data}) return {stats: stats, chart: chart} analysis_skill Skill( namedata_analysis, description对数据集进行统计分析并生成图表, tools{query_db: query_tool, calculate: calc_tool, plot: plot_tool}, handlerdata_analysis_handler )Skill 注册到 Agent 时把description作为工具描述暴露给主 Agent主 Agent 根据描述判断何时调用。handler内部的编排逻辑对主 Agent 透明它只需要知道“这个 Skill 能做数据分析”就够了。4. 踩坑实录与排查技巧4.1 子 Agent 不按预期路由这是最常见的问题。主 Agent 把天气问题路由给了数学子 Agent或者干脆自己回答了。排查思路分三步先看主 Agent 的提示词是否足够明确路由规则有没有写清楚再看子 Agent 的description是否准确主 Agent 是根据这个来判断的最后看模型本身的能力有些小模型在路由判断上确实力不从心。我的经验是路由规则要写成显式的 if-else 逻辑不要指望模型自己领悟。比如“如果用户问题包含天气温度下雨等关键词交给 weather_expert”这种明确的规则比“判断问题类型”这种模糊指令有效得多。4.2 MCP 工具调用超时MCP 走 stdio 时如果 Server 进程启动慢或者工具执行时间长容易超时。解决方案有两个一是增加超时时间在 Client 侧配置timeout参数二是把耗时操作改成异步Server 侧立即返回一个任务 IDClient 轮询结果。我遇到过 Server 进程因为依赖包导入慢导致启动超时的情况。后来把重依赖改成懒加载启动时间从 8 秒降到 1 秒以内问题就消失了。4.3 上下文在 Agent 间丢失子 Agent 执行到一半发现缺少必要信息但又不能直接问用户。这个问题根源在于主 Agent 分发任务时信息给少了。我的做法是在主 Agent 的提示词里加一条“分发任务前确认子 Agent 拥有完成任务所需的全部信息如有缺失先向用户补充询问。”另外A2A 消息的context字段要结构化不要塞一大段文本。用 JSON 明确标注每个字段的含义子 Agent 解析起来不容易出错。4.4 Skill 执行结果格式不统一不同 Skill 返回的结果格式五花八门主 Agent 整合时很头疼。解决办法是强制统一返回格式。我定义了一个标准的结果结构{ skill_name: str, status: success | error, summary: str, # 自然语言摘要 data: dict, # 结构化数据 artifacts: list # 生成的文件、图表等 }所有 Skill 的handler都必须返回这个结构主 Agent 按固定字段读取不用做格式判断。4.5 常见问题速查表问题现象可能原因排查方向解决方案主 Agent 自己回答问题提示词不够克制检查 system_prompt明确写“不自己回答”子 Agent 选错工具工具描述模糊检查 MCP tool description补充使用场景说明MCP 连接失败Server 未启动或路径错误检查 command 和 args手动运行 Server 验证结果整合混乱返回格式不统一检查各 Skill 输出强制统一返回结构token 消耗过高上下文传递过多检查 A2A context只传必要片段子 Agent 死循环重试逻辑无上限检查重试配置设置最大重试次数5. 扩展方向与个人实践体会这套架构跑通之后扩展方向其实很多。我目前在做的一个方向是动态 Skill 加载根据任务类型在运行时从 Skill 库中加载对应的技能包而不是启动时就全部注册。这样 Agent 的初始 prompt 可以保持精简只在需要时才引入相关 Skill 的描述。另一个方向是跨会话的 Agent 记忆。目前每个会话结束后子 Agent 的状态就丢了下次遇到类似任务要从头开始。我在尝试用向量数据库存储历史任务的执行轨迹新任务来时先检索相似案例把相关经验注入到子 Agent 的上下文中。初步测试下来对于重复性高的任务类型准确率有提升。MCP 生态目前还在快速扩张社区里已经有不少现成的 Server 可以直接用。我的建议是优先用社区 Server把精力放在 Agent 编排和 Skill 设计上不要重复造轮子。但要注意审查社区 Server 的质量有些实现比较粗糙参数校验和错误处理做得不到位接入前最好先单独测试一遍。最后分享一个我在调试多 Agent 系统时的小技巧给每个 Agent 的日志加上唯一标识。主 Agent 用[SUPERVISOR]天气子 Agent 用[WEATHER]数学子 Agent 用[MATH]。这样在终端里看日志时一眼就能看出消息在哪个环节流转、在哪里卡住了。这个习惯帮我省了大量排查时间尤其是 Agent 数量多起来之后没有标识的日志根本没法看。这套东西目前还在演进中DeepAgents 的 API 可能还会变MCP 的规范也在补充。但核心思路是稳定的分层解耦、协议标准化、能力可复用。把这三点抓住不管工具怎么变架构都不会过时。