1. 从单体到集群为什么我们需要重新思考 Agent 的架构过去一年里我陆陆续续帮几个团队落地过智能体项目从最简单的客服问答机器人到稍微复杂一点的代码审查助手踩过的坑可以说能写满一个笔记本。最开始大家的思路都很朴素找一个能力最强的模型写一个足够长的系统提示词把所有工具都挂上去然后祈祷它能搞定一切。这个思路在任务简单的时候确实能跑通但只要任务链条一长、涉及的工具一多问题就全冒出来了——上下文爆炸、工具调用混乱、错误无法回溯、单个环节失败导致整个流程崩溃。这就是为什么当我第一次看到“DeepAgents MCP A2A Skills”这套组合的时候眼前一亮的根本原因。它不是在教你如何把提示词写得更花哨而是在回答一个更本质的问题当单个智能体扛不住复杂任务时我们该如何像搭积木一样把多个各有所长的智能体组织成一个可编排、可互通、可扩展的集群这篇文章我想聊的就是这套架构的完整落地思路。核心关键词会围绕DeepAgents、MCP、A2A、Skills、多智能体这几个概念展开但我不想写成一份干巴巴的协议说明书。我会从“为什么这么设计”讲起把每个组件的职责边界、它们之间怎么配合、实际编码时哪些地方最容易翻车都掰开揉碎讲清楚。适合已经用过基础 Agent 框架、想往多智能体方向进阶的开发者也适合正在做技术选型、想搞清楚这几个热门概念到底解决什么问题的架构师。先说结论这套架构的本质是把“一个全能大脑”拆成“一个调度中心 若干专业执行者 一套标准接口 一批可复用技能包”。听起来像微服务那一套没错思路确实一脉相承只不过服务对象从业务逻辑变成了智能体的推理与行动能力。2. 四大核心组件到底各管什么职责边界与协作逻辑在动手写代码之前必须先把这四个东西的定位搞清楚。我见过太多人把 MCP 和 A2A 混为一谈结果架构设计得一塌糊涂。下面这张表是我自己总结的职责对照建议先看明白再往下读。组件核心职责类比解决的问题DeepAgents智能体的编排与调度框架项目总指挥谁来做、按什么顺序做、失败了怎么办MCP智能体与外部工具/数据源的连接协议标准电源插座工具怎么接、数据怎么取统一接口A2A智能体与智能体之间的通信协议部门间的沟通规范多个 Agent 怎么互相调用、传递任务Skills可复用的能力封装单元标准零件库常用能力怎么沉淀、怎么复用2.1 DeepAgents不只是“管事的”更是“兜底的”DeepAgents 在这个体系里扮演的是编排层的角色。你可以把它理解成一个项目经理它自己不干具体的活但它要决定这个任务拆成几步、每一步交给哪个 Agent、上一步的输出怎么传给下一步、某一步失败了是重试还是换人。我实际用下来DeepAgents 最值钱的地方不是它的调度能力而是它的状态管理和容错机制。多智能体系统最怕的就是“中间某个环节挂了整个流程卡死你还不知道卡在哪”。DeepAgents 通过维护一个全局的任务状态图让每一步的执行结果都有迹可循。这一点在调试阶段简直是救命稻草。注意不要一上来就把所有 Agent 都塞进 DeepAgents 的编排图里。我建议先从两三个 Agent 的线性流程开始跑通了再逐步加分支和并行。一次性设计太复杂的编排图调试成本会指数级上升。2.2 MCP让工具接入从“每家一个样”变成“统一标准”MCP 全称是 Model Context Protocol翻译过来就是模型上下文协议。它的核心价值用一句话概括把“智能体怎么调用外部工具”这件事标准化了。在没有 MCP 之前你每接一个工具就要写一套适配代码。接数据库写一套接文件系统写一套接第三方 API 再写一套。工具一多代码里全是胶水逻辑维护起来想死。MCP 做的事情就是定义了一套标准的“工具描述 调用请求 返回结果”的格式只要工具方按照这个格式暴露接口智能体这边就能用统一的方式去调用。我打个比方以前每个电器都有自己的充电口你得备一堆线。MCP 就是那个统一成 Type-C 的过程以后不管什么电器一根线搞定。实际落地时MCP Server 负责把工具能力暴露出来MCP Client 负责在智能体侧发起调用。中间走的是标准化的 JSON-RPC 消息。这意味着你可以把常用的工具——比如数据库查询、文件读写、HTTP 请求——都封装成 MCP Server然后在不同的 Agent 之间共享。2.3 A2A智能体之间的“对话规则”A2A 是 Agent-to-Agent 的缩写解决的是智能体之间怎么互相通信的问题。如果说 MCP 解决的是“Agent 怎么用工具”那 A2A 解决的就是“Agent 怎么用另一个 Agent”。这个区别非常关键。举个例子你有一个专门做数据分析的 Agent和一个专门做报告撰写的 Agent。报告 Agent 需要数据分析 Agent 先跑完分析拿到结果才能写报告。这个“报告 Agent 请求数据分析 Agent 干活”的过程走的就是 A2A 协议。A2A 的核心概念包括 Agent Card智能体的能力名片、Task任务单元、Message消息传递。Agent Card 特别重要它相当于一个智能体的“简历”声明了“我是谁、我能干什么、我怎么被调用”。有了这个其他 Agent 或者编排层就能动态发现和调用能力而不需要硬编码。2.4 Skills把重复造轮子的时间省下来Skills 这个概念最近特别火本质上它就是对常用能力的封装和复用。比如“读取 PDF 并提取关键信息”这个能力你在十个项目里可能都要用那就把它封装成一个 Skill以后直接调用。Skills 和 MCP 的区别在于MCP 更偏向于“连接外部资源”Skills 更偏向于“封装内部逻辑”。一个 Skill 可以内部调用多个 MCP 工具对外只暴露一个简洁的接口。这样上层 Agent 不需要关心底层用了哪些工具只需要知道“调用这个 Skill 能完成什么事”。我自己的习惯是凡是同一个逻辑在三个以上地方出现就抽成 Skill。这个原则跟写代码时“三次重复就重构”是一个道理。3. 环境搭建与基础配置从零把架子搭起来理论讲完了接下来是动手环节。这一部分我会给出完整的配置步骤和代码示例你可以直接照着跑。3.1 基础依赖安装与项目初始化首先确认你的运行环境。我实测下来Python 3.10 以上、Node.js 18 以上是比较稳妥的版本。低于这个版本某些依赖包会报兼容性错误。# 创建项目目录 mkdir multi-agent-cluster cd multi-agent-cluster # 初始化 Python 虚拟环境 python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate # 安装核心依赖 pip install deepagents mcp a2a-sdk pydantic httpx这里解释一下每个包的作用。deepagents是编排框架的核心库mcp提供了 MCP 协议的客户端和服务端实现a2a-sdk是 A2A 通信的 SDKpydantic用来做数据校验Agent 之间传数据格式校验非常重要httpx用于异步 HTTP 请求。实操心得安装的时候如果遇到版本冲突优先保证pydantic的版本在 2.x 以上。很多 Agent 框架的数据模型都依赖 Pydantic v2用 v1 会出一堆莫名其妙的验证错误。3.2 目录结构设计一个清晰的项目结构能省掉后面大量的找文件时间。我推荐这样组织multi-agent-cluster/ ├── agents/ # 各个 Agent 的定义 │ ├── orchestrator.py # 编排 Agent │ ├── researcher.py # 研究型 Agent │ └── writer.py # 写作型 Agent ├── mcp_servers/ # MCP 服务端 │ ├── file_server.py │ └── db_server.py ├── skills/ # 可复用技能 │ ├── pdf_extract.py │ └── web_search.py ├── configs/ # 配置文件 │ └── agent_cards.yaml └── main.py # 入口这个结构的好处是职责分明。Agent 定义、工具服务、技能封装各占一个目录后面加新东西的时候不会乱。3.3 第一个 MCP Server让 Agent 能读文件我们从最基础的文件读取开始。下面是一个最小可用的 MCP Server 实现# mcp_servers/file_server.py from mcp.server import Server from mcp.types import Tool, TextContent import asyncio 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] } ) ] app.call_tool() async def call_tool(name: str, arguments: dict): if name read_file: path arguments[path] try: with open(path, r, encodingutf-8) as f: content f.read() return [TextContent(typetext, textcontent)] except FileNotFoundError: return [TextContent(typetext, textf文件不存在: {path})] raise ValueError(f未知工具: {name}) if __name__ __main__: asyncio.run(app.run())这段代码的关键点在于list_tools和call_tool两个装饰器。前者告诉客户端“我有哪些工具可用”后者负责实际执行。inputSchema用的是 JSON Schema 格式这是 MCP 协议规定的标准描述方式。注意inputSchema一定要写清楚required字段。我踩过的坑是没写 required结果 Agent 调用的时候漏传参数服务端直接报 KeyError排查了半天才发现是 schema 没约束好。3.4 定义 Agent Card让每个 Agent 有“身份证”A2A 协议要求每个 Agent 都要有一张 Agent Card声明自己的能力。下面是一个示例# configs/agent_cards.yaml researcher: name: ResearchAgent description: 负责信息检索和资料整理 capabilities: - web_search - pdf_extract - summarize endpoint: http://localhost:8001/a2a input_format: text output_format: structured_json writer: name: WriterAgent description: 负责根据资料撰写报告 capabilities: - outline_generate - content_write - polish endpoint: http://localhost:8002/a2a input_format: structured_json output_format: markdown这张卡片的作用是让编排层能够动态发现能力。比如编排 Agent 拿到一个“写一份行业分析报告”的任务它先查 Agent Card 列表发现需要先调 Researcher 再调 Writer于是自动组装出执行链路。这就是“可编排”的基础。4. 多智能体协作的完整实操流程环境搭好之后我们来看一个完整的协作场景。我设计了一个“自动生成技术调研报告”的流程涉及三个 Agent 的配合。4.1 场景拆解与任务分配任务目标给定一个技术主题自动搜索资料、整理要点、生成一份结构化报告。拆解后的执行链路Orchestrator Agent接收任务拆解为“检索 → 整理 → 撰写”三步Researcher Agent通过 MCP 调用搜索工具和文件读取工具收集资料Writer Agent接收整理好的资料生成最终报告Orchestrator汇总结果返回给用户这个链路里Orchestrator 和 Researcher 之间、Researcher 和 Writer 之间走的都是 A2A 协议。Researcher 调用搜索工具走的是 MCP 协议。4.2 编排层的核心代码实现# agents/orchestrator.py from deepagents import Orchestrator, Task from a2a_sdk import A2AClient import yaml class ReportOrchestrator: def __init__(self, config_path: str): with open(config_path) as f: self.agent_cards yaml.safe_load(f) self.clients {} for agent_id, card in self.agent_cards.items(): self.clients[agent_id] A2AClient(card[endpoint]) async def run(self, topic: str): # 第一步调用 Researcher 收集资料 research_task Task( actionresearch, payload{topic: topic, max_sources: 5} ) research_result await self.clients[researcher].send_task(research_task) if research_result.status ! success: return {error: 资料检索失败, detail: research_result.message} # 第二步把资料传给 Writer 生成报告 write_task Task( actionwrite_report, payload{ topic: topic, materials: research_result.data, format: markdown } ) write_result await self.clients[writer].send_task(write_task) return { topic: topic, report: write_result.data, sources_count: len(research_result.data.get(sources, [])) }这段代码里send_task是 A2A 协议的核心方法。它把任务打包成标准格式发给目标 Agent然后等待结果返回。注意我加了状态检查——如果 Researcher 返回失败直接中断流程并返回错误而不是硬着头皮往下走。实操心得编排层一定要做前置校验和失败短路。我见过太多流程是“不管上一步成没成下一步照跑”结果错误层层传递最后报出来的错跟根因差了十万八千里。4.3 Researcher Agent 的实现细节# agents/researcher.py from a2a_sdk import A2AServer, TaskResult from mcp import ClientSession import asyncio class ResearcherAgent: def __init__(self, mcp_endpoints: list): self.mcp_endpoints mcp_endpoints self.server A2AServer(ResearchAgent) self.server.register_handler(research, self.handle_research) async def handle_research(self, payload: dict) - TaskResult: topic payload[topic] max_sources payload.get(max_sources, 5) # 通过 MCP 调用搜索工具 async with ClientSession(self.mcp_endpoints[0]) as session: search_result await session.call_tool( web_search, {query: topic, limit: max_sources} ) # 整理结果 materials [] for item in search_result: materials.append({ title: item.get(title), summary: item.get(snippet), url: item.get(url) }) return TaskResult( statussuccess, data{topic: topic, materials: materials, sources: materials} )这里的关键是ClientSession的用法。它负责跟 MCP Server 建立连接、发送调用请求、接收结果。整个过程对上层是透明的——Researcher Agent 不需要知道搜索工具具体是怎么实现的只需要知道“调用 web_search 能拿到结果”。4.4 Skills 的封装与复用现在来看 Skills 怎么用。假设“从搜索结果中提取关键信息”这个逻辑在 Researcher 和 Writer 里都要用到那就抽成一个 Skill# skills/key_extract.py from pydantic import BaseModel from typing import List class KeyPoint(BaseModel): content: str source: str confidence: float class KeyExtractSkill: name key_extract description 从文本材料中提取关键信息点 def __init__(self, llm_client): self.llm llm_client async def execute(self, materials: List[dict], max_points: int 10) - List[KeyPoint]: prompt self._build_prompt(materials, max_points) response await self.llm.generate(prompt) return self._parse_response(response) def _build_prompt(self, materials, max_points): text \n.join([m.get(summary, ) for m in materials]) return f从以下材料中提取最多{max_points}个关键信息点每个点标注来源\n{text} def _parse_response(self, response): # 解析逻辑略实际项目中需要处理各种格式异常 ...Skill 的封装原则是对外接口极简内部逻辑可以复杂。上层调用者只需要传材料、拿结果不关心里面用了什么模型、什么提示词、怎么解析。5. 常见问题与排查技巧实录多智能体系统的调试难度比单体 Agent 高一个数量级因为问题可能出在任何一个环节。下面这张表是我实际踩坑后整理的速查表。问题现象可能原因排查方法解决方案Agent 之间消息发不出去A2A endpoint 配置错误检查 Agent Card 里的 endpoint 是否可达用 curl 手动测试 endpointMCP 工具调用超时工具执行时间过长在 MCP Server 里加日志设置合理的 timeout加异步处理编排流程卡死某个 Agent 没有返回检查任务状态图加超时机制和失败回调结果格式不对Schema 定义不严格对比实际返回和 schema用 Pydantic 做强制校验Agent 重复调用同一工具提示词里没约束看调用日志在系统提示里明确“不要重复调用”5.1 消息传递失败的排查思路A2A 通信失败是最常见的问题。我的排查顺序是先确认目标 Agent 的服务是否启动再确认 endpoint 地址是否正确然后确认消息格式是否符合协议规范。# 手动测试 A2A endpoint 是否可达 curl -X POST http://localhost:8001/a2a \ -H Content-Type: application/json \ -d {action: ping}如果这一步就失败了那问题在服务端跟编排逻辑无关。如果这一步成功但编排层还是报错那就是消息格式的问题需要对比 SDK 文档检查字段名。5.2 上下文爆炸的预防多智能体系统里上下文爆炸是个隐形杀手。每个 Agent 都有自己的上下文窗口如果编排层把所有的历史消息都往下传很快就会撑爆。我的做法是只传必要信息不传完整历史。比如 Writer Agent 只需要 Researcher 整理好的材料不需要知道 Researcher 中间调了多少次搜索工具、每次返回了什么。在 A2A 消息里只放最终结果中间过程留在各自的日志里。注意如果你发现某个 Agent 的响应越来越慢、越来越离谱第一反应应该是检查它的上下文是不是太长了。这个问题的典型表现是“前面几步还对后面开始胡言乱语”。5.3 Skills 版本管理Skills 一旦多了版本管理就是个问题。我建议每个 Skill 都带上版本号并且在 Agent Card 里声明依赖的 Skill 版本。这样当 Skill 升级时不会意外影响还在用旧版本的 Agent。skills: key_extract: version: 1.2.0 compatible_agents: [researcher, writer]这个做法借鉴了软件依赖管理的思路。多智能体系统本质上就是一个分布式系统分布式系统踩过的坑这里一个都不会少。6. 扩展方向与个人实践体会这套架构跑通之后扩展性其实非常好。我目前尝试过的几个方向包括把 Skills 做成可热插拔的插件、用多个 Researcher Agent 并行检索不同来源、给编排层加优先级队列。有一个体会特别深多智能体系统的价值不在于“多”而在于“分工明确”。我见过有人为了用多智能体而用多智能体把本来一个 Agent 能搞定的事情拆成五个结果通信开销比干活的时间还长。判断标准很简单——如果拆开之后每个 Agent 的职责能用一句话说清楚且它们之间的依赖关系是清晰的那就值得拆如果拆完之后你自己都理不清谁调谁那就说明拆错了。另外MCP 和 A2A 这两个协议目前还在快速演进中接口可能会有变化。我的建议是在业务代码和协议层之间加一层薄薄的适配层把协议相关的调用封装起来。这样协议升级的时候只需要改适配层不用动业务逻辑。这个习惯帮我省了好几次大规模重构的时间。最后分享一个调试小技巧给每个 Agent 的输入输出都打上带时间戳的日志并且用一个统一的 trace_id 串起来。多智能体系统出问题的时候你最大的敌人不是 bug 本身而是“不知道 bug 出在哪个环节”。有了完整的调用链日志排查效率能提升好几倍。