A2A多Agent协作实战：Agent选择、协作机制与三层记忆管理

1. 从一个真实请求链路说起A2A 到底在解决什么问题假设你对着一个智能助手说了一句“帮我查一下上个月华东区的销售数据做个同比分析然后生成一份 PPT 发给我。”这句话对人来说稀松平常但对系统来说它同时包含了数据检索、数值计算、文档生成、消息投递四类完全不同的能力需求。如果把这些能力全部塞进一个模型里结果通常是模型在某一环上掉链子整条链路就崩了。A2AAgent-to-Agent要解决的核心问题就是把一件复杂的事拆给多个各有所长的 Agent让它们像一个小团队一样协作完成。这里的关键词有三个选谁来做Agent 选择、怎么配合协作机制、做完的事记在哪记忆管理。这三个问题不解决多 Agent 系统就只是一堆各自为政的接口调用谈不上“协作”。我在实际项目里踩过最典型的坑就是一开始把 A2A 理解成“多个 Agent 互相调 API”。结果做出来的东西A 调 B、B 调 C调用链一长谁该负责什么完全说不清出错之后连日志都对不上。后来才明白A2A 的本质不是“调用”而是基于能力描述的任务分派与状态同步。每个 Agent 要先说清楚“我能干什么、我干不了什么”调度方才有可能做出正确选择。这篇文章我会按真实请求的推进顺序来讲先讲 Agent 怎么被选中再讲它们之间怎么协作然后重点讲记忆到底该放在哪一层最后给出一套可以照着搭的最小实现思路。适合已经了解大模型基础调用、想往多 Agent 方向落地的开发者也适合正在评估“要不要上多 Agent”的技术负责人。提示A2A 不是银弹。如果你的任务本身是单轮的、能力单一的硬拆成多 Agent 只会增加延迟和故障点。判断标准很简单——任务是否需要两种以上差异明显的能力且这些能力由不同模型或工具更擅长。2. Agent 选择不是“谁最强”而是“谁最合适”2.1 能力描述卡让调度方有据可依Agent 选择的第一步是每个 Agent 都得有一张能力描述卡。这张卡不是随便写句“我擅长数据分析”就完事它需要结构化至少包含四个字段能力域、输入格式、输出格式、限制条件。我一般会把它设计成类似下面这样的结构{ agent_id: data_analyst_01, capability: [sales_data_query, trend_analysis, ratio_calculation], input_schema: {region: string, period: string}, output_schema: {metrics: object, chart_data: array}, constraints: {max_rows: 100000, timeout_sec: 30} }为什么要这么细因为调度方在做选择时靠的不是“感觉这个 Agent 行”而是把当前任务的输入输出和候选 Agent 的 schema 做匹配。输入对不上、输出接不住、超出限制条件直接排除。这一步能过滤掉大部分错误选择。2.2 选择策略规则匹配、语义匹配与混合打分实际落地时Agent 选择通常有三种策略各有适用场景。规则匹配最简单就是按能力标签硬匹配。任务需要trend_analysis就找 capability 里带这个标签的 Agent。优点是快、可预测缺点是标签体系一旦不统一匹配就会失效。我建议在项目早期就用规则匹配先把链路跑通。语义匹配是用向量相似度把任务描述和 Agent 能力描述都转成向量算余弦相似度。它适合任务描述比较模糊的场景比如用户说“帮我看看数据有没有异常”语义匹配能把它关联到anomaly_detection能力上。但它的缺点是不稳定同一个任务换个说法选出来的 Agent 可能就变了。混合打分是我目前最推荐的方案先用规则匹配做硬过滤把不满足 schema 和约束的 Agent 剔除再在剩下的候选里用语义相似度排序最后叠加一个历史成功率权重。公式大致是final_score 0.5 * semantic_similarity 0.3 * success_rate 0.2 * (1 / avg_latency)这个权重不是拍脑袋定的。语义相似度占大头是因为它决定了“能力对不对口”成功率占三成是因为一个能力对口但经常失败的 Agent 不该被优先选延迟权重最小是因为在多数业务里正确性比速度更重要。你可以根据自己业务调整但一定要把成功率纳入考量否则系统会反复选中那个“看起来最匹配、实际最不靠谱”的 Agent。2.3 一个容易忽略的点Agent 的“拒绝能力”我在项目里吃过一个亏某个 Agent 被选中后发现任务超出了它的能力边界但它没有拒绝而是硬着头皮返回了一个看似合理、实则错误的结果。这种错误最难排查因为链路上没有任何报错。后来我给所有 Agent 加了一个显式拒绝机制当输入不满足 schema、或任务超出 capability 范围时Agent 必须返回一个标准化的reject响应而不是硬答。调度方收到 reject 后会重新走一遍选择流程选下一个候选。这个机制的价值在于让失败变得可见。宁可让系统多绕一圈也不要让错误结果悄悄流到下游。这是多 Agent 系统和单模型系统在工程上最大的区别之一。3. 协作机制从“互相调用”到“任务编排”3.1 三种协作拓扑链式、星型与网状Agent 之间的协作本质上是在决定控制流怎么走。常见的拓扑有三种。链式是 A 做完交给 BB 做完交给 C。它适合步骤严格串行的任务比如“取数 → 计算 → 生成报告”。优点是逻辑清晰缺点是任何一环失败整条链就断且无法并行。星型是有一个中心调度 Agent所有子 Agent 都只和它通信。子 Agent 之间不直接联系。这是我最推荐的拓扑因为控制权集中状态好管理出错好定位。中心调度方掌握全局任务状态知道每一步进行到哪、谁在做、结果是什么。网状是 Agent 之间可以任意通信。它理论上最灵活但工程上最难维护——调用关系一旦复杂就会出现循环调用、状态不一致、死锁等问题。除非你有非常明确的理由否则不要一上来就做网状。拓扑适用场景优点风险链式严格串行任务逻辑简单单点失败即全断星型多能力协作状态集中、易排查调度方可能成瓶颈网状高度动态任务灵活循环调用、状态混乱3.2 任务状态机让协作过程可追踪星型拓扑能跑起来的关键是中心调度方维护一个任务状态机。每个任务从创建到完成会经历若干状态pending待分派、dispatched已分派、running执行中、succeeded成功、failed失败、rejected被拒绝。状态机的价值不只是“记录”而是驱动下一步动作。比如一个任务处于rejected状态调度方就知道要重新选择 Agent处于failed状态就要判断是重试还是降级处于succeeded就把结果写入记忆并触发下游任务。我一般会把状态机持久化到一张表里字段包括task_id、parent_task_id、agent_id、status、input、output、created_at、updated_at。这样任何一个请求进来我都能顺着parent_task_id把整条协作链路还原出来。排查问题时这张表比任何日志都好用。3.3 结果传递结构化数据优先自然语言兜底Agent 之间传什么是个很容易被忽视但影响巨大的问题。我的经验是能用结构化数据就用结构化数据自然语言只作为兜底。原因很直接。自然语言有歧义A Agent 说“数据略有增长”B Agent 可能理解成 5%也可能理解成 50%。而结构化数据{growth_rate: 0.052}没有歧义。在多 Agent 链路里歧义会被逐级放大最后结果完全不可控。具体做法是每个 Agent 的输出 schema 里核心字段必须是结构化的如果确实需要传递解释性内容就单独放一个notes字段且下游 Agent 默认不依赖它做判断。这样既保留了可读性又保证了链路的确定性。注意不要为了“看起来智能”而让 Agent 之间用自然语言自由对话。自由对话在演示里很酷在生产里是灾难。控制流必须由状态机驱动而不是由模型“聊”出来。4. 记忆放在哪里三层记忆的职责划分4.1 为什么记忆不能只放一个地方多 Agent 系统里“记忆”这个词被用得很泛。有人把对话历史叫记忆有人把向量库叫记忆有人把任务状态也叫记忆。如果不把记忆分层就会出现两个极端要么所有 Agent 都去读同一份巨大的上下文又慢又贵要么每个 Agent 各记各的协作时对不上。我的做法是把记忆分成三层会话记忆、任务记忆、长期记忆。它们服务不同的对象存储不同的内容生命周期也完全不同。4.2 会话记忆面向用户的短期上下文会话记忆是面向用户的记录的是这次交互里用户说过什么、系统回过什么。它的生命周期通常就是一次会话会话结束就可以归档或丢弃。它放在哪里我一般放在调度方而不是各个子 Agent。因为用户只和调度方对话子 Agent 不需要知道用户上一句说了什么。调度方在分派任务时只把和当前任务相关的那部分上下文提取出来作为任务输入传给子 Agent。这样既保证了子 Agent 拿到的信息是干净的也避免了把整段对话历史塞进每个 Agent 的上下文里。4.3 任务记忆协作链路的中间状态任务记忆是面向协作的记录的是每个任务的输入、输出、状态、依赖关系。它就是我前面说的那张状态机表。这层记忆必须持久化而且必须能被所有参与协作的 Agent 按权限读取。为什么因为下游 Agent 经常需要上游的结果。比如生成报告的 Agent需要拿到分析 Agent 算出的指标。如果任务记忆不共享下游就只能重新算一遍既浪费又可能算出不一样的结果。任务记忆的读取要有边界每个 Agent 只能读和它当前任务直接相关的上游结果不能随便翻整条链路。这既是性能考虑也是安全考虑。4.4 长期记忆跨会话的知识沉淀长期记忆是面向系统的记录的是跨会话、跨任务沉淀下来的知识。比如某个用户偏好什么样的报告格式、某类数据查询常用的过滤条件、某个 Agent 在什么情况下容易失败。长期记忆通常用向量库加结构化存储的组合来实现。向量库负责语义检索结构化存储负责精确查询。它的写入要谨慎——不是什么都要记只记那些未来会被复用的信息。我见过一些项目把所有中间结果都往长期记忆里塞结果检索时噪声极大反而降低了效果。记忆层服务对象典型内容生命周期存储方式会话记忆用户对话历史单次会话内存/缓存任务记忆协作链路任务状态与结果任务周期关系型数据库长期记忆整个系统偏好与经验长期向量库结构化库4.5 记忆写入时机成功才写失败要记原因记忆写入的时机也有讲究。我的原则是任务成功时写结果任务失败时写原因。成功的结果写入任务记忆供下游使用同时如果这个结果有复用价值再写入长期记忆。失败的任务结果不写但失败原因一定要写——是输入不合法、超时、还是能力不匹配。这些失败原因积累起来就是优化 Agent 选择策略的依据。前面说的“历史成功率权重”数据就是从这来的。5. 搭一套最小可用的 A2A 系统从零到跑通5.1 组件清单与职责边界要跑通一套最小 A2A 系统你需要的组件其实不多但每个的职责必须清晰。调度器Orchestrator接收用户请求拆解任务选择 Agent维护状态机管理会话记忆。Agent 注册中心存放所有 Agent 的能力描述卡供调度器查询。任务存储持久化任务状态与结果即任务记忆。长期记忆库向量库加结构化库存跨会话知识。若干 Agent 实例各自实现一个能力域暴露统一的调用接口。这里最容易出问题的是调度器和 Agent 的职责边界。我的建议是调度器只负责“分派和协调”不负责“具体计算”Agent 只负责“执行和返回”不负责“决定下一步做什么”。一旦 Agent 开始自己决定调用谁控制流就散了。5.2 一次完整请求的推进过程还是用开头那个例子“查上个月华东区销售数据做同比分析生成 PPT 发给我。”调度器收到请求后先做任务拆解得到三个子任务数据查询、同比分析、报告生成与投递。然后它查注册中心为每个子任务匹配 Agent。数据查询匹配到data_analyst同比分析也匹配到它因为它有ratio_calculation能力报告生成匹配到report_writer。接着调度器按依赖关系排序数据查询先跑拿到结果后触发同比分析分析完成后再触发报告生成。每一步的状态都写入任务存储。报告生成 Agent 需要分析结果时从任务存储里读而不是重新算。整个过程中会话记忆留在调度器任务记忆在任务存储长期记忆只在需要时读写。三层记忆各司其职链路清晰。5.3 关键接口设计示例Agent 的统一调用接口我一般设计成这个样子def invoke(agent_id, task_input, context): agent registry.get(agent_id) if not schema_match(task_input, agent.input_schema): return {status: rejected, reason: input_schema_mismatch} try: result agent.execute(task_input, context) if not schema_match(result, agent.output_schema): return {status: failed, reason: output_schema_mismatch} return {status: succeeded, output: result} except TimeoutError: return {status: failed, reason: timeout}这个接口里有两个校验点入口校验输入 schema出口校验输出 schema。入口校验防止 Agent 接到不该接的任务出口校验防止 Agent 返回不符合约定的结果。这两个校验是多 Agent 系统稳定性的底线千万别省。5.4 跑通之后必须补的三件事链路跑通只是开始。我踩过的坑告诉我下面三件事不补系统上线后一定出问题。第一件是超时与重试策略。每个 Agent 调用都要设超时超时后不能无限重试要有次数上限和退避。我一般设 2 次重试间隔指数增长。第二件是幂等性。同一个任务可能因为重试被多次执行Agent 的执行逻辑必须保证幂等否则会重复写入数据。做法是给每个任务一个唯一task_idAgent 执行前先查这个 id 是否已处理过。第三件是可观测性。每个任务的开始、结束、状态变更都要打点最好能按parent_task_id串成完整链路。没有这个线上出问题你只能靠猜。6. 那些文档里不会写的踩坑经验6.1 Agent 数量不是越多越好我一开始觉得 Agent 越多越灵活结果注册中心里堆了十几个能力高度重叠的 Agent调度器选择时经常在几个“差不多”的 Agent 之间反复横跳同一个任务两次执行选中的 Agent 都不一样结果自然不稳定。后来我做了合并能力重叠度超过 70% 的 Agent 直接合并只保留能力边界最清晰的那几个。合并之后选择稳定性明显提升。经验是Agent 的粒度应该由能力边界决定而不是由实现方便决定。6.2 上下文膨胀是隐形杀手多 Agent 协作时很容易把越来越多的上下文往 Agent 里塞觉得“信息越多判断越准”。实际上上下文一长模型注意力会被稀释关键信息反而被淹没而且成本直线上升。我的做法是给每个 Agent 的输入上下文设一个硬上限只传和当前任务直接相关的字段。需要更多信息时让 Agent 主动去任务记忆里查而不是一股脑塞进去。这个“按需拉取”的模式比“全量推送”稳定得多。6.3 失败要快速暴露不要静默降级有些系统为了“用户体验”在某个 Agent 失败时悄悄降级返回一个凑合的结果。这在多 Agent 链路里非常危险因为降级结果会作为下游的输入错误被逐级放大最后你根本不知道问题出在哪一环。我的原则是失败就失败明确返回失败状态让调度器决定是重试、换 Agent 还是终止。宁可让用户看到一次明确的失败也不要让一个错误结果悄悄流完全程。6.4 记忆的读写权限要收紧任务记忆是共享的但共享不等于谁都能读写。我见过因为权限没收紧某个 Agent 误写了另一个任务的结果导致下游拿到脏数据的情况。做法很简单每个 Agent 只能写自己任务的输出只能读自己任务依赖的上游输出。这个规则在调度器层面强制执行Agent 自己无权越界。多花十分钟做权限校验能省掉后面几天的排查时间。7. 关于 A2A 选型与落地的一点个人判断回到最开始的问题什么时候该上 A2A什么时候不该上。我的判断标准是看任务的能力异构程度。如果一件事需要两种以上差异明显的能力且这些能力用不同模型或工具实现效果更好那 A2A 值得上。如果只是单轮问答或者单一能力硬拆只会增加复杂度。选型上我倾向于先规则后语义、先星型后网状、先单机后分布。规则匹配能跑通就别急着上向量检索星型拓扑能解决就别碰网状单机能扛住就别急着拆服务。多 Agent 系统的复杂度是乘法增长的每加一个维度排查难度就翻一倍。记忆这块我的核心体会是会话记忆归调度器任务记忆归存储层长期记忆归系统。三层不要混混了就会又慢又乱。任务记忆是重中之重它既是协作的纽带也是排查的依据值得你花最多精力去设计。最后分享一个我一直在用的小技巧给每个 Agent 加一个“自检”步骤在执行前先判断自己能不能做这个任务不能做就明确拒绝。这个自检看起来多了一步但它把大量潜在的错误挡在了执行之前整体稳定性反而更高。多 Agent 系统的可靠性往往不是靠“让每个 Agent 更聪明”而是靠“让每个 Agent 更清楚自己的边界”。