1. 从单体到集群:为什么我们需要重新理解 Agent 架构
过去一年我一直在折腾各种 Agent 项目,从最简单的单轮对话机器人,到带工具调用的 ReAct 循环,再到多角色协作的流水线,踩过的坑比写过的代码还多。最开始我以为把 Prompt 写得更精细、把工具函数挂得更多,Agent 就能变聪明,结果发现这条路走到某个临界点就彻底卡住了——单个 Agent 的上下文窗口是有限的,工具描述堆到几十个之后模型开始犯迷糊,任务一复杂就陷入死循环,更别提多个业务系统之间还要互相调用。
这就是DeepAgents + MCP + A2A + Skills这套组合拳出现的背景。它要解决的核心问题不是"让一个 Agent 更聪明",而是"让一群 Agent 能像一支训练有素的团队一样协同干活"。你可以把它理解成从"单兵作战"升级到"特种小队作战":DeepAgents 负责编排调度,MCP 负责把外部能力标准化地接进来,A2A 负责让不同团队造的 Agent 能互相通信,Skills 则是每个 Agent 随身携带的"技能包"。
这套东西适合谁?如果你已经在用 LangChain、AutoGen 或者自己手搓过 Agent 循环,感觉单 Agent 已经到天花板了,那这篇内容就是给你准备的。如果你是完全的新手,我建议先把单个 Agent 的工具调用跑通再来看,不然容易一头雾水。接下来我会把这四个概念拆开揉碎,讲清楚它们各自解决什么问题、怎么配合、以及我在实际搭建过程中遇到的那些文档里不会写的坑。
2. 四大核心组件拆解:各自解决什么问题
2.1 DeepAgents:编排层的"项目经理"
DeepAgents 这个概念本质上是一个编排框架,它的职责是决定"哪个任务该交给哪个 Agent 去做、按什么顺序做、做完之后结果怎么汇总"。我习惯把它类比成项目经理:它自己不写代码,但它知道整个项目分几个阶段,每个阶段需要什么角色,谁先谁后,谁依赖谁。
它和普通的 Agent 循环最大的区别在于分层。普通 Agent 是一个 while 循环,模型自己决定下一步调什么工具。DeepAgents 则是把任务拆成子任务,每个子任务分配给专门的子 Agent,子 Agent 有自己的上下文和工具集,做完之后把结果汇报给编排层。这样做的好处非常明显:每个子 Agent 的上下文都很干净,只关心自己那一小块,不会被无关信息干扰。
我在实际项目里最常用的编排模式有三种。第一种是顺序流水线,比如"抓取数据 → 清洗 → 分析 → 生成报告",每个环节一个 Agent,前一个的输出是后一个的输入。第二种是并行扇出,比如同时让三个 Agent 从不同角度调研同一个问题,最后汇总。第三种是带条件分支的循环,比如"写代码 → 跑测试 → 如果失败就回到写代码",这个最容易出问题,后面会专门讲。
2.2 MCP:把外部能力标准化接进来
MCP 全称是 Model Context Protocol,你可以把它理解成 Agent 世界的"USB 接口标准"。在 MCP 出现之前,每接一个外部工具(数据库、文件系统、第三方 API),你都要写一套适配代码,工具多了之后维护成本爆炸。MCP 做的事情就是定义一套统一的协议,任何工具只要按这个协议暴露自己的能力,Agent 就能直接调用,不用关心底层是什么。
MCP 的核心抽象有三个:Resources(可读取的数据源,比如文件、数据库记录)、Tools(可执行的操作,比如发邮件、查天气)、Prompts(预定义的提示模板)。我实测下来,Resources 和 Tools 用得最多,Prompts 相对少一些。
为什么 MCP 这么重要?因为它把"能力接入"这件事从"每个项目重复造轮子"变成了"一次封装到处复用"。我举个例子,我写了一个查询 PostgreSQL 的 MCP Server,那么任何支持 MCP 的 Agent 框架都能直接用这个 Server,不用改一行代码。这就是标准化的威力。
2.3 A2A:让不同团队造的 Agent 能对话
A2A 是 Agent-to-Agent 的缩写,解决的是跨系统 Agent 通信的问题。MCP 解决的是"Agent 怎么调用工具",A2A 解决的是"Agent 怎么调用另一个 Agent"。这两个是不同层次的问题,很多人会搞混。
我打个比方:MCP 像是你给员工配了一台电脑,他能用电脑上的软件干活;A2A 像是两个不同部门的员工能互相发消息、派任务。A2A 定义了一套 Agent 之间通信的标准,包括怎么发现对方、怎么描述自己的能力(Agent Card)、怎么发起任务、怎么返回结果。
A2A 最实用的场景是跨组织协作。比如你公司有一个专门做风控的 Agent,合作方有一个专门做授信的 Agent,两边用不同的框架、不同的模型,但只要都实现了 A2A 协议,就能互相调用。这在以前是不可想象的,你得写一堆胶水代码。
2.4 Skills:Agent 的"技能包"
Skills 这个概念相对轻量,它指的是封装好的、可复用的能力单元。一个 Skill 通常包含一段提示词、一组工具、以及一些示例。你可以把它理解成 Agent 的"插件"或者"技能卡"。
Skills 和 MCP Tools 的区别在于粒度。MCP Tool 是原子操作,比如"读文件"、"发请求";Skill 是完成某个具体任务的完整能力,比如"写一篇论文"、"做一次代码审查"。一个 Skill 内部可能会调用多个 MCP Tools。
我个人的经验是,Skills 是让 Agent 从"能用"到"好用"的关键。因为原子工具谁都会调,但怎么组合工具、用什么提示词、注意哪些细节,这些经验封装成 Skill 之后,就能被所有 Agent 复用。这就像公司里的 SOP 文档,新人照着做也能达到老手的水平。
3. 架构设计:四者如何协同工作
3.1 整体分层架构
把这四个东西拼在一起,整个架构大概分四层。最底层是能力层,由 MCP Server 提供各种原子能力,比如文件读写、数据库查询、API 调用。往上一层是技能层,Skills 把多个 MCP 能力组合成完成具体任务的技能包。再往上是Agent 层,每个 Agent 加载自己需要的 Skills,具备独立完成任务的能力。最顶层是编排层,DeepAgents 负责调度多个 Agent,A2A 负责 Agent 之间的通信。
这个分层的好处是职责清晰。能力层只管提供能力,不管怎么用;技能层只管封装经验,不管谁来调;Agent 层只管完成自己的任务;编排层只管调度。任何一层出问题,排查范围都很明确。
3.2 一次完整任务的流转过程
我拿一个实际例子来说明。假设用户提了一个需求:"帮我分析一下最近三个月的销售数据,生成一份报告,并给出改进建议。"
第一步,编排层的 DeepAgents 接到任务,把它拆成三个子任务:数据获取、数据分析、报告生成。第二步,数据获取子任务分配给"数据 Agent",这个 Agent 加载了"数据库查询 Skill",Skill 内部调用 MCP 的 PostgreSQL Server 拿到数据。第三步,数据分析子任务分配给"分析 Agent",它加载了"统计分析 Skill"。第四步,报告生成子任务分配给"写作 Agent",它加载了"报告撰写 Skill"。
如果分析 Agent 发现数据有问题,需要重新获取,它可以通过 A2A 直接给数据 Agent 发消息,不用绕回编排层。这就是 A2A 的价值——点对点通信,减少不必要的往返。
3.3 为什么这样设计而不是别的方案
有人可能会问,为什么不直接用一个超级 Agent 加一堆工具?我试过,答案是上下文爆炸。当工具超过 20 个,模型选择工具的准确率就明显下降;当任务步骤超过 10 步,模型很容易忘记前面的约束。分层之后,每个 Agent 的上下文都很小,专注度更高。
还有人问,为什么不用简单的函数调用而要用 MCP?因为复用性。函数调用是写死在代码里的,换个项目就得重写;MCP Server 是独立的服务,任何项目都能接。我现在的做法是,常用的能力(文件、数据库、HTTP、搜索)都封装成 MCP Server,新项目直接接进来,省了大量重复工作。
4. 实操搭建:从零跑通一个多智能体集群
4.1 环境准备与依赖安装
先说环境。我用的 Python 3.11,太新的版本有些库还没适配,太老的版本类型提示会有问题。虚拟环境用 venv 就行,没必要上 conda。
python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install deepagents mcp a2a-sdk这里有个坑要提醒:mcp这个包名和另一个同名的老包冲突,如果你之前装过别的 mcp,先pip uninstall mcp再装。我在这上面浪费了半小时,一直报 import 错误。
模型方面,我建议至少准备两个:一个能力强的做主 Agent(编排和复杂推理),一个速度快成本低的做子 Agent(执行简单任务)。这样能显著降低成本。
4.2 写第一个 MCP Server
MCP Server 是整个体系的基础,先把它跑通。我写一个最简单的文件操作 Server 作为例子。
from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent import os app = Server("file-server") @app.list_tools() async def list_tools(): return [ Tool( name="read_file", description="读取指定路径的文件内容", inputSchema={ "type": "object", "properties": { "path": {"type": "string", "description": "文件路径"} }, "required": ["path"] } ), Tool( name="write_file", description="写入内容到指定文件", inputSchema={ "type": "object", "properties": { "path": {"type": "string"}, "content": {"type": "string"} }, "required": ["path", "content"] } ) ] @app.call_tool() async def call_tool(name: str, arguments: dict): if name == "read_file": with open(arguments["path"], "r", encoding="utf-8") as f: return [TextContent(type="text", text=f.read())] elif name == "write_file": with open(arguments["path"], "w", encoding="utf-8") as f: f.write(arguments["content"]) return [TextContent(type="text", text="写入成功")] 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 传输,适合本地进程调用。如果要跨机器,得换成 SSE 或者 HTTP 传输,配置会复杂一些。
注意:MCP Server 的 inputSchema 一定要写清楚,模型靠这个理解工具怎么用。description 写得越具体,模型调用越准确。我见过太多人 description 就写一句"读取文件",结果模型老是传错参数。
4.3 定义 Skills 并加载
Skill 我一般用一个 YAML 文件定义,包含名称、描述、提示词、依赖的工具。
name: data_analysis description: 对结构化数据做统计分析并生成洞察 prompt: | 你是一个数据分析专家。拿到数据后,按以下步骤操作: 1. 先检查数据完整性,缺失值超过 30% 的列要标注出来 2. 计算关键指标的均值、中位数、同比环比 3. 找出异常值(超过 3 倍标准差的点) 4. 用自然语言总结 3-5 条核心洞察 注意:不要编造数据,所有结论必须有数据支撑。 tools: - read_file - run_python - write_file examples: - input: "分析 sales.csv" output: "报告包含:数据概况、关键指标、异常点、洞察建议"加载 Skill 的时候,把 prompt 注入到 Agent 的系统提示里,把 tools 注册到 Agent 的工具列表。这样 Agent 就"学会"了这个技能。
4.4 用 DeepAgents 编排多个 Agent
编排层是重头戏。我用 DeepAgents 定义一个主管 Agent 和三个子 Agent。
from deepagents import create_deep_agent from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o") data_agent = create_deep_agent( name="data_agent", llm=llm, skills=["data_fetch"], description="负责从各种数据源获取数据" ) analysis_agent = create_deep_agent( name="analysis_agent", llm=llm, skills=["data_analysis"], description="负责数据分析和洞察提取" ) writer_agent = create_deep_agent( name="writer_agent", llm=llm, skills=["report_writing"], description="负责生成结构化报告" ) supervisor = create_deep_agent( name="supervisor", llm=llm, sub_agents=[data_agent, analysis_agent, writer_agent], system_prompt="""你是一个项目主管。接到任务后: 1. 拆解任务,判断需要哪些子 Agent 参与 2. 按依赖关系排序,能并行的并行 3. 每个子任务完成后检查结果质量 4. 全部完成后汇总输出 如果某个子任务失败,最多重试 2 次,还失败就报告问题。""" )这里的关键是 supervisor 的 system_prompt,它决定了整个编排的质量。我调了很多版才找到比较稳的写法,核心是明确重试策略和失败处理,不然模型遇到错误会一直重试或者直接放弃。
4.5 配置 A2A 实现 Agent 间通信
A2A 的配置稍微复杂一点,需要给每个 Agent 暴露一个 Agent Card,描述它的能力。
from a2a import A2AServer, AgentCard, Skill as A2ASkill card = AgentCard( name="analysis_agent", description="数据分析专家,能处理结构化数据的统计分析", url="http://localhost:8001", skills=[ A2ASkill( id="statistical_analysis", name="统计分析", description="对数值型数据做描述性统计和异常检测" ) ] ) server = A2AServer(card=card, agent=analysis_agent) server.run(port=8001)其他 Agent 要调用它,先通过 URL 拿到 Agent Card,知道它能干什么,然后发任务请求。A2A 的好处是能力发现——你不需要提前知道对方有什么能力,运行时查询就行。
5. 踩坑实录:那些文档不会告诉你的问题
5.1 上下文污染与隔离
我遇到最头疼的问题是上下文污染。子 Agent 完成任务后,如果把完整的中间过程都返回给主管 Agent,主管的上下文很快就被塞满了,后面做决策就开始犯糊涂。
解决办法是只返回结构化摘要。我要求每个子 Agent 返回固定格式:任务状态、关键结果、遇到的问题、建议下一步。中间过程全部丢弃。这样主管 Agent 的上下文始终很干净。
RESULT_FORMAT = """ 请按以下格式返回结果: - status: success/failed/partial - result: 核心结果(不超过 200 字) - issues: 遇到的问题(没有就写 none) - next: 建议的下一步(没有就写 none) """这个格式我强制加在每个子 Agent 的提示词末尾,实测下来主管 Agent 的决策准确率提升非常明显。
5.2 死循环与超时控制
多 Agent 系统最容易出的问题是死循环。A 让 B 做事,B 做不了让 A 帮忙,A 又让 B 做,来回踢皮球。我见过最夸张的一次跑了 40 多轮还没结束,token 烧了一大堆。
我的做法是加三层防护。第一层是每个 Agent 的最大步数限制,超过就强制返回。第二层是编排层的最大轮次限制,比如整个任务最多 20 轮。第三层是全局超时,比如 5 分钟没结果就中断。三层任意一层触发都会终止并返回当前状态。
提示:超时时间不要设太短,复杂任务确实需要时间。我的经验是简单任务 60 秒,中等任务 3 分钟,复杂任务 10 分钟。宁可设长一点,也不要频繁中断导致任务失败。
5.3 工具调用失败的优雅降级
MCP 工具调用失败是常态,网络抖动、参数错误、权限问题都会导致失败。如果每次失败都让整个任务崩掉,那系统就没法用了。
我的做法是分级降级。第一级,同一个工具重试 2 次,间隔 1 秒。第二级,如果还是失败,尝试用替代工具(比如主数据库挂了就用缓存)。第三级,如果替代也没有,返回明确的错误信息让上层决策。关键是错误信息要具体,不能只说"失败了",要说"连接超时"还是"参数错误"还是"权限不足",上层才能做出正确判断。
5.4 常见问题速查表
| 问题现象 | 可能原因 | 排查方向 | 解决方案 |
|---|---|---|---|
| Agent 反复调用同一工具 | 提示词没说明终止条件 | 检查 system_prompt | 明确"拿到结果后停止" |
| 子 Agent 返回空结果 | 上下文太长被截断 | 看 token 用量 | 精简输入,只传必要信息 |
| MCP 连接失败 | 端口占用或进程没起 | 检查进程和端口 | 重启 Server,换端口 |
| A2A 找不到对方 | Agent Card 没注册 | 检查注册中心 | 确认 URL 和注册状态 |
| 任务卡住不动 | 死循环或等待超时 | 看日志最后一步 | 加超时和步数限制 |
| 结果质量差 | Skill 提示词太笼统 | 检查 Skill 定义 | 补充示例和约束条件 |
6. 性能优化与扩展思路
6.1 并行化能省多少时间
多 Agent 最大的性能优势是并行。我做过一个测试,一个包含 5 个独立子任务的分析任务,串行执行平均 4 分 20 秒,并行执行只要 1 分 10 秒,提速接近 4 倍。当然前提是子任务之间没有依赖。
判断能不能并行的标准很简单:子任务之间有没有数据依赖。如果 B 需要 A 的输出才能开始,那就必须串行;如果 A 和 B 各自独立,只是最后汇总,那就可以并行。我在编排层加了一个依赖分析步骤,让主管 Agent 先画出任务依赖图,再决定执行顺序。
6.2 成本控制的几个实用技巧
多 Agent 系统的 token 消耗是单 Agent 的好几倍,成本控制很重要。我总结了几个技巧。
第一,分级用模型。主管 Agent 用强模型,子 Agent 用便宜模型。实测下来,子 Agent 用便宜模型完成简单任务的质量差距很小,但成本能降 70%。
第二,缓存重复调用。同样的查询、同样的分析,结果缓存起来,下次直接返回。我用了简单的文件缓存,key 是输入内容的哈希,命中率能到 30% 左右。
第三,精简提示词。系统提示词每多 100 字,每次调用就多消耗 100 token。我把所有提示词都精简过一遍,去掉了所有客套话和重复说明,整体 token 用量降了 25%。
6.3 后续可以怎么扩展
这套架构搭好之后,扩展性很好。我目前想到几个方向。
一是接入更多 MCP Server。社区里已经有大量现成的 Server,数据库、云服务、办公软件都有,接进来就能用。
二是建立 Skill 市场。把团队里积累的 Skill 集中管理,新项目直接引用,避免重复造轮子。
三是跨组织 A2A 协作。如果合作方也用了 A2A,可以直接对接,不用写胶水代码。这个在供应链、金融风控场景特别有价值。
四是加监控和可观测性。多 Agent 系统的调试比单 Agent 难得多,我现在用 LangSmith 做追踪,每一步的输入输出都能看到,排查问题快很多。
7. 我个人的一些实战体会
搭这套系统最大的感受是,架构比模型重要。我试过用最强的模型配最烂的架构,结果还不如用中等模型配好架构。分层、隔离、降级、超时,这些工程上的东西才是决定系统能不能用的关键。
另一个体会是,Skill 的质量决定上限。同样的框架,Skill 写得好和写得差,输出质量天差地别。我现在花在打磨 Skill 提示词上的时间,比写代码的时间还多。一个好的 Skill 提示词,要包含角色定义、操作步骤、注意事项、输出格式、示例,缺一不可。
最后说个细节,日志一定要打全。多 Agent 系统出问题时,你根本不知道是哪一步、哪个 Agent 出的错。我的做法是每个 Agent 的每次调用都记录:时间戳、Agent 名称、输入摘要、输出摘要、耗时、token 用量。出问题时按时间线一看就清楚。这个习惯帮我省了无数排查时间。
如果你也在搭多 Agent 系统,我的建议是先跑通最小闭环再扩展。别一上来就搞十几个 Agent,先从两个 Agent 加一个 MCP Server 开始,跑通了再加。每加一个组件都要确保它真的解决了问题,而不是为了用新技术而用。这套东西的价值在于解决实际问题,不在于技术本身有多炫。