企业官网建设流程全解析

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.10，FastAPI做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.12
• FastAPI 0.111
• uvicorn 0.30
• a2a-sdk 0.1.x（官方Python SDK）
• httpx 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(text=result) 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(), port=5001) 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(text=user_text)])) # 3. 把结果原样返回给用户 if response and response.parts: return TextPart(text=response.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(), port=5000) 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模拟用户发一条消息给调度Agent：

import 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/", json=payload) 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_agent@step=1] 北京当前天气：晴，温度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_code=401, detail="Unauthorized")

然后在注册路由的时候，把这个依赖加进去。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没反应

遇到这个问题的第一反不应该是全网都说。

排查步骤：

1. 确认目标Agent进程还在跑（ps -ef | grep python）
2. 确认端口通不通（curl http://127.0.0.1:5001/agent/recommendations）
3. 看目标Agent的日志，确认它有没有真的收到请求
4. 检查是不是路由配置里的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协作系统时挨个踩过的。有些东西写下来就几句话，但它消耗的调试时间往往是几个小时甚至一整天。

7. 从Demo到生产：额外需要注意的细节

如果你想把这套东西真正放到生产环境，下面几个问题迟早会遇到。

7.1 超时和重试策略

A2A是基于HTTP长连接的，当一个Agent处理一个复杂的任务需要很长时间，下游Agent可能早就超时断开。所以在调用A2A客户端时，建议设置合理的超时时间和重试策略：

client = A2AClient( url="http://127.0.0.1:5001/", timeout=10.0, # 连接超时 max_retries=2, retry_backoff=0.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”），开多个终端虽然乱，但配合日志看起来一目了然。等你要真正部署了，再把这些能力收敛到监控平台里不迟。