1. 从单体智能到协作网络多智能体系统到底在解决什么问题如果你最近在折腾 AI Agent大概率会有一种感觉单个 Agent 能做的事情很快就摸到天花板了。你给它接上工具、挂上知识库、写好提示词它能帮你查资料、写代码、跑命令但一旦任务变复杂——比如帮我分析这份需求文档拆出模块生成接口定义再写一版可运行的骨架代码最后跑一遍测试——单体 Agent 就开始顾此失彼。它要么在长上下文里迷失要么在工具调用链里绕圈要么干脆把中间步骤忘干净。这不是模型不够聪明的问题而是架构问题。一个 Agent 同时扮演需求分析师、架构师、程序员、测试工程师本质上是在让一个大脑同时处理四种截然不同的认知模式。人类团队不会这么干AI 系统也不该这么干。多智能体系统Multi-Agent System的核心思路就是把一个复杂任务拆成若干个职责单一、边界清晰的智能体让它们通过标准化的协议互相通信、分工协作。而 MCPModel Context Protocol和 A2AAgent-to-Agent这两个协议恰好分别解决了这个架构里最关键的两个问题Agent 怎么连接外部能力以及Agent 之间怎么互相说话。这一篇是系列第七篇前面几篇我们聊过 MCP 的基础接入、工具注册、上下文管理也聊过 A2A 的消息格式和任务编排。这一篇我想把视角拉高一点不再纠结单个 API 怎么调而是讲清楚当你真的要设计一个多智能体 AI 系统时MCP 和 A2A 各自应该放在架构的哪个位置它们怎么配合以及我在实际搭建过程中踩过的那些坑。适合读这篇的人已经了解 LLM 基础调用、写过至少一个能跑通的 Agent Demo、现在想往多智能体方向进阶的开发者。如果你还没写过 Agent建议先回去补一下工具调用和 ReAct 循环的基础不然这篇里的很多设计取舍你会觉得为什么要这么麻烦。2. MCP 与 A2A 的职责边界别把两个协议用混了2.1 MCP 解决的是Agent 与能力之间的接口问题MCP 的本质是给 LLM 定义了一套标准化的能力接入方式。在没有 MCP 之前你每接一个工具就要写一套 function calling 的 schema每换一个模型schema 格式可能还要改。MCP 把这个过程抽象成了 Server 和 Client 两层能力提供方实现一个 MCP Server把工具、资源、提示词模板暴露出来Agent 侧作为 MCP Client通过统一协议去发现和调用这些能力。我习惯用一个类比来解释MCP 就像是 USB-C 接口。以前每个设备有自己的充电口现在统一了你不需要关心对面是硬盘还是显示器插上就能用。MCP Server 可以是本地进程也可以是远程服务Agent 不需要知道工具背后是 Python 脚本还是 HTTP API它只需要知道这里有一个叫search_docs的工具输入是 query输出是文档列表。在实际项目里我通常会把 MCP Server 按能力域来划分而不是按工具数量。比如一个filesystemServer负责所有文件读写、目录遍历一个databaseServer封装所有 SQL 查询和 schema 探查一个browserServer处理网页抓取和 DOM 操作一个code_runnerServer负责沙箱内执行代码这样划分的好处是每个 Server 的职责边界清晰权限也好控制。你不可能让一个负责查数据库的 Agent 顺手把文件系统删了因为它的 MCP Client 根本没连那个 Server。2.2 A2A 解决的是Agent 与 Agent 之间的协作问题A2A 要处理的是完全不同层面的问题。当你有多个 Agent每个 Agent 有自己的角色、自己的上下文、自己的工具集它们之间怎么传递任务、怎么同步状态、怎么处理依赖关系最原始的做法是让 Agent 之间直接互相调用函数A 把结果传给 BB 处理完传给 C。但这样做的问题是耦合太紧。A 必须知道 B 的接口长什么样B 改了参数 A 就得跟着改。而且一旦任务链变长错误处理、超时重试、状态回滚都会变成噩梦。A2A 的思路是引入一个消息层。Agent 之间不直接调用而是通过标准化的消息格式通信。每条消息包含发送方、接收方、任务描述、输入数据、期望的输出格式、超时时间、优先级。接收方处理完后再发一条响应消息回去。这个模式听起来很像微服务里的消息队列实际上设计理念也确实相通。我在实际项目里用 A2A 时最看重的是它带来的可观测性。因为所有 Agent 间的通信都走消息层你可以很轻松地记录每一条消息的流转、耗时、成功失败状态。调试多智能体系统最痛苦的就是不知道哪一步出了问题有了消息日志你可以像查快递轨迹一样看到任务卡在哪个 Agent 手里。2.3 两个协议在架构中的位置对比维度MCPA2A连接对象Agent 与工具/资源Agent 与 Agent核心问题能力如何被标准化调用任务如何被标准化传递通信方向通常是 Agent 主动调用双向可请求可响应状态管理无状态为主每次调用独立需要维护任务状态和会话典型实现MCP Server Client消息总线 Agent 注册中心失败处理工具级重试任务级重试、降级、回滚这张表是我自己在设计系统时反复对照的。一个常见的误区是有人试图用 MCP 来做 Agent 间的通信把另一个 Agent 包装成一个 MCP 工具。短期看能跑通但长期会出问题——因为 Agent 是有状态的、会主动发起任务的而 MCP 工具的设计假设是被动、无状态的。用错了抽象层后面扩展会非常痛苦。3. 多智能体系统的分层设计从任务入口到结果汇总3.1 编排层谁来决定任务怎么分多智能体系统的第一层是编排层Orchestration Layer。这一层的职责是接收用户输入理解意图把大任务拆成子任务然后决定每个子任务交给哪个 Agent。编排层可以是一个专门的 Orchestrator Agent也可以是一段确定性的代码逻辑。我的经验是能用代码编排的就别用 Agent 编排。原因很简单代码编排是确定性的、可测试的、可调试的Agent 编排虽然灵活但引入了不确定性一旦编排逻辑出错你很难定位是模型理解错了还是提示词写歪了。具体做法上我会把任务分成两类结构化任务流程固定比如先解析文档再抽取字段再校验再入库。这种直接用代码写死流程每个步骤调用对应的 Agent。非结构化任务流程需要根据中间结果动态决定比如根据用户问题判断需要查哪些数据源。这种才交给 Orchestrator Agent 去决策。Orchestrator Agent 的提示词里我会明确列出所有可用的下游 Agent 及其能力描述让它输出一个结构化的任务计划。这个计划通常是一个 JSON 数组每个元素包含agent_name、task_description、input_data、depends_on。有了depends_on字段编排层就能构建出任务依赖图并行执行没有依赖的任务。3.2 执行层每个 Agent 只做一件事执行层的每个 Agent 都应该遵循单一职责原则。我见过太多项目一个 Agent 既负责写代码又负责跑测试还负责修 bug结果就是提示词越写越长行为越来越不可预测。一个设计良好的执行层 Agent 应该包含这几个要素明确的角色定义一句话说清楚它是干什么的比如你是一个专门负责从需求文档中抽取功能点的分析 Agent。受限的工具集只挂载它真正需要的 MCP Server。写代码的 Agent 不需要数据库权限查数据的 Agent 不需要文件写入权限。固定的输出格式每个 Agent 的输出都应该是结构化的方便下游消费。我通常要求所有 Agent 输出 JSON并在提示词里给出 schema。清晰的失败信号Agent 遇到无法处理的情况时应该返回一个明确的错误码和原因而不是硬编一个看起来像答案的东西。这里有个实操细节Agent 的上下文要隔离。不要让所有 Agent 共享一个巨大的对话历史那样既浪费 token 又容易互相干扰。每个 Agent 只应该拿到它完成任务所需的最小上下文。A2A 消息里传递的input_data就是这个最小上下文的载体。3.3 汇总层结果怎么合并、冲突怎么处理汇总层是最容易被忽视的一层。多个 Agent 的输出汇总到一起时经常会出现冲突分析 Agent 说这个字段是必填的校验 Agent 说这个字段可以为空代码 Agent 生成的接口和文档 Agent 描述的接口对不上。处理冲突的策略我一般分三档优先级裁决给每个 Agent 的输出设定可信度权重冲突时以高权重为准。比如文档 Agent 的可信度高于代码 Agent因为文档是源头。投票机制对于分类、判断类任务让多个 Agent 独立给出结论取多数。这个在需要高可靠性的场景很有用但成本也高。人工介入对于无法自动裁决的冲突挂起任务并通知人工处理。别想着所有冲突都能自动解决有些就是需要人来拍板。汇总层的输出应该是最终交付物而不是又一份中间结果。这意味着汇总层要负责格式转换、字段补全、一致性校验。我通常会在汇总层加一个 Final Validator用确定性的代码检查输出是否符合预期 schema不符合就直接打回重跑。4. 用 MCP 搭建 Agent 的能力底座实操中的取舍4.1 MCP Server 的粒度怎么定这是我在实际项目里被问得最多的问题一个 MCP Server 应该暴露多少个工具我的答案是按能力域划分单个 Server 的工具数量控制在 5 到 15 个之间。太少了Server 数量爆炸管理成本高太多了Agent 在选择工具时容易混淆而且单个 Server 的权限范围太大不符合最小权限原则。举个例子我做代码生成系统时把 MCP Server 分成了这几个repo_server代码仓库操作包括读文件、写文件、列目录、搜索代码、查看 git 历史test_server测试相关包括运行测试、查看测试报告、获取覆盖率lint_server代码检查包括跑 linter、格式化、类型检查doc_server文档相关包括读需求文档、写接口文档、查 API 手册每个 Server 的工具数量都在 10 个左右Agent 挂载时按需选择。写代码的 Agent 挂repo_server和lint_server跑测试的 Agent 挂test_server和repo_server只读权限。4.2 工具描述怎么写才能让 Agent 选对MCP 工具的 description 字段直接决定了 Agent 能不能在正确的时候调用正确的工具。我见过太多项目工具描述写得像 API 文档什么执行查询操作返回结果集Agent 根本不知道什么时候该用它。好的工具描述应该包含三部分什么时候用明确使用场景。比如当需要根据关键词查找代码文件时使用此工具。输入是什么参数的含义和格式。别只写参数名要写清楚期望的值长什么样。输出是什么返回结果的结构。Agent 需要知道拿到结果后怎么解析。我通常会这样写工具名search_code 描述在代码仓库中根据关键词搜索匹配的代码片段。当你需要定位某个函数、类或变量的定义位置时使用此工具。输入为搜索关键词字符串支持正则表达式。返回匹配的文件路径、行号和代码片段列表最多返回 20 条结果。这段描述里当你需要定位某个函数、类或变量的定义位置时使用就是使用场景输入为搜索关键词字符串支持正则表达式是输入说明返回匹配的文件路径、行号和代码片段列表是输出说明。Agent 读完这段基本不会用错。4.3 MCP 连接的生命周期管理MCP 连接不是建立一次就一劳永逸的。在实际运行中Server 可能崩溃、网络可能抖动、token 可能过期。我踩过最大的坑就是Agent 在任务执行到一半时 MCP 连接断了它没有报错而是继续用空结果往下走最后产出一个看起来完整但完全错误的交付物。后来我加了几层保护连接健康检查每次调用工具前先 ping 一下 Server确认连接可用。调用超时设置每个工具调用设置合理的超时时间超时后返回明确的错误而不是无限等待。重试与降级对于幂等的读操作失败后自动重试 2 到 3 次对于写操作失败后不自动重试而是上报给编排层决定。结果校验工具返回后检查返回结构是否符合预期 schema不符合就当作失败处理。这些保护措施看起来繁琐但在生产环境里它们能把很多静默失败变成显式失败大大降低调试难度。5. 用 A2A 串起 Agent 协作消息设计与状态管理5.1 A2A 消息的字段设计A2A 消息是整个协作层的血液字段设计得好不好直接决定了系统的可扩展性。我经过几轮迭代后固定下来一套消息结构包含这些字段message_id消息唯一标识用于追踪和去重correlation_id关联 ID同一个任务链上的所有消息共享这个 ID方便串联sender发送方 Agent 标识receiver接收方 Agent 标识task_type任务类型接收方根据这个字段决定用哪个处理逻辑payload任务输入数据结构化 JSONexpected_output_schema期望的输出格式让接收方知道该返回什么timeout_ms超时时间priority优先级用于消息队列调度retry_count当前重试次数这套字段里我觉得最有用的是correlation_id和expected_output_schema。前者让全链路追踪成为可能后者让 Agent 之间的契约变得显式。接收方拿到消息后先看expected_output_schema就知道自己该产出什么结构的数据不用猜。5.2 同步调用还是异步消息A2A 通信有两种模式同步和异步。同步模式下发送方发出消息后阻塞等待直到收到响应或超时。这种模式实现简单适合短任务、强依赖的场景。比如编排层让分析 Agent 抽取字段必须等它返回才能进行下一步。异步模式下发送方发出消息后立即返回接收方处理完后通过回调或消息队列通知。这种模式适合长任务、弱依赖的场景。比如代码生成和文档生成可以并行谁先完成都行最后汇总时再等齐。我的经验是默认用异步只在确实需要立即结果时才用同步。因为同步调用会阻塞 Agent降低整体吞吐。而且同步调用链一长超时时间很难设置——设短了容易误杀设长了整体延迟高。异步模式下状态管理就变得很重要。每个任务需要有一个状态机记录它当前处于哪个阶段pending、running、completed、failed、timeout。编排层通过查询状态机来决定下一步动作。这个状态机我通常用一个轻量的存储来实现比如 Redis 或者 SQLite关键是保证并发安全。5.3 Agent 注册与发现机制当 Agent 数量多起来之后你需要一个注册中心来管理现在有哪些 Agent 可用它们各自能处理什么任务。最简单的做法是配置文件启动时加载。但这样每次加 Agent 都要改配置重启不够灵活。稍微好一点的做法是用一个注册服务Agent 启动时向注册中心报到声明自己的agent_id、capabilities、endpoint。编排层在分配任务时先查注册中心找到能处理该任务的 Agent 列表再根据负载情况选一个。我在项目里用的是能力标签匹配的方式。每个 Agent 注册时声明自己能处理的task_type列表编排层根据任务的task_type去匹配。如果多个 Agent 都能处理就按当前队列长度做负载均衡。这套机制不复杂但能让系统在 Agent 增减时保持弹性。6. 实战踩坑那些文档里不会写的教训6.1 上下文膨胀导致 Agent 变傻这是我在多智能体系统里踩过最贵的坑。一开始为了让 Agent 有足够信息我把所有上游 Agent 的输出都塞进下游 Agent 的上下文。结果任务链跑到第四五个 Agent 时上下文已经几万 tokenAgent 开始出现遗忘——明明前面给了字段定义它生成代码时还是用错。后来我改成按需传递每个 Agent 只拿到它完成任务必需的最小信息。具体做法是在 A2A 消息的payload里只放当前任务相关的数据而不是整个任务链的累积结果。如果某个 Agent 确实需要历史信息让它通过 MCP 工具去查而不是一股脑塞进上下文。这个改动之后不仅 Agent 的输出质量提升了token 成本也降了一大半。6.2 Agent 之间的踢皮球多智能体系统里有一种很隐蔽的失败模式A Agent 认为这个任务该 B 处理B 认为该 A 处理结果任务在两者之间来回传递永远不落地。这个问题的根源是职责边界模糊。解决办法是在设计阶段就把每个 Agent 的职责写清楚并且在 A2A 消息里加一个handled_by字段记录这个任务已经被哪些 Agent 处理过。如果一个任务被同一个 Agent 处理超过两次或者流转超过预设的最大跳数就强制上报给编排层由编排层裁决或转人工。我在编排层加了一个任务跳数计数器任何任务流转超过 10 跳就自动挂起并告警。这个简单的保护机制帮我拦下了好几次潜在的无限循环。6.3 工具调用的副作用没有隔离MCP 工具里如果有写操作——写文件、改数据库、发请求——一定要做副作用隔离。我遇到过 Agent 在重试逻辑下把同一个文件写了三遍因为前两次它以为失败了实际上只是响应超时。解决办法有两个一是写操作幂等化比如写文件时先检查内容是否已存在存在就跳过二是写操作不自动重试失败后上报由编排层决定是否重试。我倾向于后者因为幂等化需要每个工具自己实现容易漏而统一在编排层控制重试策略更可靠。另外所有写操作我都要求 Agent 先输出一个计划说明它打算改什么编排层审核通过后才真正执行。这个计划-执行两步走虽然增加了一次往返但能有效防止 Agent 误操作。6.4 超时设置的经验值超时时间设多少这个问题没有标准答案但有一些经验值可以参考任务类型建议超时说明简单工具调用10-30 秒读文件、查数据库等LLM 生成任务60-120 秒写代码、写文档等复杂分析任务180-300 秒多轮推理、长文档分析外部 API 调用30-60 秒取决于对方服务 SLA这些值不是拍脑袋定的是我在实际运行中根据 P99 延迟不断调整出来的。原则是超时时间应该略大于正常情况下的 P99 延迟这样既能拦住真正的异常又不会误杀正常但稍慢的请求。7. 一个可复现的最小多智能体系统骨架7.1 目录结构与依赖说了这么多设计原则最后给一个可以直接跑起来的最小骨架。这个骨架实现了需求文档 - 功能点抽取 - 接口定义 - 代码骨架的流程用到了 MCP 做文件操作用 A2A 做 Agent 间通信。目录结构如下multi_agent_system/ ├── orchestrator/ │ ├── main.py # 编排层入口 │ └── task_planner.py # 任务拆解逻辑 ├── agents/ │ ├── analyzer.py # 需求分析 Agent │ ├── designer.py # 接口设计 Agent │ └── coder.py # 代码生成 Agent ├── mcp_servers/ │ ├── filesystem.py # 文件操作 MCP Server │ └── doc_parser.py # 文档解析 MCP Server ├── a2a/ │ ├── message.py # 消息结构定义 │ └── bus.py # 消息总线 └── config/ └── agents.yaml # Agent 注册配置依赖主要是mcp官方 SDK、pydantic做数据校验、asyncio做异步调度。不需要额外的消息队列中间件初期用内存队列就够等 Agent 数量上来了再换 Redis 或 RabbitMQ。7.2 核心代码片段消息结构定义from pydantic import BaseModel from typing import Any, Dict, Optional import uuid class A2AMessage(BaseModel): message_id: str str(uuid.uuid4()) correlation_id: str sender: str receiver: str task_type: str payload: Dict[str, Any] expected_output_schema: Optional[Dict] None timeout_ms: int 60000 priority: int 5 retry_count: int 0编排层的任务拆解async def plan_and_execute(user_input: str): # 第一步分析需求 analyze_msg A2AMessage( correlation_idstr(uuid.uuid4()), senderorchestrator, receiveranalyzer, task_typeextract_features, payload{document: user_input}, expected_output_schema{features: list[str]} ) features await bus.send_and_wait(analyze_msg) # 第二步设计接口 design_msg A2AMessage( correlation_idanalyze_msg.correlation_id, senderorchestrator, receiverdesigner, task_typedesign_api, payload{features: features[features]}, expected_output_schema{apis: list[dict]} ) apis await bus.send_and_wait(design_msg) # 第三步生成代码 code_msg A2AMessage( correlation_idanalyze_msg.correlation_id, senderorchestrator, receivercoder, task_typegenerate_code, payload{apis: apis[apis]}, expected_output_schema{files: list[dict]} ) code await bus.send_and_wait(code_msg) return codeMCP Server 的注册以文件操作为例from mcp.server import Server from mcp.types import Tool, TextContent server Server(filesystem) server.list_tools() async def list_tools(): return [ Tool( nameread_file, description读取指定路径的文件内容。当需要查看文件内容时使用。输入为文件路径字符串返回文件文本内容。, inputSchema{ type: object, properties: {path: {type: string}}, required: [path] } ), Tool( namewrite_file, description将内容写入指定路径的文件。当需要创建或修改文件时使用。输入为文件路径和内容字符串返回操作结果。, inputSchema{ type: object, properties: { path: {type: string}, content: {type: string} }, required: [path, content] } ) ]7.3 运行与验证跑起来之后你可以用一个简单的需求文档测试输入实现一个用户注册功能需要邮箱、密码、昵称三个字段注册成功后发送欢迎邮件。预期输出应该包含分析 Agent 抽出的功能点列表、设计 Agent 给出的接口定义注册接口的路径、方法、请求体、响应体、代码 Agent 生成的骨架代码。验证的时候重点看三件事一是每个 Agent 的输出是否符合expected_output_schema二是correlation_id是否贯穿全链路三是总耗时是否在可接受范围内。如果某个环节卡住去消息日志里查对应的message_id就能定位到问题。8. 扩展方向从骨架到生产系统还差什么这个骨架能跑通流程但离生产系统还有距离。根据我的经验接下来需要补的主要是这几块可观测性。加一套完整的日志和指标采集记录每个 Agent 的调用次数、成功率、P99 延迟、token 消耗。这些数据是后续优化的基础。我通常用 OpenTelemetry 做链路追踪把 A2A 消息的correlation_id作为 trace_id这样在追踪系统里能看到完整的调用链。容错与降级。给每个 Agent 配置降级策略主 Agent 失败时切备用 Agent或者降级到更简单的处理逻辑。比如代码生成 Agent 失败时可以降级为只输出接口定义让用户手动补代码。权限与审计。生产环境里每个 Agent 能访问哪些 MCP Server、能执行哪些操作都要有明确的权限控制。所有写操作都要留审计日志记录谁在什么时候改了什么。成本控制。多智能体系统的 token 消耗是单体 Agent 的好几倍必须做成本监控。我会给每个任务设置 token 预算超预算就告警或中断。另外能缓存的中间结果尽量缓存避免重复计算。人工介入通道。再智能的系统也会遇到处理不了的情况必须留一个人工介入的入口。我的做法是在编排层加一个human_review状态任务进入这个状态后暂停等待人工确认或修改后再继续。这些东西每一个展开都能写一整篇这里就不细说了。核心思路是先把流程跑通再逐步加固。别一上来就追求完美架构那样很容易陷入过度设计最后连 Demo 都跑不起来。我在实际搭建多智能体系统时最大的体会是架构的复杂度应该由任务的实际复杂度决定而不是由技术的可能性决定。能用两个 Agent 解决的问题别硬拆成五个。MCP 和 A2A 是工具不是目标。它们的价值在于让系统在需要扩展时能平滑扩展而不是让简单问题变复杂。