1. 先聊清楚A2A协议到底解决了什么问题接上一篇。我们在上篇里把A2A协议的核心概念、Agent Card、消息结构和基本通信流程都过了一遍这篇重点讲实战——多个Agent之间到底怎么协作消息怎么路由任务怎么拆解以及我在实际跑通整套流程时踩过的坑。如果你还没看过上篇建议先补一下基础因为接下来的所有内容都建立在一个前提上你手里已经有一个能跑通A2A基本握手流程的环境知道怎么发布Agent Card也清楚JSON-RPC 2.0的消息长什么样。没看过也没关系这篇会尽量把关键点再带一句但核心还是“怎么让多个Agent真正协同干活”。先说A2A协议最核心的价值它定义了一种与具体实现框架无关的通信标准。什么意思就是说你用Python写的Agent可以和用Node.js写的Agent直接对话只要两边都实现了A2A协议。这跟以前完全不一样——以前我们做多Agent协作基本就是在同一个框架里用同一个SDK或者自己写HTTP接口互相调用搞得每个项目都有一套私有的通信协议Agent之间根本没法互通。A2A的思路其实特别好理解它就相当于给Agent世界定了一个“普通话”。每个Agent只需要暴露自己的Agent Card说明自己能干什么、怎么调用然后通过统一的JSON-RPC消息格式进行通信。至于Agent内部是用大模型、规则引擎还是传统算法A2A协议不关心也不需要关心。这篇的实战环境我用的是一台普通的开发机Python 3.10FastAPI做HTTP服务A2A SDK版本是0.1.x三个Agent分别跑在本地的不同端口上。整个系统规模不大但足够说明问题。2. 多Agent协作的三种典型模式在动手写代码之前先把架构层面的东西想清楚。多Agent协作不是简单地把几个Agent连起来就完事消息怎么走、任务怎么拆、谁来决策这些决定了系统的复杂度和可靠性。我基于实际经验把常见的协作模式分成三种。2.1 中心化调度模式这种模式最简单也好理解有一个“调度Agent”或者“编排Agent”它负责接收用户请求然后把任务拆成子任务分发给其他Agent最后汇总结果返回给用户。打个比方这就像公司里的项目经理。项目经理自己不干活但他知道团队里谁擅长什么把活分下去最后把大家的成果整合起来交差。这种模式的优势是逻辑清晰、好调试。所有消息都经过调度节点你可以很方便地在这层做日志记录、超时控制、重试策略。缺点则是调度节点容易成为瓶颈一旦它挂了整个系统就瘫了。在A2A协议里调度Agent其实就是个普通的Agent只不过它额外承担了任务路由的职责。它需要知道其他Agent的Agent Card然后把用户的请求转化成合适的消息发给对应的Agent。2.2 去中心化模式去中心化模式没有中心节点每个Agent都可以主动和其他Agent通信。这种模式更像是“群聊”——谁有问题就谁谁有空就响应谁的请求。这种模式适合任务边界不清晰、需要互相协商的场景。比如一个写代码的Agent发现自己缺一个接口定义它可以直接去问负责架构设计的Agent而不需要经过中间人。但去中心化带来的问题是协调成本高。你没法在一个地方看到全局的消息流向出了问题很难追踪。如果没有额外的手段Agent之间还可能形成循环调用比如A问B、B问C、C又问A直接死循环。在A2A协议层面去中心化模式并没有额外的机制来阻止循环——协议不关心你的拓扑结构。所以如果选择这种模式一定要自己在业务层面加上“消息跳数限制”或者“任务深度限制”。2.3 分层混合模式第三种是我个人在实际项目中用得最多的分层混合模式。顶层有一个调度Agent下面按业务域分成若干子组每个子组内部可以去中心化协作子组之间不直接通信必须通过上层调度Agent转发。这个模式的好处是兼顾了前两种的优点——上层有整体的控制力下层又有灵活的协作空间。而且扩展性很好新增一个业务域只需要在上层注册新的组信息不大会影响已有链路。在A2A协议中实现这种模式核心是要设计好Agent Card的元数据让上层Agent能准确判断“这个任务该往哪个子组发”。我一般会在Agent Card的description字段里写清楚这个Agent擅长什么、不擅长什么甚至把典型的任务样例写进去方便调度方做意图匹配。三种模式没有绝对的好坏全看业务场景。我见过有人用纯中心化模式跑通了十几个Agent的协作系统也见过去中心化模式在一两个Agent的场景下表现得很灵活。但如果你同时要处理的Agent数量超过五个我个人强烈建议用分层混合模式——否则光维护消息路由规则就能让你崩溃。3. 动手搭一个最小可行系统说再多理论不如直接跑一个能动的系统。这一节带你从零搭一个三Agent协作系统一个调度Agent两个业务Agent一个负责查天气一个负责算日期差。用户对调度Agent发问调度Agent决定把问题分给谁。3.1 环境准备和依赖安装先说环境。我用的版本组合如下不一定最新但稳定性实测没问题Python 3.10.12FastAPI 0.111uvicorn 0.30a2a-sdk 0.1.x官方Python SDKhttpx 0.27用于Agent之间的HTTP调用安装命令很简单pip install fastapi uvicorn a2a-sdk httpx这里有个容易踩的坑a2a-sdk的版本更新很快API变动也大。我一开始照着网上某个教程写的代码结果人家用的是旧版API方法名都对不上。建议装好之后先跑一下from a2a.sdk import A2AClient, A2AServer print(A2AClient, A2AServer)把这两个类打印出来确认一下再动手写代码。如果你用的是某个特定版本最好直接看本地安装的源码里的类型定义不要盲信网上文章。3.2 业务Agent的骨架代码先写一个通用的天气查询Agent。这个Agent做的事情很简单接收A2A请求提取里面的消息内容调一个模拟的天气API返回结构化结果。# weather_agent.py import json from a2a.sdk import A2AServer, A2AClient from a2a.types import Message, TextPart class WeatherAgent: def __init__(self): self.agent_info { name: weather_agent, description: 根据城市名查询实时天气支持输入城市中文名或英文名, url: http://127.0.0.1:5001/, } async def handle_message(self, message: Message): # A2A协议的消息文本存在 message.parts 里 text_parts [part for part in message.parts if isinstance(part, TextPart)] if not text_parts: return TextPart(text没有找到有效文本内容) user_text text_parts[0].text.strip() city self._parse_city(user_text) if not city: return TextPart(text我没听懂你想查哪个城市的天气) weather self._mock_weather(city) result f{city}当前天气{weather[condition]}温度{weather[temperature]}度 return TextPart(textresult) def _parse_city(self, text: str): # 简单的关键词匹配生产环境可以换成NLP模型 common_cities [北京, 上海, 广州, 深圳, 杭州, 成都] for city in common_cities: if city in text: return city return None def _mock_weather(self, city: str): import random conditions [晴, 多云, 小雨, 阴] return { condition: random.choice(conditions), temperature: random.randint(18, 32), } server A2AServer(WeatherAgent(), port5001) server.start()代码里的_parse_city用了最简单的关键词匹配意图清晰但能力有限。真正生产环境我建议在这个位置接一个轻量级的分类模型或者直接用大模型的function calling来提取参数。但作为Demo关键词匹配足够了而且方便排查问题。3.3 调度Agent的实现逻辑调度Agent是核心它决定了消息往哪走。我的实现思路是先解析用户问题提取意图关键词然后根据关键词匹配到对应的业务Agent再用A2A客户端发起请求。# dispatcher.py import asyncio from a2a.sdk import A2AServer, A2AClient from a2a.types import Message, TextPart AGENT_REGISTRY { weather: { url: http://127.0.0.1:5001/, keywords: [天气, 温度, 下雨, 气温], }, calendar: { url: http://127.0.0.1:5002/, keywords: [日期, 几天, 间隔, 工作日], }, } class DispatcherAgent: def __init__(self): self.agent_info { name: dispatcher, description: 调度Agent负责把用户问题分发给对应的业务Agent, url: http://127.0.0.1:5000/, } async def handle_message(self, message: Message): text_parts [part for part in message.parts if isinstance(part, TextPart)] user_text text_parts[0].text if text_parts else # 1. 意图匹配 target_agent self._match_agent(user_text) if not target_agent: return TextPart(text当前没有Agent能处理你的请求请换个说法试试) # 2. 调用目标Agent client A2AClient(target_agent[url]) response await client.send_message(Message(parts[TextPart(textuser_text)])) # 3. 把结果原样返回给用户 if response and response.parts: return TextPart(textresponse.parts[0].text) return TextPart(text目标Agent没有返回有效内容) def _match_agent(self, text: str): for agent_name, info in AGENT_REGISTRY.items(): for kw in info[keywords]: if kw in text: return info return None server A2AServer(DispatcherAgent(), port5000) server.start()这里的_match_agent同样是基于关键词的简单、直观、可预测。在实际项目中这种硬编码映射很快会变得难以维护。我有一个经验是把意图映射的配置抽出来放到一个YAML或者JSON文件里这样运营人员也能修改映射关系不需要动代码。比如# intent_mapping.yaml intents: - agent: weather keywords: [天气, 温度, 下雨, 气温] - agent: calendar keywords: [日期, 几天, 间隔, 工作日]代码启动时加载这个文件以后加新Agent只需要在这个文件里加一段配置。这个改动虽然小但实际维护起来差别巨大。3.4 启动顺序和本地联调三个Agent都写好之后启动顺序有讲究。我一般先启动业务Agent再启动调度Agent——因为调度Agent在启动时不会主动去连接别人它只在收到消息后按需调用所以顺序其实不影响功能但先启动业务Agent能避免测试的时候出现连接拒绝。启动方式# 终端1 python weather_agent.py # 终端2 python calendar_agent.py # 终端3 python dispatcher.py然后打开另一个终端直接用httpx模拟用户发一条消息给调度Agentimport httpx payload { jsonrpc: 2.0, id: 1, method: message/send, params: { message: { role: user, parts: [{kind: text, text: 北京今天天气怎么样}] } } } response httpx.post(http://127.0.0.1:5000/, jsonpayload) print(response.json())正常情况下你应该能在响应里看到类似“北京当前天气晴温度25度”的结果。这就意味着调度Agent正确地把消息转发给了天气Agent并且把结果带回来了。4. 任务拆解与消息路由的进阶玩法上面的最小系统跑通之后你可能会发现一个很现实的问题大部分真实任务不是一个Agent就能搞定的。比如用户问“北京这周六适合户外活动吗”这个问题至少涉及两个能力查天气和查日期周六是哪天。这时候就需要任务拆解和结果合并。4.1 任务拆解的两种思路第一种是在调度层拆。调度Agent拿到问题自己判断需要哪些Agent参与然后向每个Agent发送子任务最后把结果拼起来。这种方式的优点是调度方完全掌控流程缺点是调度逻辑会变得很重——如果任务拆解的规则复杂调度Agent本身就成了一个“超级Agent”。我之前见过一个失败的例子就是把调度Agent写得过于复杂最后它自己成了整个系统里最难维护的组件一换模型就全崩。第二种是由Agent自主决定是否还需要其他人帮忙。当一个Agent发现自己的任务依赖于另一个Agent的能力时它可以直接向对方发送请求。这种方式灵活但正如前面说的如果没有中心控制点容易失控。我的建议是小规模系统用第一种规模大了再用第二种而且第二种模式一定要配全链路追踪。4.2 用步骤编号追踪多轮调用链不管用哪种思路追踪消息的流转路径都是必须做的事。A2A协议本身的消息结构里没有太多针对多跳路由的专用字段但我们可以利用现有的字段做文章。我的做法是在每个Agent处理完消息后往返回的TextPart前面加上自己的Agent ID和步骤编号。比如[weather_agentstep1] 北京当前天气晴温度25度这样在最终返回给用户之前调度Agent可以按顺序拼出完整链路[dispatcher] - [weather_agent] - 北京当前天气晴温度25度看着简单但在排查问题的时候指标特别有用。有一次我部署了五六个Agent用户一直反馈某个任务返回超时我就是在日志里看到某个Agent的步骤编号重复出现了两次才定位到是消息循环导致的时间消耗。4.3 消息路由的配置化改造最小系统里的AGENT_REGISTRY是写死在代码里的只适合Demo。生产环境我一般会引入一个简单的路由表配置包含目标Agent的URL、认证信息、能力关键词、超时时间、最大重试次数。一个典型的配置长这样agent_route: weather: url: http://127.0.0.1:5001/ timeout: 5.0 max_retry: 2 keywords: - 天气 - 温度 - 下雨 calendar: url: http://127.0.0.1:5002/ timeout: 3.0 max_retry: 1 keywords: - 日期 - 工作日 - 间隔天数路由模块启动时读取这个文件把它加载成一个内存里的字典。每次收到消息先用配置里的关键词做粗匹配如果粗匹配命中多个Agent再进入一个“意图消歧”逻辑——最简单的做法是把这几个Agent的描述字段丢进一个小模型里让模型选一个最合适的。这个思路我试过多次效果比纯关键词匹配好很多而且不至于像完全用大模型做意图识别那么重。关键点是粗匹配一定要快且确定消歧可以慢一点但要准。5. 消息加密、认证与Agent间信任跑内网Demo的时候不需要考虑安全但一旦你的Agent要跨网络通信或者接外部客户端安全就是绕不开的话题。A2A协议本身没有强制要求环境部署方式但好消息是它基于HTTP所以所有Web领域的成熟方案都能直接用。5.1 最简单的Token认证我在实际项目里最常用的就是在每个Agent上加一个静态Token校验用FastAPI的依赖注入就很方便from fastapi import Header, HTTPException def verify_token(x_a2a_token: str Header(...)): expected_token your_secure_token if x_a2a_token ! expected_token: raise HTTPException(status_code401, detailUnauthorized)然后在注册路由的时候把这个依赖加进去。A2A SDK在发起请求时可以通过参数附带自定义Header这样每个Agent之间互相调用都会带上Token。不建议在公网环境直接用明文Token建议至少用HTTPS传输并定期更换Token。如果有条件更规范的方案是走OAuth2.0或者mTLS但这部分往往会引入额外的证书管理成本小团队可以根据自己的风险承受能力决定。5.2 认证与任务审计日志还有一点容易被忽略多Agent协作系统里一定要记录每个协作请求的身份来源。现在很多场景下主调度Agent收的是最终用户的请求但它再转发给下游Agent时下游Agent只看到“调度Agent”看不到原始用户。我踩过这个坑。有一次某个Agent的调用方变了权限没对齐导致数据被不该看的人查走了。排查了半天才发现问题出在链路中间的身份丢失。解决方案是在消息里加一个task_id字段非A2A标准字段但可以在自定义扩展里带上调度Agent在接收用户请求时生成一个唯一ID并把用户身份信息填进去再转发给下游Agent。这样即使链路很长也能一眼看出这个请求最初是从哪个用户发起的。很多A2A SDK允许你在消息的extensions或者元数据区域放入自定义信息利用这个能力去传递task_id是非常有效的做法。6. 常见问题与排查技巧实录这一节把自己平时排障的过程直接放出来比标准文档实用得多。6.1 消息发出去了但Agent没反应遇到这个问题的第一反不应该是全网都说。排查步骤确认目标Agent进程还在跑ps -ef | grep python确认端口通不通curl http://127.0.0.1:5001/agent/recommendations看目标Agent的日志确认它有没有真的收到请求检查是不是路由配置里的URL写错了多了一个斜杠或者端口不对我在这个坑上栽过的版本是本地FastAPI监听的是0.0.0.0但服务注册表里写的是127.0.0.1测试机上看没问题换到另一台机器调用就失败了。6.2 返回结果的格式不规范A2A协议对TextPart要求不严但下游Agent如果解析不定长的自定义格式还是容易出问题。我自己的经验是所有业务Agent在返回结果时尽量走结构化字段。比如在文本中间用一行容易解析的模板{action: query_weather, data: {city: 北京, temp: 25, condition: 晴}}然后调度Agent拿的时候先用正则在文本里找JSON再取值填充给用户。这样做的好处是后续想接一个展示层或者语音播报层都很容易。6.3 多Agent协作时怎么定位是哪个环节出错我现在的做法很土但很有效每个Agent的响应头部都加上自己的识别信息。用A2A可以给响应消息加extensions字段里面写清楚agent_id、processing_time_ms、parent_task_id。当最终结果出问题时沿着parent_task_id往回追溯就能看到是哪一层出了问题。这个完全不需要引入特别复杂的链路追踪系统只要所有Agent约定好写这个字段就行。写一个简单的中间件每个Agent启动时自动注入这个逻辑省心很多。6.4 常见问题速查表现象可能原因快速处理请求超时目标Agent没启动 / Agent卡在长任务检查端口给业务Agent加上超时控制返回内容为空Agent解析的parts类型不对检查是不是用了TextPart而不是非文本类型消息被错误路由关键词匹配命中多个Agent拉长关键词列表或加消歧逻辑出现循环调用去中心化模式缺少跳数限制统一代理关系加最大跳数身份不明确下游不知道是谁在调用增加task_id透传以上这些坑基本上是我从零开始做多Agent协作系统时挨个踩过的。有些东西写下来就几句话但它消耗的调试时间往往是几个小时甚至一整天。7. 从Demo到生产额外需要注意的细节如果你想把这套东西真正放到生产环境下面几个问题迟早会遇到。7.1 超时和重试策略A2A是基于HTTP长连接的当一个Agent处理一个复杂的任务需要很长时间下游Agent可能早就超时断开。所以在调用A2A客户端时建议设置合理的超时时间和重试策略client A2AClient( urlhttp://127.0.0.1:5001/, timeout10.0, # 连接超时 max_retries2, retry_backoff0.5, )如果Agent的业务确实很重比如需要调大模型生成内容可以考虑把任务改造成异步提交先用message/subscribe或者自己的任务队列接口把任务提交过去然后再单独轮询结果。A2A协议里已有针对长任务处理的机制可以基于这个思路集成。7.2 服务注册与发现Agent少的时候写死URL没问题但Agent多了动态扩缩容的时候就头疼了。我自己在项目里用过最简单的方案用Redis做服务注册表。每个Agent启动时把自己的Agent Card信息写进Redis设置一个TTL然后定时续期。调度方查询的时候从Redis拉取实时列表。这个方案的好处是不用额外搭注册中心对中小系统完全够用。如果要更加工程化的话可以接Consul或Nacos但那个复杂度就不在A2A本身的范围里了。7.3 监控和日志多Agent系统里日志的格式必须统一。我的建议是每条日志至少包含Agent ID、Task ID、时间戳、消息类型使用结构化日志如JSON格式方便做日志采集和分析对调度Agent额外记录每个子任务的耗时开头就说过A2A协议其实不关心Agent内部怎么实现。你完全可以在通信层保持A2A的代理透明而内部模块则自行设计。通信协议统一内部实现自由这是A2A协议最值得大家借鉴的一点也是它能流行开来的原因。最后再分享一个小技巧在本地调多个Agent时把每个Agent的终端窗口标题改成Agent的名字比如“调度Agent”“天气Agent”开多个终端虽然乱但配合日志看起来一目了然。等你要真正部署了再把这些能力收敛到监控平台里不迟。