我一直觉得Agent系统最让人头疼的并不是模型选型也不是Prompt写得不够好而是当你真的把一个系统拆成多个Agent之后通信问题会像滚雪球一样滚过来。熟悉这个系列的朋友应该知道前面我们陆续聊过不少Agent侧的落地手法像是怎么让Agent自己规划任务、怎么安全地接工具、怎么做结果反思之类。到了第十五篇我想换一个视角从单个Agent的内部结构转向Agent与Agent之间那个看不见的接口——也就是标题里写的A2A Agent间通信模式。A2A这个词现在有两层含义。狭义上它指2025年Google开源出来的Agent2Agent协议目标是让不同厂商、不同语言栈、甚至不同云平台上的Agent能用一套标准方式互相协作。但在本文里我更想把它当成一种设计模式来讲以任务为中心用标准化的消息结构和明确的状态机让多个Agent解耦、协作、并且能被完整追踪。哪怕你完全不引入Google那套协议这套思路也值得照搬到自己的架构里。文章的安排比较直接先聊聊为什么单Agent到最后一定撑不住然后把A2A模式的核心骨架拆给你看接着用FastAPI和LangGraph给你一个能跑起来的最小实现再把RPC、消息队列、共享存储这些方案摆在一起对比最后聊到生产环境里真正的硬骨头——并发、超时、安全、幂等。如果你正在搭多Agent系统或者已经在团队里维护一个AI Agent平台这篇应该能帮你少踩不少坑。1. 为什么单Agent模型会在复杂任务里崩塌——A2A要解决的问题1.1 单Agent的本质把一切押在一条上下文里先说为什么要做Agent间通信。大多数团队最开始都是单Agent起手的一个系统提示词再挂上一长串工具列表所有业务逻辑全挤在同一条上下文里。这个模式在任务少、工具少的时候确实够用但项目一旦做起来第一个撑不住的不是模型本身而是上下文。你会发现系统提示词越写越长从一两千字一路膨胀到上万字工具列表从十个变成五十个模型每次调用都要从这么多信息里决定现在到底该干什么注意力被严重稀释决策质量肉眼可见地下降。这个阶段可以理解成一个人的全能工具箱一个人干所有事东西再多也能硬扛但你不可能指望他24小时保持稳定高效。一旦业务复杂度上来单Agent就是那个手上同时拿着十把扳手的师傅每件事都能干但没有一件能保证干得漂亮。1.2 拆分Agent之后真正的新问题不是模型更聪明而是接口没有标准于是很多团队会自然走向第二条路把业务拆成多个Agent一个负责规划一个负责执行一个负责质检。刚拆完的时候很多人觉得模块清爽了终于可以做点大事了。但很快你会发现一个尴尬的事实——模块之间的通信方式完全是拍脑袋定的。今天大家还在同一个代码库里A调用B就是一次普通的Python函数调用明天你把B单独部署成服务函数调用就变成了HTTP后天团队决定引入Kafka结果Kafka只负责把消息送出去但没有人定义消息送出去之后我怎么知道对方有没有处理成功。通信标准和任务语义的缺失会让多Agent系统变成一场大型脑补现场。每个Agent都在用自己的方式理解任务状态不同步、失败无人处理、结果互相覆盖这些问题在单体架构里根本不存在一旦拆分就全涌出来了。这里我想起一个真实案例。我经手过一个AI内容生成系统初期团队用共享的PostgreSQL表来交换Agent之间的状态规划Agent写一行任务写作Agent查表拿到任务再把结果写回。前两周跑得挺顺直到某天两个Agent同时处理同一个任务一张表被两个写入互相覆盖排查了半天才发现是竞争条件。这个例子特别典型——不是模型不行是通信方式从头到尾就没被设计过。1.3 把通信当成架构里的一等公民分布式领域一直有一句话你不能假装网络调用和本地调用一样可靠。多Agent系统也一样你无法把跨Agent的协作当成普通函数调用处理必须把通信本身当成架构里的一等公民来对待。A2A模式解决的正是这个问题。它给Agent之间定义了一套稳定的骨架我如何声明自己会什么Agent Card、我如何理解对方交给我的一项任务Task、我如何把结果结构化地交回去Message与Artifact、对方如何知道任务进行到哪一步状态机。有了这套骨架Agent之间就不再依赖恰好住在同一个进程里这种侥幸。把Agent拆成独立服务、部署在不同团队甚至不同云厂商只要遵循这套设计协作依然稳定。2. A2A不是协议而是设计骨架Agent Card、任务与消息流转2.1 Agent Card让每个Agent先学会自我介绍A2A模式里第一个要定义的概念是Agent Card。你可以把它理解成Agent对外发布的一份能力说明书。一个Agent要和其他Agent协作对方第一步必然是先搞清楚三件事你是谁、你能干什么、怎么调用你。一份典型的Agent Card长这样{ name: writer-agent, version: 1.0, url: http://writer-agent:8001, auth: {type: api-key}, skills: [ { id: write_article, service: /tasks, description: 根据主题与大纲生成文章草稿 }, { id: rewrite_article, service: /tasks, description: 按照风格要求重写给定文本 } ] }注意这张卡片里完全没有我的Prompt是什么我用的哪个模型我的温度参数是多少——这些是实现细节协作方不需要也不应该知道。Agent Card里只暴露外部可见的信息端点地址、鉴权方式、能力列表。这背后是一条很重要的设计原则Agent之间通信只谈接口不谈实现。通信模式之所以能解耦靠的就是把内部细节彻底关在门里。2.2 任务与状态机通信的主体永远是任务而不是指令A2A模式的第二个核心概念是任务Task。Agent之间通信传递的不是一段随意的文本而是一份有明确语义的任务。这也是A2A和普通消息队列最本质的区别消息队列只管把消息送出去A2A关心的是这个任务现在处于什么状态、结果在哪、失败了吗。我常用的任务状态机是参考A2A协议思想之后做的简化版状态含义谁可以转出submitted任务已提交等待处理WorkerworkingWorker正在处理中Workerinput-required需要发起方补充信息发起方completed已完成结果在artifacts里-failed处理失败error信息可查询-canceled已取消发起方或Worker为什么状态机这么重要因为在异步通信里我发出去了和我做完了是两件完全不同的事。没有状态机双方只能靠脑补进度有了状态机任何一个Agent在任何时候都能给出一个确定性的回答这个任务走到哪一步了。这一条在排查问题的时候价值极大——你拿到一个task_id查一下状态整个链路就清晰了。2.3 消息与工件结构化传递内容而不是裸字符串任务执行过程中需要交换内容。A2A对消息做了分层消息Message负责携带语义内容继续作为结构化的part存在最终的产出放到工件Artifact里。我的习惯是给每条消息至少带三个字段role谁说的、content具体内容、以及一些可选元信息比如耗时、token用量。千万不要图省事把消息设计成一个大字符串。一旦消息变成无结构文本后续任何消费者都要自己写正则去提取关键信息那时候A2A就只剩一个空壳了。结构化消息看似多写几行代码实际是在为整个系统的可维护性买单。2.4 三种通信范式同步、异步轮询、回调A2A模式在传输层并不限定具体方式。以我的落地经验来看你有三种范式可选范式发起方式什么时候用注意点同步请求-响应直接等待结果任务耗时短、可预估调用方必须设超时异步任务轮询提交后轮询状态任务耗时长、无回调基础轮询频率要设上限Webhook回调提交后等对方主动调用事件驱动、实时性要求高回调必须鉴权加重试没有任何一种范式能通吃所有场景。我自己的取舍标准很简单任务能在几秒钟内完成直接用同步省心任务可能要跑几十秒甚至几分钟同步会长时间占着连接宁可提交任务后轮询或者直接上Webhook。不要把某个范式当成信仰A2A的骨架不应该限制传输方式。3. 落地参考用FastAPI LangGraph手写一个A2A通信最小实现3.1 先确定一个具体的协作场景概念讲得再多不如一个能跑的最小实现。这里我拿一条AI写作管道举例它有三个角色planner-agent负责把用户需求拆成大纲writer-agent根据大纲撰写全文reviewer-agent对成稿做质检并给出修改建议。这三个Agent如果全部塞进一个进程用函数调用连起来是最简单的方式也完全没必要上A2A。但为了把通信链路演示清楚我让writer-agent独立成一个FastAPI服务planner-agent作为调用方通过网络发起任务。这样每一步协作都落到明面上你也能清楚看到A2A解决的是哪些问题。3.2 先写任务模型再谈Agent Card第一步把任务模型和状态机落成代码。我用PythonPydantic来做因为它能和FastAPI无缝衔接写起来也很干净。# task_models.py from enum import Enum from datetime import datetime from pydantic import BaseModel, Field class TaskStatus(str, Enum): SUBMITTED submitted WORKING working COMPLETED completed FAILED failed CANCELED canceled class Message(BaseModel): role: str # planner、writer、system content: str created_at: datetime Field(default_factorydatetime.utcnow) class TaskRequest(BaseModel): type: str # skill id例如 write_article context: list[Message] task_id: str | None None # 由发起方生成保证幂等 class Task(BaseModel): id: str type: str status: TaskStatus context: list[Message] artifacts: list [] error: str | None None这里有个细节我要特别强调task_id的生成逻辑建议放在发起方而不是接收方。为什么因为发起方在超时后很可能重试同一个任务。如果task_id由接收方生成重试时接收方根本无法区分这是同一个任务的新请求还是两个不同的重复任务。由发起方生成task_id接收方只需要做一个如果这个id已存在就直接返回现有任务的幂等判断整个系统的重试语义立刻清晰了。3.3 用FastAPI暴露任务端点用异步Worker处理接下来是接收方writer-agent的服务端。我搭一个最简版本去掉鉴权、日志、持久化这些噪声让你一眼看清A2A骨架# writer_agent.py import asyncio import uuid from fastapi import FastAPI, HTTPException from task_models import Task, TaskRequest, TaskStatus app FastAPI() tasks: dict[str, Task] {} async def run_writer(task: Task) - str: # 这里替换成你真实的模型调用 topic task.context[-1].content await asyncio.sleep(2) # 模拟模型耗时 return f这是关于《{topic}》的完整草稿…… app.post(/tasks, status_code201) async def create_task(req: TaskRequest): if req.task_id and req.task_id in tasks: return tasks[req.task_id] # 幂等返回 task Task( idreq.task_id or str(uuid.uuid4()), typereq.type, statusTaskStatus.SUBMITTED, contextreq.context, ) tasks[task.id] task asyncio.create_task(process(task.id)) return task async def process(task_id: str): task tasks[task_id] task.status TaskStatus.WORKING try: result await run_writer(task) task.status TaskStatus.COMPLETED task.artifacts.append({name: draft, content: result}) except Exception as e: task.status TaskStatus.FAILED task.error str(e) app.get(/tasks/{task_id}) async def get_task(task_id: str): task tasks.get(task_id) if not task: raise HTTPException(404, task not found) return task这个服务简单到有点粗暴创建任务的端点、查询状态的端点、内存字典存状态。但A2A模式最核心的骨架已经完整呈现出来了。真实生产环境当然不会用内存字典存任务至少要换成Redis或者PostgreSQL但整体结构不会变。3.4 LangGraph负责Agent内部编排A2A负责Agent之间通信接下来是调用方planner-agent的实现。它内部的业务逻辑可以交给LangGraph管理但跨Agent的调用走标准A2A协议。我的习惯是把这两件事严格分开Agent内部怎么思考推理用Graph去编排Agent之间怎么传任务用统一的任务接口。# planner_agent.py import asyncio import httpx from langgraph.graph import StateGraph, END from task_models import TaskStatus AGENT_CARD { name: writer-agent, url: http://localhost:8001, } async def invoke_writer(state: dict) - dict: async with httpx.AsyncClient() as client: resp await client.post( f{AGENT_CARD[url]}/tasks, json{ type: write_article, context: [{role: planner, content: state[outline]}], task_id: state[task_id], # 幂等 }, ) resp.raise_for_status() task resp.json() while task[status] in (TaskStatus.SUBMITTED.value, TaskStatus.WORKING.value): await asyncio.sleep(1) resp await client.get(f{AGENT_CARD[url]}/tasks/{task[id]}) task resp.json() if task[status] TaskStatus.FAILED.value: state[error] task.get(error) else: state[draft] task[artifacts][0][content] return state # 用StateGraph连接节点 graph StateGraph(dict) graph.add_node(invoke_writer, invoke_writer) graph.set_entry_point(invoke_writer) graph.add_edge(invoke_writer, END) app_planner graph.compile()这段代码里的轮询循环我故意写得很简单1秒一次。生产环境我不会用sleep轮询而是换Webhook或者更高效的长轮询。但演示场景里轮询反而能更直观地展示A2A的任务生命周期。3.5 团队主栈是Java怎么办Spring AI与A2A的路子很多后台团队的主栈是Java看到Python示例可能会犹豫。这里可以肯定地说Google A2A协议本身与语言无关Java团队不需要引入任何Python组件。你完全可以用Spring Boot WebClient照着上面的骨架写一套等价实现RestController负责接收任务ExecutorService或者虚拟线程处理异步逻辑Redis存任务状态。热搜里的a2a spring和spring ai agent我也在关注。Spring AI目前确实在Agent编排方向上持续演进对MCP和A2A的官方支持迟早会补全。但在那之前自己用Spring Boot实现任务端点、状态机和幂等判断并不复杂而且能更好地贴合团队现有技术栈。不要被框架不齐卡住——A2A的本质是一份JSON结构任何HttpServer都能实现。4. A2A与其他协作模式的取舍RPC、事件总线、共享记忆谁更适合你的场景4.1 先把五种协作方式摆到一张表里很多团队不是不知道怎么实现而是不知道该怎么选。我经常被问到到底用消息队列好还是让Agent之间直接HTTP调用这里我给出一个自己反复对比过的表格希望能帮你快速做判断。协作方式耦合程度异步支持可靠性与追踪适合场景进程内函数调用最高无一般单进程、低延迟、强事务RPCgRPC/HTTP高弱一般同步服务调用、接口稳定消息队列/事件总线低强中需要额外做结果路由事件流、解耦、广播共享数据库/黑板低强低竞争条件风险高数据中心型协作、历史包袱重A2A模式中强高任务状态天然可追踪跨Agent、跨团队、长任务协作表里有一条值得展开消息队列的异步性强但它和A2A关注的并不是同一个问题。Kafka能帮你把任务事件可靠地送出去可一条消息送出去之后任务的最终状态是什么、结果存在哪里、对方有没有处理失败消息队列自己不负责。你完全可以用消息队列作为A2A底层的传输通道但任务状态机必须由业务层自己维护。4.2 我的选型决策路径我自己选型的时候基本沿着这条思路走如果两个Agent部署在同一个进程里任务简单到不需要追踪那就直接用函数调用。不要为了设计模式而设计模式那是本末倒置。如果Agent已经拆成了独立服务而且需要跨语言、跨团队协作优先考虑A2A。任务状态机带来的可观测性收益非常直接。如果系统里存在大量广播式业务事件比如订单状态变更要通知十几个下游不要硬套A2A用Kafka做事件总线更合适。如果团队已经有成熟的共享数据库暂时不想引入更多中间件可以先拿共享表过渡。但每一条任务记录一定要带上owner和version并且做好并发控制否则迟早踩我之前讲过的那种互相覆盖的坑。4.3 什么时候不该用A2AA2A不是银弹有几种情况我明确不建议上。第一延迟要求极低的场景。比如两个Agent在一个请求链路里必须几十毫秒内完成交互这时候A2A的异步状态机反而是负担同进程内直接调用更合理。第二强事务场景。如果两个Agent必须严格地在同一次事务里一起成功、一起失败通过HTTP或者消息传递几乎不可能保证原子性。正确做法是缩小事务边界不要让跨Agent调用掺和进事务里。第三系统总共只有两个Agent、还是一个团队维护并且你很确定未来不会扩展。这种情况下A2A的收益很小先把业务跑通比什么都重要。4.4 混合架构通常是常态还有一个经验想分享真实生产环境里A2A不是用来替代消息队列的而是和消息队列共存的。我常用的组合是A2A负责请求-响应式的长任务协作事件总线负责通知-订阅式的信息扩散。比如planner-agent完成规划后往Kafka发一个plan.ready事件下游的writer-agent收到事件之后再通过A2A的POST /tasks接口领取写作任务。这样既有了事件流的解耦又保留了任务状态的追踪能力。5. 并发、超时、安全与幂等A2A真正上生产后的工程坑5.1 Agent怎么扛并发从限流到编排层热搜词里有一条ai agent怎么扛并发这在A2A场景下尤其现实。当多个Agent同时调用我的服务如果Agent端不做任何保护第一轮冲击就会把上游模型API的rate limit打爆然后雪崩式失败。最基本的防线是在Agent服务端加并发信号量。FastAPI里可以用asyncio.Semaphore包住模型调用把并发上限封死在合理范围import asyncio sem asyncio.Semaphore(5) async def call_llm_with_limit(*args, **kwargs): async with sem: return await call_llm(*args, **kwargs)如果并发量再大就要引入队列和Worker池而不是把每个请求都硬塞给模型。这里有个常见误区很多人以为扛并发等于无脑开多线程调用。实际上对LLM服务来说瓶颈往往不在你的服务进程而在上游模型API的QPS配额。与其无限增加并发不如在Agent Card里声明自己的并发上限让客户端做自适应退避。5.2 超时与重试没有幂等的重试就是事故现场A2A里至少要有三层超时连接超时、读取超时、任务总时长超时。连接和读取超时属于传输层任务总时长超时属于业务层——如果writer-agent处理一个任务超过了预设的10分钟planner-agent不应该老老实实等着而是把任务标记为超时再决定是重试还是降级。重试同样有讲究。我强烈建议每次任务的task_id由发起方生成并要求接收方具备幂等返回的能力同样的task_id来了第二次直接返回第一次的处理结果而不是再跑一遍模型。这个设计可以让重试变得非常安全代价只是后端多做一次字段判断。没有幂等设计的重试本质上就是在为事故提前买票。5.3 回调风暴Webhook模式的隐藏风险当你用Webhook做异步结果通知时会碰到一个经典难题回调重复。对方网络抖动导致回调没送达重试之后可能重复投递或者对方回调处理得太慢上游以为失败又触发了一次。解决办法是在回调接口里同样维护task_id的处理记录同一个task_id的回调最多生效一次。另外回调接口必须做鉴权否则任何人都可以伪造一个completed状态把你的任务标记成完成。别觉得这是危言耸听公网环境下不加鉴权的回调接口基本等于给攻击者留了一扇门。5.4 安全Agent之间的信任不能裸奔Agent之间通信的安全级别取决于Agent的部署范围。如果全部在同一个内网简单API Key就够了如果Agent跨公网部署建议至少HTTPS加mTLS或者用短期令牌做鉴权。这里有一个原则任意两个Agent之间的调用都必须回答清楚我怎么确认对方的身份以及我能放心把任务交给对方吗。另外限流是比鉴权还容易被忽略的安全措施。不加限流的Agent服务等于在公网上裸奔。你不想因为某个调用方写了个死循环把自己整个Agent系统的模型预算烧光。在Agent端做全局限流同时支持按调用方做配额隔离是更稳妥的做法。5.5 可观测性trace_id和任务存储不能省多Agent系统调试最痛的点在于报错发生在writer-agent但根因可能在planner-agent。所以我从很早开始就规定整个系统必须有一个贯穿的trace_idA2A的每一条消息、每一次任务状态变更、每一行日志都要带上这个trace_id。一张链路图下来问题出在哪一跳一目了然。任务存储也不能用内存dict糊弄。生产环境至少要换成Redis或PostgreSQL原因有两个一是进程重启之后任务不能丢二是多个Worker实例需要共享任务状态。用Redis做状态存储时记得用SET NX这类原子操作避免并发覆盖用PostgreSQL则给task表加唯一索引天然帮助幂等。最后再分享一个小建议别一上来就追求把A2A协议实现得多标准。这套设计模式真正的核心是三个骨架——Agent Card、任务状态机、带着trace_id的结构化消息。我见过太多团队在还没跑通一个跨Agent调用的时候就急着讨论该用JSON Schema还是protobuf最后被协议细节拖住了进度。先把任务、状态、幂等这三件事定下来A2A带给你的收益会远超它的成本。至于协议要不要往Google的A2A标准靠拢那是系统真正跑起来之后的第二阶段再考虑的事。