ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

DeepAgents + MCP + A2A + Skills:多智能体集群架构设计与实战

DeepAgents + MCP + A2A + Skills:多智能体集群架构设计与实战 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 SkillsAgent 的技能包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 负责调度多个 AgentA2A 负责 Agent 之间的通信。这个分层的好处是职责清晰。能力层只管提供能力不管怎么用技能层只管封装经验不管谁来调Agent 层只管完成自己的任务编排层只管调度。任何一层出问题排查范围都很明确。3.2 一次完整任务的流转过程我拿一个实际例子来说明。假设用户提了一个需求帮我分析一下最近三个月的销售数据生成一份报告并给出改进建议。第一步编排层的 DeepAgents 接到任务把它拆成三个子任务数据获取、数据分析、报告生成。第二步数据获取子任务分配给数据 Agent这个 Agent 加载了数据库查询 SkillSkill 内部调用 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 ServerMCP 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( nameread_file, description读取指定路径的文件内容, inputSchema{ type: object, properties: { path: {type: string, description: 文件路径} }, required: [path] } ), Tool( namewrite_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, encodingutf-8) as f: return [TextContent(typetext, textf.read())] elif name write_file: with open(arguments[path], w, encodingutf-8) as f: f.write(arguments[content]) return [TextContent(typetext, 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(modelgpt-4o) data_agent create_deep_agent( namedata_agent, llmllm, skills[data_fetch], description负责从各种数据源获取数据 ) analysis_agent create_deep_agent( nameanalysis_agent, llmllm, skills[data_analysis], description负责数据分析和洞察提取 ) writer_agent create_deep_agent( namewriter_agent, llmllm, skills[report_writing], description负责生成结构化报告 ) supervisor create_deep_agent( namesupervisor, llmllm, 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( nameanalysis_agent, description数据分析专家能处理结构化数据的统计分析, urlhttp://localhost:8001, skills[ A2ASkill( idstatistical_analysis, name统计分析, description对数值型数据做描述性统计和异常检测 ) ] ) server A2AServer(cardcard, agentanalysis_agent) server.run(port8001)其他 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 开始跑通了再加。每加一个组件都要确保它真的解决了问题而不是为了用新技术而用。这套东西的价值在于解决实际问题不在于技术本身有多炫。
返回列表