尧图网页设计 
如果你最近在搞 AI Agent一定对 DeepAgents、MCP、A2A、Skills 这四个词不陌生。它们的共同点都在回答同一个问题当一个 Agent 不够用的时候我们怎么用一套清晰可扩展的架构把它变成一支能各自干活、又能互相协作的多智能体集群。这篇文章我会从架构设计讲到底层协议拆解再给出一套可以直接落地的多智能体集群搭建过程最后聊一聊我实际跑集群时踩过的坑。适合正在做 Agent 编排、工具接入、或者想把多个大模型节点串成流水线的朋友内容偏工程实践但我会尽量把概念讲明白新手也能跟着操作。我先把话说在前面DeepAgents 不是某个特定的开源项目而是一套“深度任务理解 规划 反思 行动”的智能体实现思路MCP 解决的是 Agent 和工具之间要如何标准化接入A2A 解决的是 Agent 和 Agent 之间要如何对话和交接Skills 则是给 Agent 注入可复用的“工作手册”。把它们拼起来才算是真正意义上的多智能体集群架构。下面开始。1. 多智能体集群架构先把“集群”这件事想清楚1.1 从单 Agent 到多 Agent瓶颈在协作而不在数量很多团队一开始的想法是一个 Agent 能力不够那就多写几个 Agent让它们各自处理一个子任务。真做起来就发现问题了。Agent 之间没有统一通信语言A 的输出没法直接变成 B 的输入工具接入方式五花八门有的用函数调用有的直接让模型输出 JSON每个 Agent 的 prompt 和技能逻辑互相复制改一处就要连带改好几个文件。我在一开始搭建多智能体集群的时候也犯过“堆数量”的错误。比如做研究型任务安排了 5 个 Agent 同时去搜资料结果它们各自返回了格式完全不同的笔记没有一个统一的上下文结构最终汇总时人还得手工整理。后来我意识到多智能体的核心不是多而是“分得清、传得动、合得拢”。要分就要有任务拆解和角色定义要传就要有统一的传输协议要合就要有明确的结果格式和验收标准。这才是我理解的多智能体集群架构一组自治的 Agent 节点通过标准化协议连接、协同形成一个整体系统对外服务。再具体一点一个可运行的多智能体集群通常由几个角色组成Planner 负责拆解任务Researcher 负责检索信息Writer 负责产出内容Reviewer 负责检查质量。它们不是随便几个“LLM 调用”堆在一起而是每个角色都承担明确的职责通过消息或任务对象互相交接。这种设计的好处是任何单一节点可以被替换、升级或水平扩展而不影响整个集群的对外行为。1.2 四个关键词各自负责哪一层我们经常听到“DeepAgents MCP A2A Skills”这种组合但很少有人解释清楚它们之间是什么关系。我用一个团队来类比DeepAgents 是项目经理的思维方式。它强调 Agent 面对复杂任务时要先拆解、再执行、然后反思修正。这是应用层的行为范式决定 Agent 怎么“思考”。MCP 是办公室里统一的插座标准。Agent 要调用数据库、网页搜索、文件系统等工具如果不统一接口每接一个新工具就要改一遍 Agent 代码。MCP 把这些能力统一成“即插即用”的协议。A2A 是团队之间的沟通契约。A2A 解决的是“Agent 与 Agent 怎么发现彼此、怎么派活、怎么回传结果”。它让不同团队、不同框架开发的 Agent 可以被统一调度。Skills 是给员工的新人培训手册。它把某个任务怎么做的步骤、模板、示例、脚本打包到一个目录里Agent 遇到对应场景时按手册执行。这四个层次可以互相叠加。比如一个 Researcher Agent它内部用 DeepAgents 的“规划-行动-反思”循环来思考它通过 MCP 调用网络搜索工具它把自己的能力声明成 A2A 的 Agent Card供编配器调用它加载了一个“高效检索”的 Skill知道如何整理搜索结果。这就是一套完整的多智能体工程化组合。1.3 一张“集群拓扑”长什么样先说拓扑后面所有实操都围绕着它。我建议的第一版集群是“中心化编配 分布式执行”一个 Orchestrator编配器作为入口负责任务理解、拆解和调度若干个 Worker Agent 作为执行节点各自暴露 A2A 端点所有 Worker 通过 MCP 客户端连接所需工具Skills 作为局部文件目录或远程仓库挂在每个 Worker 上。用户请求进入 Orchestrator 后拆出多个子任务按依赖关系发给不同 WorkerWorker 完成后再把产物回传最终由 Orchestrator 汇总输出。这个拓扑很朴素甚至有点“过时”但它能帮你把每个环节的职责理清。等跑通了再根据性能要求做去中心化改造也不迟。2. MCP、A2A、Skills 三件套的底层逻辑2.1 MCP 是“万能工具插座”不是又一种 RPC我第一次看到 MCPModel Context Protocol时第一反应是这不就是个远程调用协议吗后来深入用才明白它和普通 RPC 有个本质区别它面向的是大模型应用定义了一套“模型上下文”的边界。在 MCP 架构里有 Host、Client、Server 三层。Host 是大模型应用本身Client 负责维护与 Server 的连接Server 把能力暴露成三种原语Tools、Resources、Prompts。Tools 是让模型执行动作Resources 是让模型读取信息Prompts 是给模型复用预设提示词。你发现没有MCP 强调的是“模型上下文”不是单纯的函数调用。比如一个搜索工具它返回的不仅仅是字符串还应该有足够的上下文线索source、date、confidence这样才能让模型判断信息是否可信。因此 MCP 非常适合用于把企业内部数据库、网页搜索、代码仓库、设计稿等能力统一接入 Agent。我们用一段极简的 Python 示例来说清楚 MCP Server 的形态。假设我们写一个能查询用户信息的小服务使用官方 Python SDKfrom mcp.server.fastmcp import FastMCP mcp FastMCP(user-service) mcp.tool() def get_user_info(user_id: str) - dict: 根据 user_id 查询用户基础信息 return {user_id: user_id, name: 张三, role: admin, email: zhangsanexample.com} if __name__ __main__: mcp.run(transportstdio)这个 Server 通过 stdio 和 Host 进程通信协议消息是 JSON-RPC 2.0。Host 端的 Agent 只需要一个 MCP Client就能动态发现 get_user_info 这个工具以及它的参数结构。这种“能力即插即用”的体验和给电脑插 USB 设备很像所以我说它是“万能工具插座”。还要澄清一个热词里经常看到的疑问MCP 是软件协议还是硬件协议它和物理层硬件协议完全是两码事它运行在应用层描述的是大模型应用和外部工具之间的语义交互方式你可以理解为一份“软件接口契约”而不是网线或 USB 接口那种物理协议。2.2 A2A 是智能体之间的“协作语言”如果说 MCP 解决的是“人Agent和工具”的关系那 A2AAgent-to-Agent解决的就是“人和人Agent 和 Agent”的关系。A2A 的核心思想是让不同 Agent 通过一组标准对象进行交流而不是互相写死对方的私有接口。协议里最重要的几个对象是Agent Card一个 Agent 的公开身份和能力描述通常是一个 JSON 文件写清楚名称、说明、能力、认证方式、端点地址。Task表示一个需要完成的请求有 task_id、status、input、artifacts 等字段。MessageAgent 之间传输的具体信息可以是文本、对象、文件引用。Artifact任务产出的结果可以是报告、数据表、图片等。我自己的实现里会为每个 Worker 暴露一个 HTTP 端点比如 /a2a/task。当 Orchestrator 要派活时它会做两件事先根据 Agent Card 决定选哪个 Agent然后发送一个 Task 请求。Worker 接收 Task 后在自己的循环中处理并通过 PATCH 或者 WebSocket 持续更新状态。最后把结果 Artifact 返回。下面是一段用 FastAPI 写的极简 A2A 接收端点from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional app FastAPI(titleResearcher A2A Endpoint) class TaskInput(BaseModel): task_id: str instructions: str context: Optional[dict] None class TaskOutput(BaseModel): task_id: str status: str artifacts: Optional[list] None app.post(/a2a/task, response_modelTaskOutput) async def submit_task(task: TaskInput): # 这里真实场景会丢到队列里由 Worker 异步处理 artifacts [{name: research_note.md, content: f基于 instructions 生成的调研笔记}] return TaskOutput(task_idtask.task_id, statuscompleted, artifactsartifacts)在实际集群中我不会让请求直接同步处理完而是先返回状态 working再通过回调或轮询更新状态。A2A 的价值就在于它把这些“派活”和“回传”的流程规范好了不同语言的 Agent 也能互相协作。比如 Java 写的 Agent 和 Python 写的 Agent只要都遵守 A2A 的对象结构就能互通。现在社区已经有 A2A Python SDK、Java SDK甚至 Spring 生态里也有对应的集成方案这就是热词里为什么大量出现“a2a spring”的原因。2.3 Skills给智能体装“肌肉记忆”Skills 这个概念比 MCP 和 A2A 更轻也更像“内容包”。一个 Skill 通常是一个目录里面有 SKILL.md技能描述、少量示例、模板、脚本。它们共同描述“遇到某类任务时Agent 应该怎么处理”。在社区里比较火的 SKILL.md 格式一般长这样--- name: report_writer description: 编写结构化 Markdown 行业报告的技能适用于数据分析、竞品研究和调研总结 --- ## When to use - 当需要把多个来源的信息整理成一份可交付报告时 ## Steps 1. 提取所有输入资料中的关键结论按主题聚类。 2. 使用模板 templates/report.md 生成初稿。 3. 在初稿中补充数据对比和事实来源。 4. 对报告标题、目录、结论进行一致性检查。 ## Examples - 输入三篇竞品分析文章 两份数据库记录 - 输出一份包含摘要、市场分析、竞品矩阵、行动建议的 Markdown 报告Agent 通过一个 Skill Manager 扫描 skills 目录读 frontmatter 里的 name 和 description然后在触发任务时做语义匹配。有了 Skills你就不需要为每个新任务重写一大段 prompt而是把“怎么做”沉淀为可复用、可版本化的文件。有人问 Skills 和 MCP 到底有什么区别我的理解是MCP 是“运行时能力”比如一个搜索工具它必须连接、调用、返回结果Skills 是“静态知识/流程”更像手册不一定要起一个服务。但两者可以配合Skill 里可以包含一段说明指引 Agent 去调用某个 MCP 工具。热词里那些“superpower skills”“claude agent skills”之所以火就是因为它让 Agent 的学习成本大幅下降很多以前要写在 prompt 里的约束现在变成了结构化的技能包。3. 实战搭建一个可运行的 DeepAgents 多智能体集群3.1 架构选型与运行环境准备我按以下方案搭建第一版集群成本低、好调试后续可平滑升级组件选型说明模型网关LiteLLM / one-api统一多种模型接口设置路由和限额Agent 运行时Python 3.11 FastAPI轻量SDK 生态好MCP Server官方 Python SDK自己写的业务工具比如搜索、数据库A2A 端点FastAPI 独立路由每个 Agent 一个 /a2a/task 入口Skill StoreGit 仓库按目录存放 SKILL.md 和脚本编排器一个 Python 进程负责任务拆解、派发、汇总我推荐先用 uv 管理 Python 依赖因为它比 pip 快太多。核心依赖其实很简单uv venv .venv source .venv/bin/activate uv pip install mcp[cli] fastapi uvicorn httpx openai pydantic python-frontmatter如果后续要接 Docker 部署每个 Worker 镜像里放好 Python 环境和 skills 目录用同一套配置中心给不同容器发环境变量即可。3.2 用 MCP 把工具接入 Agent我们在实际项目里最常用的是“网页搜索 数据库查询”两个工具。先用 MCP 定义两个 Toolfrom mcp.server.fastmcp import FastMCP mcp FastMCP(data-tools) mcp.tool() async def search_web(query: str, limit: int 5) - list[dict]: 执行网页搜索返回标题、链接、摘要和发布时间 # 这里接真实的搜索服务或爬虫 return [{title: 示例标题, link: https://example.com, snippet: 摘要内容, date: 2025-01-01}] mcp.tool() def query_database(sql: str) - list[dict]: 执行只读 SQL 查询返回查询结果 # 实际中会走连接池并校验 SQL 是否只读 return [{row: 1}]注意我给第二个工具加了一个限制只读 SQL。MCP 的 Tools 没有强制鉴权能力所以安全的边界需要在业务代码里自己做后面会再展开。Agent 侧通过 MCP Client 连接这个 Server。如果 Server 走 stdio那么 Agent 启动时会作为子进程拉起它如果走 HTTP就用流式连接。我用官方 SDK 连接的示例from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client server_params StdioServerParameters(commandpython, args[data_tools_server.py]) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: tools await session.list_tools() result await session.call_tool(search_web, {query: MCP协议原理})这段代码是连接 MCP 的最小骨架。有了它Agent 就能动态获取工具列表和参数不需要事先知道每个函数怎么调。这一步很关键集群里每个 Worker 接入工具的方式完全统一后续替换数据源或增加新工具只需要改 MCP Server 本身。3.3 把 Skills 注入到 AgentSkills 的加载不需要复杂框架但要规范。我用的目录结构是skills/ report_writer/ SKILL.md templates/ report.md scripts/ format_report.py data_mining/ SKILL.md examples/ example_input.md在 Agent 启动时我会写一个轻量 SkillManagerimport frontmatter from pathlib import Path class SkillManager: def __init__(self, skills_dir: Path): self.skills [] for skill_dir in skills_dir.iterdir(): md skill_dir / SKILL.md if md.exists(): post frontmatter.load(md) self.skills.append({ name: post[name], description: post[description], content: post.content, path: skill_dir, }) def find_skill(self, task: str): # 这里可以用 embedding 或者关键词匹配 for skill in self.skills: if skill[name] in task or any(word in task for word in [报告, 调研]): return skill return None然后当 Agent 收到任务时会先调用find_skill如果命中就把 SKILL.md 里的步骤和模板一并塞进 prompt。我在实践里发现Skills 的匹配策略不要做得太重先用关键词、再用少量样本测试待稳定后换成 embedding 检索也不迟。目前 VS Code、IntelliJ 等 IDE 也陆续支持导入 skills 和 MCP Server热词里大量“codex skills”“idea 使用 skills”说明这个方向已经成了主流工作流的一部分。3.4 用 A2A 串起集群协作集群里最关键的就是 A2A 派发。我在上一节给的 FastAPI 端点只是接收接口真正跑起来还需要一张 Agent Card. 我习惯在启动时加载如下 JSON{ name: researcher, description: 负责资料检索、数据采集和事实核查, url: http://localhost:8001/a2a/task, skills: [data_mining], capabilities: [search_web, query_database] }编配器在启动时会到所有 Worker 上 GET /a2a/card拉取到这张卡片并缓存成一张“能力路由表”。当用户提交任务后编配器先把任务拆成多个子任务再看每个子任务需要什么能力和 skill最终决定派给谁。这里我给出一个简化版的任务派发器使用 httpx 请求 A2A 端点import httpx import asyncio AGENT_CARDS [ {name: researcher, url: http://localhost:8001/a2a/task, skills: [data_mining]}, {name: writer, url: http://localhost:8002/a2a/task, skills: [report_writer]}, ] async def dispatch(agent_name: str, task: dict, timeout: float 30.0): card next(a for a in AGENT_CARDS if a[name] agent_name) payload { task_id: task[id], instructions: task[instructions], context: task.get(context, {}), } async with httpx.AsyncClient(timeouttimeout) as client: resp await client.post(card[url], jsonpayload) resp.raise_for_status() return resp.json()这只是同步阻塞版本实际生产里我会让任务返回 working 状态然后用后台任务轮询或 WebSocket 接收完成通知。A2A 任务的状态机建议至少包含submitted、working、completed、failed、canceled。所有分支都要有超时我一般设置单次 Agent 调用不超过 60 秒超过后进入重试重试两次仍失败就把失败原因交给 Planner 重新拆任务。3.5 一个完整示例自动生成行业调研报告我们把上面所有模块拼起来跑一个真实任务生成“2025 年企业级 MCP 中间件选型调研报告”。流程是这样的用户把任务交给 Orchestrator。Orchestrator 内部使用“规划-反思”循环生成 3 个子任务市场概览、竞品功能对比、实测性能数据收集。编配器发现 researcher Agent 带有 data_mining skill且具备 search_web 和 query_database 能力所以前两个子任务派给它第三个子任务需要调用性能测试工具派给另一个 tester Agent。researcher 通过 MCP search_web 检索了几篇资料用 data_mining skill 整理成结构笔记返回 artifacts。tester Agent 调用数据库工具拿到测试数据并汇总成表格。全部子任务完成后Orchestrator 把 artifacts 打包给 writer Agentwriter 加载 report_writer skill按模板生成最终 Markdown 报告。最后由 Reviewer Agent 检查报告引用和数据是否完整不达标则打回重写。运行日志里大致能看到这种内容[orchestrator] task split - 3 subtasks [orchestrator] dispatch market_overview to researcher [researcher] skill matched: data_mining [researcher] mcp call search_web(queryMCP 中间件 功能对比) [researcher] writing artifact: research_note.md [orchestrator] receiving artifact from researcher [orchestrator] dispatch writer to generate final report [writer] skill matched: report_writer [writer] output: 2025-mcp-middleware-report.md [orchestrator] all subtasks complete. final_output size: 128KB这个示例虽然简单但它把 DeepAgents 的规划、MCP 的工具接入、A2A 的任务派发、Skills 的知识注入全部串在了一起。你可以把它当成一个最小闭环基于它往里面加记忆、更复杂的工具链、或者更多 Worker 类型。4. 集群协同开发的实践要点4.1 权限边界与安全不能后补多智能体集群里最容易被忽视的是安全问题。MCP 工具一旦暴露给 AgentAgent 的 prompt 就可能被“注入”。比如网页搜索工具返回的内容里藏了恶意指令Agent 读了之后可能去执行不该执行的动作。所以我在所有 MCP Server 上加了三道防线工具注册表里给每个 Tool 标注 access scope例如 read-only、internal-only、admin-only。对于外部内容Agent 接收后要经过“信息清洗”只提取事实字段忽略命令性文本。更稳妥的是用独立的小模型先做一次内容过滤。对于危险操作写数据库、执行命令MCP Server 内部强制要求审批 token不能由 Agent 直接调用。A2A 方面Agent Card 里应该携带安全声明包括认证方式和信任域。我在内网环境里最常用的还是 mTLS或者简单的服务间 token header。不要把 A2A 端点裸奔在公网。社区里讨论“codex 接入 figma MCP 怎么授权”本质就是 OAuth 授权流程Agent 需要以用户身份获取一个短期 token再通过 MCP 的 auth 配置传给 Server。这个流程要提前在架构里预留否则后期改造很痛苦。4.2 编排策略与记忆管理编配方式我建议先中央后自治。中央编配器的好处是逻辑简单、易于追踪坏处是可能成为瓶颈。当 Worker 数量超过 10 个后我会把“任务分发”演化成“能力路由 工作流”每个 Worker 从共享队列拉取子任务完成后把结果写入共享存储编配器只负责定义任务依赖和验收标准。多智能体集群还必须有记忆层。我的方案分三级会话级记忆每个 Agent 内部存储对话历史控制在最近 10 轮左右。集群级记忆一个 KV 存储记录“任务 ID - 状态 - artifact 引用”。长期记忆把重要的方法论和步骤写成 Skill沉淀到 Skill Store。这样下次遇到类似任务Agent 不需要从头摸索。层级分离很重要。如果你把所有 Agent 的上下文都塞到同一个向量库里检索会变得非常嘈杂。我建议集群级记忆只存结构化元数据让 Agent“知道有这段内容”而不是把内容本身灌进上下文。4.3 可观测性与调试多智能体集群的调试比单 Agent 难得多因为一次用户请求可能在一分钟内触发几十次模型调用和工具调用。我习惯给所有跨 Agent 调用打上同一个 trace_id从入口编配器生成传给所有子任务。日志统一输出成 JSON包含 trace_id、agent_id、event_type、duration_ms 等字段。在没有专门链路追踪系统的情况下我用结构化日志也能定位问题。示例日志{trace_id: a1b2c3, agent: researcher, event: mcp_call_start, tool: search_web, time: 1720000000} {trace_id: a1b2c3, agent: researcher, event: mcp_call_end, tool: search_web, duration_ms: 1200, result_size: 3456}只要日志格式统一用grep trace_id就能看完整链路。等集群规模大了再接入 OpenTelemetry 和 Jaeger 也不迟但基础规范必须一开始就定好。4.4 性能和成本控制多智能体集群的成本比单 Agent 高很多原因是同样的信息会在多个 Agent 之间流转。如果每个中间产物都直接压进 prompt成本会指数级增长。我总结的几条控制策略中间产物尽量以文件或对象引用传递而不是把原始文本塞进 A2A 消息。模型选型分层简单分类用轻量模型复杂推理用旗舰模型不要让所有 Agent 都调用最强模型。对重复度高的 MCP 工具结果做缓存比如同一关键词的搜索在 10 分钟内直接命中缓存。控制并行度我实测并行调用超过 8 个时收益递减且容易触发限流。现在默认并发 5失败了再重试。性能方面A2A 和 MCP 的传输本身很轻真正的瓶颈往往出现在模型推理和外部 API 上。遇到慢任务时优先检查外部依赖而不是优化协议。5. 常见问题速查与排查实录5.1 我踩过的坑和对应解法这里整理一张速查表方便你对照排查现象可能原因解决方法A2A 握手失败Agent Card URL 不可达或认证头缺失用 curl 单独测试 /a2a/card 端点MCP 连接成功但 tool call 超时Server 端逻辑阻塞或外部服务等待把 Server 内部逻辑改成异步单独调外部服务Skills 没有生效SKILL.md frontmatter 格式错误校验 name/description 字段路径不能有中文Agent 来回重复派单没有任务 hops 限制在 Task 中加入 hop_limit超过则丢弃上下文爆炸把 artifact 全量塞进 prompt只传摘要和文件路径按需读取MCP 和 A2A 都接好后模型还是不调用工具模型指令里没写清楚 tool 的调用规则在 system prompt 中明确“优先调用 MCP 工具”还有一个热词里经常被问到的“browser use MCP 跟 playwright MCP 有什么区别”我的理解是browser use 类的 MCP 工具偏向“自然语言驱动浏览器”适合读网页、点按钮、抓取页面playwright MCP 偏向“代码级控制浏览器”适合精确的自动化测试和复杂交互。如果只是给 Agent 加一个看网页的能力browser use 更省事如果要跑回归测试playwright 更合适。选型时多考虑你是要“快”还是“稳”。5.2 排查五步法当集群整体出了问题我有一套固定的排查顺序看结构化日志定位 trace_id 断了哪一环。用 curl 单测 A2A 端点和 MCP 工具确认协议层正常。用最小任务复现比如只让 Planner 拆一个子任务不触发多 Agent。检查模型返回看是不是因为输出格式问题导致派发失败。检查版本兼容性MCP、A2A 的 SDK 升级很快锁定版本不要自动升到 latest。这套流程虽然朴素但能解决 80% 的问题。尤其是在初期不要急着上监控系统先把日志规范做起来效率反而更高。6. 最后的经验心得如果你问我搭建这个多智能体集群最大的体会是什么我会说先把单点做稳再谈协同。我最初的错误就是直接搭了 6 个 Agent结果每个 Agent 都有各自的 MCP 连接问题、Skill 路径问题调试时根本分不清是协议出了问题还是模型理解出了问题。后来我把修改方向改成“先单后多”先跑通一个 Worker 的 MCP Skills A2A 接收端点再复制扩展出其他 Worker整个集群的稳定性一下子就上来了。另一个经验是不要把“多智能体”当作终极目标。很多时候一个 Agent 一个好用的 MCP 工具 几个精心设计的 Skills就能解决实际问题。多智能体集群适合的是“真的有多角色、多任务、跨领域协作需求”的场景。如果任务简单引入集群只会增加延迟和不确定性。后续如果你想扩展这个方案有两个方向我建议优先尝试一是把 Skill Store 从本地目录升级成远程 Git 仓库甚至接进社区市场这样新 Agent 可以秒级获得技能二是给集群加一层向量记忆让 Planner 在拆任务时能参考历史相似任务的做法。这个小技巧配合上 MCP 和 A2A可以让多智能体集群越来越接近一支真正的虚拟团队。 DeepAgents+MCP+A2A+Skills：构建多智能体集群实战指南
如果你最近在搞 AI Agent一定对 DeepAgents、MCP、A2A、Skills 这四个词不陌生。它们的共同点都在回答同一个问题当一个 Agent 不够用的时候我们怎么用一套清晰可扩展的架构把它变成一支能各自干活、又能互相协作的多智能体集群。这篇文章我会从架构设计讲到底层协议拆解再给出一套可以直接落地的多智能体集群搭建过程最后聊一聊我实际跑集群时踩过的坑。适合正在做 Agent 编排、工具接入、或者想把多个大模型节点串成流水线的朋友内容偏工程实践但我会尽量把概念讲明白新手也能跟着操作。我先把话说在前面DeepAgents 不是某个特定的开源项目而是一套“深度任务理解 规划 反思 行动”的智能体实现思路MCP 解决的是 Agent 和工具之间要如何标准化接入A2A 解决的是 Agent 和 Agent 之间要如何对话和交接Skills 则是给 Agent 注入可复用的“工作手册”。把它们拼起来才算是真正意义上的多智能体集群架构。下面开始。1. 多智能体集群架构先把“集群”这件事想清楚1.1 从单 Agent 到多 Agent瓶颈在协作而不在数量很多团队一开始的想法是一个 Agent 能力不够那就多写几个 Agent让它们各自处理一个子任务。真做起来就发现问题了。Agent 之间没有统一通信语言A 的输出没法直接变成 B 的输入工具接入方式五花八门有的用函数调用有的直接让模型输出 JSON每个 Agent 的 prompt 和技能逻辑互相复制改一处就要连带改好几个文件。我在一开始搭建多智能体集群的时候也犯过“堆数量”的错误。比如做研究型任务安排了 5 个 Agent 同时去搜资料结果它们各自返回了格式完全不同的笔记没有一个统一的上下文结构最终汇总时人还得手工整理。后来我意识到多智能体的核心不是多而是“分得清、传得动、合得拢”。要分就要有任务拆解和角色定义要传就要有统一的传输协议要合就要有明确的结果格式和验收标准。这才是我理解的多智能体集群架构一组自治的 Agent 节点通过标准化协议连接、协同形成一个整体系统对外服务。再具体一点一个可运行的多智能体集群通常由几个角色组成Planner 负责拆解任务Researcher 负责检索信息Writer 负责产出内容Reviewer 负责检查质量。它们不是随便几个“LLM 调用”堆在一起而是每个角色都承担明确的职责通过消息或任务对象互相交接。这种设计的好处是任何单一节点可以被替换、升级或水平扩展而不影响整个集群的对外行为。1.2 四个关键词各自负责哪一层我们经常听到“DeepAgents MCP A2A Skills”这种组合但很少有人解释清楚它们之间是什么关系。我用一个团队来类比DeepAgents 是项目经理的思维方式。它强调 Agent 面对复杂任务时要先拆解、再执行、然后反思修正。这是应用层的行为范式决定 Agent 怎么“思考”。MCP 是办公室里统一的插座标准。Agent 要调用数据库、网页搜索、文件系统等工具如果不统一接口每接一个新工具就要改一遍 Agent 代码。MCP 把这些能力统一成“即插即用”的协议。A2A 是团队之间的沟通契约。A2A 解决的是“Agent 与 Agent 怎么发现彼此、怎么派活、怎么回传结果”。它让不同团队、不同框架开发的 Agent 可以被统一调度。Skills 是给员工的新人培训手册。它把某个任务怎么做的步骤、模板、示例、脚本打包到一个目录里Agent 遇到对应场景时按手册执行。这四个层次可以互相叠加。比如一个 Researcher Agent它内部用 DeepAgents 的“规划-行动-反思”循环来思考它通过 MCP 调用网络搜索工具它把自己的能力声明成 A2A 的 Agent Card供编配器调用它加载了一个“高效检索”的 Skill知道如何整理搜索结果。这就是一套完整的多智能体工程化组合。1.3 一张“集群拓扑”长什么样先说拓扑后面所有实操都围绕着它。我建议的第一版集群是“中心化编配 分布式执行”一个 Orchestrator编配器作为入口负责任务理解、拆解和调度若干个 Worker Agent 作为执行节点各自暴露 A2A 端点所有 Worker 通过 MCP 客户端连接所需工具Skills 作为局部文件目录或远程仓库挂在每个 Worker 上。用户请求进入 Orchestrator 后拆出多个子任务按依赖关系发给不同 WorkerWorker 完成后再把产物回传最终由 Orchestrator 汇总输出。这个拓扑很朴素甚至有点“过时”但它能帮你把每个环节的职责理清。等跑通了再根据性能要求做去中心化改造也不迟。2. MCP、A2A、Skills 三件套的底层逻辑2.1 MCP 是“万能工具插座”不是又一种 RPC我第一次看到 MCPModel Context Protocol时第一反应是这不就是个远程调用协议吗后来深入用才明白它和普通 RPC 有个本质区别它面向的是大模型应用定义了一套“模型上下文”的边界。在 MCP 架构里有 Host、Client、Server 三层。Host 是大模型应用本身Client 负责维护与 Server 的连接Server 把能力暴露成三种原语Tools、Resources、Prompts。Tools 是让模型执行动作Resources 是让模型读取信息Prompts 是给模型复用预设提示词。你发现没有MCP 强调的是“模型上下文”不是单纯的函数调用。比如一个搜索工具它返回的不仅仅是字符串还应该有足够的上下文线索source、date、confidence这样才能让模型判断信息是否可信。因此 MCP 非常适合用于把企业内部数据库、网页搜索、代码仓库、设计稿等能力统一接入 Agent。我们用一段极简的 Python 示例来说清楚 MCP Server 的形态。假设我们写一个能查询用户信息的小服务使用官方 Python SDKfrom mcp.server.fastmcp import FastMCP mcp FastMCP(user-service) mcp.tool() def get_user_info(user_id: str) - dict: 根据 user_id 查询用户基础信息 return {user_id: user_id, name: 张三, role: admin, email: zhangsanexample.com} if __name__ __main__: mcp.run(transportstdio)这个 Server 通过 stdio 和 Host 进程通信协议消息是 JSON-RPC 2.0。Host 端的 Agent 只需要一个 MCP Client就能动态发现 get_user_info 这个工具以及它的参数结构。这种“能力即插即用”的体验和给电脑插 USB 设备很像所以我说它是“万能工具插座”。还要澄清一个热词里经常看到的疑问MCP 是软件协议还是硬件协议它和物理层硬件协议完全是两码事它运行在应用层描述的是大模型应用和外部工具之间的语义交互方式你可以理解为一份“软件接口契约”而不是网线或 USB 接口那种物理协议。2.2 A2A 是智能体之间的“协作语言”如果说 MCP 解决的是“人Agent和工具”的关系那 A2AAgent-to-Agent解决的就是“人和人Agent 和 Agent”的关系。A2A 的核心思想是让不同 Agent 通过一组标准对象进行交流而不是互相写死对方的私有接口。协议里最重要的几个对象是Agent Card一个 Agent 的公开身份和能力描述通常是一个 JSON 文件写清楚名称、说明、能力、认证方式、端点地址。Task表示一个需要完成的请求有 task_id、status、input、artifacts 等字段。MessageAgent 之间传输的具体信息可以是文本、对象、文件引用。Artifact任务产出的结果可以是报告、数据表、图片等。我自己的实现里会为每个 Worker 暴露一个 HTTP 端点比如 /a2a/task。当 Orchestrator 要派活时它会做两件事先根据 Agent Card 决定选哪个 Agent然后发送一个 Task 请求。Worker 接收 Task 后在自己的循环中处理并通过 PATCH 或者 WebSocket 持续更新状态。最后把结果 Artifact 返回。下面是一段用 FastAPI 写的极简 A2A 接收端点from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Optional app FastAPI(titleResearcher A2A Endpoint) class TaskInput(BaseModel): task_id: str instructions: str context: Optional[dict] None class TaskOutput(BaseModel): task_id: str status: str artifacts: Optional[list] None app.post(/a2a/task, response_modelTaskOutput) async def submit_task(task: TaskInput): # 这里真实场景会丢到队列里由 Worker 异步处理 artifacts [{name: research_note.md, content: f基于 instructions 生成的调研笔记}] return TaskOutput(task_idtask.task_id, statuscompleted, artifactsartifacts)在实际集群中我不会让请求直接同步处理完而是先返回状态 working再通过回调或轮询更新状态。A2A 的价值就在于它把这些“派活”和“回传”的流程规范好了不同语言的 Agent 也能互相协作。比如 Java 写的 Agent 和 Python 写的 Agent只要都遵守 A2A 的对象结构就能互通。现在社区已经有 A2A Python SDK、Java SDK甚至 Spring 生态里也有对应的集成方案这就是热词里为什么大量出现“a2a spring”的原因。2.3 Skills给智能体装“肌肉记忆”Skills 这个概念比 MCP 和 A2A 更轻也更像“内容包”。一个 Skill 通常是一个目录里面有 SKILL.md技能描述、少量示例、模板、脚本。它们共同描述“遇到某类任务时Agent 应该怎么处理”。在社区里比较火的 SKILL.md 格式一般长这样--- name: report_writer description: 编写结构化 Markdown 行业报告的技能适用于数据分析、竞品研究和调研总结 --- ## When to use - 当需要把多个来源的信息整理成一份可交付报告时 ## Steps 1. 提取所有输入资料中的关键结论按主题聚类。 2. 使用模板 templates/report.md 生成初稿。 3. 在初稿中补充数据对比和事实来源。 4. 对报告标题、目录、结论进行一致性检查。 ## Examples - 输入三篇竞品分析文章 两份数据库记录 - 输出一份包含摘要、市场分析、竞品矩阵、行动建议的 Markdown 报告Agent 通过一个 Skill Manager 扫描 skills 目录读 frontmatter 里的 name 和 description然后在触发任务时做语义匹配。有了 Skills你就不需要为每个新任务重写一大段 prompt而是把“怎么做”沉淀为可复用、可版本化的文件。有人问 Skills 和 MCP 到底有什么区别我的理解是MCP 是“运行时能力”比如一个搜索工具它必须连接、调用、返回结果Skills 是“静态知识/流程”更像手册不一定要起一个服务。但两者可以配合Skill 里可以包含一段说明指引 Agent 去调用某个 MCP 工具。热词里那些“superpower skills”“claude agent skills”之所以火就是因为它让 Agent 的学习成本大幅下降很多以前要写在 prompt 里的约束现在变成了结构化的技能包。3. 实战搭建一个可运行的 DeepAgents 多智能体集群3.1 架构选型与运行环境准备我按以下方案搭建第一版集群成本低、好调试后续可平滑升级组件选型说明模型网关LiteLLM / one-api统一多种模型接口设置路由和限额Agent 运行时Python 3.11 FastAPI轻量SDK 生态好MCP Server官方 Python SDK自己写的业务工具比如搜索、数据库A2A 端点FastAPI 独立路由每个 Agent 一个 /a2a/task 入口Skill StoreGit 仓库按目录存放 SKILL.md 和脚本编排器一个 Python 进程负责任务拆解、派发、汇总我推荐先用 uv 管理 Python 依赖因为它比 pip 快太多。核心依赖其实很简单uv venv .venv source .venv/bin/activate uv pip install mcp[cli] fastapi uvicorn httpx openai pydantic python-frontmatter如果后续要接 Docker 部署每个 Worker 镜像里放好 Python 环境和 skills 目录用同一套配置中心给不同容器发环境变量即可。3.2 用 MCP 把工具接入 Agent我们在实际项目里最常用的是“网页搜索 数据库查询”两个工具。先用 MCP 定义两个 Toolfrom mcp.server.fastmcp import FastMCP mcp FastMCP(data-tools) mcp.tool() async def search_web(query: str, limit: int 5) - list[dict]: 执行网页搜索返回标题、链接、摘要和发布时间 # 这里接真实的搜索服务或爬虫 return [{title: 示例标题, link: https://example.com, snippet: 摘要内容, date: 2025-01-01}] mcp.tool() def query_database(sql: str) - list[dict]: 执行只读 SQL 查询返回查询结果 # 实际中会走连接池并校验 SQL 是否只读 return [{row: 1}]注意我给第二个工具加了一个限制只读 SQL。MCP 的 Tools 没有强制鉴权能力所以安全的边界需要在业务代码里自己做后面会再展开。Agent 侧通过 MCP Client 连接这个 Server。如果 Server 走 stdio那么 Agent 启动时会作为子进程拉起它如果走 HTTP就用流式连接。我用官方 SDK 连接的示例from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client server_params StdioServerParameters(commandpython, args[data_tools_server.py]) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: tools await session.list_tools() result await session.call_tool(search_web, {query: MCP协议原理})这段代码是连接 MCP 的最小骨架。有了它Agent 就能动态获取工具列表和参数不需要事先知道每个函数怎么调。这一步很关键集群里每个 Worker 接入工具的方式完全统一后续替换数据源或增加新工具只需要改 MCP Server 本身。3.3 把 Skills 注入到 AgentSkills 的加载不需要复杂框架但要规范。我用的目录结构是skills/ report_writer/ SKILL.md templates/ report.md scripts/ format_report.py data_mining/ SKILL.md examples/ example_input.md在 Agent 启动时我会写一个轻量 SkillManagerimport frontmatter from pathlib import Path class SkillManager: def __init__(self, skills_dir: Path): self.skills [] for skill_dir in skills_dir.iterdir(): md skill_dir / SKILL.md if md.exists(): post frontmatter.load(md) self.skills.append({ name: post[name], description: post[description], content: post.content, path: skill_dir, }) def find_skill(self, task: str): # 这里可以用 embedding 或者关键词匹配 for skill in self.skills: if skill[name] in task or any(word in task for word in [报告, 调研]): return skill return None然后当 Agent 收到任务时会先调用find_skill如果命中就把 SKILL.md 里的步骤和模板一并塞进 prompt。我在实践里发现Skills 的匹配策略不要做得太重先用关键词、再用少量样本测试待稳定后换成 embedding 检索也不迟。目前 VS Code、IntelliJ 等 IDE 也陆续支持导入 skills 和 MCP Server热词里大量“codex skills”“idea 使用 skills”说明这个方向已经成了主流工作流的一部分。3.4 用 A2A 串起集群协作集群里最关键的就是 A2A 派发。我在上一节给的 FastAPI 端点只是接收接口真正跑起来还需要一张 Agent Card. 我习惯在启动时加载如下 JSON{ name: researcher, description: 负责资料检索、数据采集和事实核查, url: http://localhost:8001/a2a/task, skills: [data_mining], capabilities: [search_web, query_database] }编配器在启动时会到所有 Worker 上 GET /a2a/card拉取到这张卡片并缓存成一张“能力路由表”。当用户提交任务后编配器先把任务拆成多个子任务再看每个子任务需要什么能力和 skill最终决定派给谁。这里我给出一个简化版的任务派发器使用 httpx 请求 A2A 端点import httpx import asyncio AGENT_CARDS [ {name: researcher, url: http://localhost:8001/a2a/task, skills: [data_mining]}, {name: writer, url: http://localhost:8002/a2a/task, skills: [report_writer]}, ] async def dispatch(agent_name: str, task: dict, timeout: float 30.0): card next(a for a in AGENT_CARDS if a[name] agent_name) payload { task_id: task[id], instructions: task[instructions], context: task.get(context, {}), } async with httpx.AsyncClient(timeouttimeout) as client: resp await client.post(card[url], jsonpayload) resp.raise_for_status() return resp.json()这只是同步阻塞版本实际生产里我会让任务返回 working 状态然后用后台任务轮询或 WebSocket 接收完成通知。A2A 任务的状态机建议至少包含submitted、working、completed、failed、canceled。所有分支都要有超时我一般设置单次 Agent 调用不超过 60 秒超过后进入重试重试两次仍失败就把失败原因交给 Planner 重新拆任务。3.5 一个完整示例自动生成行业调研报告我们把上面所有模块拼起来跑一个真实任务生成“2025 年企业级 MCP 中间件选型调研报告”。流程是这样的用户把任务交给 Orchestrator。Orchestrator 内部使用“规划-反思”循环生成 3 个子任务市场概览、竞品功能对比、实测性能数据收集。编配器发现 researcher Agent 带有 data_mining skill且具备 search_web 和 query_database 能力所以前两个子任务派给它第三个子任务需要调用性能测试工具派给另一个 tester Agent。researcher 通过 MCP search_web 检索了几篇资料用 data_mining skill 整理成结构笔记返回 artifacts。tester Agent 调用数据库工具拿到测试数据并汇总成表格。全部子任务完成后Orchestrator 把 artifacts 打包给 writer Agentwriter 加载 report_writer skill按模板生成最终 Markdown 报告。最后由 Reviewer Agent 检查报告引用和数据是否完整不达标则打回重写。运行日志里大致能看到这种内容[orchestrator] task split - 3 subtasks [orchestrator] dispatch market_overview to researcher [researcher] skill matched: data_mining [researcher] mcp call search_web(queryMCP 中间件 功能对比) [researcher] writing artifact: research_note.md [orchestrator] receiving artifact from researcher [orchestrator] dispatch writer to generate final report [writer] skill matched: report_writer [writer] output: 2025-mcp-middleware-report.md [orchestrator] all subtasks complete. final_output size: 128KB这个示例虽然简单但它把 DeepAgents 的规划、MCP 的工具接入、A2A 的任务派发、Skills 的知识注入全部串在了一起。你可以把它当成一个最小闭环基于它往里面加记忆、更复杂的工具链、或者更多 Worker 类型。4. 集群协同开发的实践要点4.1 权限边界与安全不能后补多智能体集群里最容易被忽视的是安全问题。MCP 工具一旦暴露给 AgentAgent 的 prompt 就可能被“注入”。比如网页搜索工具返回的内容里藏了恶意指令Agent 读了之后可能去执行不该执行的动作。所以我在所有 MCP Server 上加了三道防线工具注册表里给每个 Tool 标注 access scope例如 read-only、internal-only、admin-only。对于外部内容Agent 接收后要经过“信息清洗”只提取事实字段忽略命令性文本。更稳妥的是用独立的小模型先做一次内容过滤。对于危险操作写数据库、执行命令MCP Server 内部强制要求审批 token不能由 Agent 直接调用。A2A 方面Agent Card 里应该携带安全声明包括认证方式和信任域。我在内网环境里最常用的还是 mTLS或者简单的服务间 token header。不要把 A2A 端点裸奔在公网。社区里讨论“codex 接入 figma MCP 怎么授权”本质就是 OAuth 授权流程Agent 需要以用户身份获取一个短期 token再通过 MCP 的 auth 配置传给 Server。这个流程要提前在架构里预留否则后期改造很痛苦。4.2 编排策略与记忆管理编配方式我建议先中央后自治。中央编配器的好处是逻辑简单、易于追踪坏处是可能成为瓶颈。当 Worker 数量超过 10 个后我会把“任务分发”演化成“能力路由 工作流”每个 Worker 从共享队列拉取子任务完成后把结果写入共享存储编配器只负责定义任务依赖和验收标准。多智能体集群还必须有记忆层。我的方案分三级会话级记忆每个 Agent 内部存储对话历史控制在最近 10 轮左右。集群级记忆一个 KV 存储记录“任务 ID - 状态 - artifact 引用”。长期记忆把重要的方法论和步骤写成 Skill沉淀到 Skill Store。这样下次遇到类似任务Agent 不需要从头摸索。层级分离很重要。如果你把所有 Agent 的上下文都塞到同一个向量库里检索会变得非常嘈杂。我建议集群级记忆只存结构化元数据让 Agent“知道有这段内容”而不是把内容本身灌进上下文。4.3 可观测性与调试多智能体集群的调试比单 Agent 难得多因为一次用户请求可能在一分钟内触发几十次模型调用和工具调用。我习惯给所有跨 Agent 调用打上同一个 trace_id从入口编配器生成传给所有子任务。日志统一输出成 JSON包含 trace_id、agent_id、event_type、duration_ms 等字段。在没有专门链路追踪系统的情况下我用结构化日志也能定位问题。示例日志{trace_id: a1b2c3, agent: researcher, event: mcp_call_start, tool: search_web, time: 1720000000} {trace_id: a1b2c3, agent: researcher, event: mcp_call_end, tool: search_web, duration_ms: 1200, result_size: 3456}只要日志格式统一用grep trace_id就能看完整链路。等集群规模大了再接入 OpenTelemetry 和 Jaeger 也不迟但基础规范必须一开始就定好。4.4 性能和成本控制多智能体集群的成本比单 Agent 高很多原因是同样的信息会在多个 Agent 之间流转。如果每个中间产物都直接压进 prompt成本会指数级增长。我总结的几条控制策略中间产物尽量以文件或对象引用传递而不是把原始文本塞进 A2A 消息。模型选型分层简单分类用轻量模型复杂推理用旗舰模型不要让所有 Agent 都调用最强模型。对重复度高的 MCP 工具结果做缓存比如同一关键词的搜索在 10 分钟内直接命中缓存。控制并行度我实测并行调用超过 8 个时收益递减且容易触发限流。现在默认并发 5失败了再重试。性能方面A2A 和 MCP 的传输本身很轻真正的瓶颈往往出现在模型推理和外部 API 上。遇到慢任务时优先检查外部依赖而不是优化协议。5. 常见问题速查与排查实录5.1 我踩过的坑和对应解法这里整理一张速查表方便你对照排查现象可能原因解决方法A2A 握手失败Agent Card URL 不可达或认证头缺失用 curl 单独测试 /a2a/card 端点MCP 连接成功但 tool call 超时Server 端逻辑阻塞或外部服务等待把 Server 内部逻辑改成异步单独调外部服务Skills 没有生效SKILL.md frontmatter 格式错误校验 name/description 字段路径不能有中文Agent 来回重复派单没有任务 hops 限制在 Task 中加入 hop_limit超过则丢弃上下文爆炸把 artifact 全量塞进 prompt只传摘要和文件路径按需读取MCP 和 A2A 都接好后模型还是不调用工具模型指令里没写清楚 tool 的调用规则在 system prompt 中明确“优先调用 MCP 工具”还有一个热词里经常被问到的“browser use MCP 跟 playwright MCP 有什么区别”我的理解是browser use 类的 MCP 工具偏向“自然语言驱动浏览器”适合读网页、点按钮、抓取页面playwright MCP 偏向“代码级控制浏览器”适合精确的自动化测试和复杂交互。如果只是给 Agent 加一个看网页的能力browser use 更省事如果要跑回归测试playwright 更合适。选型时多考虑你是要“快”还是“稳”。5.2 排查五步法当集群整体出了问题我有一套固定的排查顺序看结构化日志定位 trace_id 断了哪一环。用 curl 单测 A2A 端点和 MCP 工具确认协议层正常。用最小任务复现比如只让 Planner 拆一个子任务不触发多 Agent。检查模型返回看是不是因为输出格式问题导致派发失败。检查版本兼容性MCP、A2A 的 SDK 升级很快锁定版本不要自动升到 latest。这套流程虽然朴素但能解决 80% 的问题。尤其是在初期不要急着上监控系统先把日志规范做起来效率反而更高。6. 最后的经验心得如果你问我搭建这个多智能体集群最大的体会是什么我会说先把单点做稳再谈协同。我最初的错误就是直接搭了 6 个 Agent结果每个 Agent 都有各自的 MCP 连接问题、Skill 路径问题调试时根本分不清是协议出了问题还是模型理解出了问题。后来我把修改方向改成“先单后多”先跑通一个 Worker 的 MCP Skills A2A 接收端点再复制扩展出其他 Worker整个集群的稳定性一下子就上来了。另一个经验是不要把“多智能体”当作终极目标。很多时候一个 Agent 一个好用的 MCP 工具 几个精心设计的 Skills就能解决实际问题。多智能体集群适合的是“真的有多角色、多任务、跨领域协作需求”的场景。如果任务简单引入集群只会增加延迟和不确定性。后续如果你想扩展这个方案有两个方向我建议优先尝试一是把 Skill Store 从本地目录升级成远程 Git 仓库甚至接进社区市场这样新 Agent 可以秒级获得技能二是给集群加一层向量记忆让 Planner 在拆任务时能参考历史相似任务的做法。这个小技巧配合上 MCP 和 A2A可以让多智能体集群越来越接近一支真正的虚拟团队。