AI Agent系统A2A通信与Token资源优化实践指南

A2A通信这个词最近在AI圈里频繁出现但说实话真正把Agent之间通信讲透彻、同时把资源优化落地的文章并不多。我自己先后搭过三套智能体系统从最早的单体Agent到多Agent协作从裸调LLM到搭建完整的A2A通信链路踩过不少坑也沉淀了一些比较实用的经验。这几个月正好在做一版面向生产环境的智能体系统重构从通信协议选型、消息设计到算力和Token资源的调度走完了一遍完整的流程。这篇就把这方面的实践做个梳理重点聊聊AI Agent系统里A2A通信为什么这么关键、资源优化到底在优化什么以及一套可以直接复用的实现方案适合正准备从单体Agent迁移到多Agent协作、或者接手了一套Agent系统但不知道怎么下手的读者。1. 智能体系统为什么绕不开A2A通信1.1 单体Agent的瓶颈什么都能干等于什么都干不好早期做Agent最简单粗暴的做法是把所有任务塞给一个Agent让它自己决定怎么调用工具、怎么组织回答。单个Agent的能力确实越来越强但一旦业务复杂起来问题就暴露了。我遇到过最典型的场景一个Agent既要理解用户意图又要去检索知识库还要调用外部API最后还得生成答案。这个链路一旦超过五六步任何一步出错都会沿着链路往下传播最终输出质量很难控制。更麻烦的是上下文膨胀。LLM的上下文窗口是有限的单Agent把所有中间结果都塞进同一个上下文几轮对话之后历史信息就把窗口塞满了。这时候再塞新的任务指令模型要么忘掉前面的关键信息要么直接超长报错。我见过一个线上Agent系统运行了大概两小时上下文膨胀直接导致单次请求耗时从2秒涨到了9秒Token消耗翻了4倍。这就是典型的单体架构撑不住规模化业务的信号。把单体拆成多个专业Agent让每个Agent只做一件事问题就能被拆解得比较干净。但拆完之后的第一个难题就是这些Agent之间怎么配合。这就引出了A2A通信。1.2 A2A通信要回答的三个核心问题A2A全称Agent-to-Agent解决的是智能体之间相互协作的问题。我自己梳理下来一套完整的A2A通信体系至少要回答三个问题。第一个问题Agent之间怎么互相发现。A刚上线它怎么知道B存在B能处理哪些任务是直接配置B的地址还是通过一个注册中心去查找这就像新同事入职你得知道他负责什么遇到问题要找谁。没有一套服务发现机制Agent之间就只能靠硬编码维护成本极高。第二个问题消息格式和协议怎么定义。A要给B派一个任务这个任务以什么结构传递JSONProtobuf是同步等待结果还是异步回调如果A调B超时了怎么办没有统一的消息协议两个Agent之间就像两个讲不同语言的人对话根本没法协作。第三个问题协作模式怎么组织。是A调用BB调用C这种链式调用还是有一个中央调度器统一编排是同步阻塞等B返回还是异步消息队列慢慢消费这个选择直接决定了系统的吞吐上限和链路时延。这三个问题不解决多Agent不是协作是多了一个更复杂的混乱系统。我在第二套系统上就吃过这个亏当时把任务拆给了4个Agent但通信全靠REST接口硬调结果Agent之间互相等待、超时、重试最后系统吞吐量比单体还低。1.3 通信方式的取舍没有万金油只有适不适合结合我自己的实践主流的A2A通信方式大致有四类各有各的适用场景。通信方式优点缺点适用场景HTTP/REST同步调用实现简单、调试方便、生态成熟同步阻塞、链路上任一节点变慢都会拖垮整条链路Agent数量少、调用链短、对实时性要求高的场景消息队列异步通信解耦彻底、削峰填谷、天然支持背压时延增加、消息顺序难以保证、排查链路麻烦请求量大、链路长、允许秒级延迟的场景gRPC性能高、强类型接口、支持双向流调试麻烦、浏览器端支持差、错误处理复杂后端Agent之间高频内部调用、对性能有极致要求的场景事件总线/发布订阅实时性强、扩展Agent方便语义复杂、消息丢失风险需额外补偿机制事件驱动型场景、多个Agent对同一事件感兴趣的场景我现在的做法是混合使用关键链路上用gRPC保证性能非关键路径和跨模块通知走消息队列对外暴露的统一接口用HTTP。没有一种通信方式能解决所有问题关键是想清楚每个链路的时延要求和解耦需求。2. 资源优化到底在优化什么从Token预算到算力调度2.1 容易被忽视的第四类资源Token谈到资源优化大家第一反应总是算力也就是CPU和GPU。但做智能体系统的都知道真正卡脖子的资源往往是Token。每一个LLM调用都在烧Token而Token直接等于真金白银。我算过一笔账假设一个客服系统的请求平均需要经过5次Agent调用每次调用平均消耗600个Token一天10万次请求那一天就是100万次调用总共消耗6亿Token。按常见的模型定价输入Token大约在几美元每百万Token的级别一天就是几百美元。这还是保守估算实际上加上工具调用返回的中间结果、系统提示词、历史对话Token消耗通常会超出预期30%到50%。更关键的是Token不只是成本问题还是性能问题。高Token消耗意味着更长的生成时间更深入的上下文窗口管理。如果系统里每个Agent都在重复传递大段历史记录那内存和带宽也会跟着遭殃。所以我现在做资源优化第一步不是看GPU利用率而是看Token的流向和消耗。先把Token的消耗路径摸清楚整个系统的成本模型和性能瓶颈基本就浮出水面了。2.2 一个成本模型的计算案例这里用一个实际的计算案例来说明怎么量化Token开销。假设一个内容审核智能体系统三个Agent协作质检Agent负责判断内容安全问题摘要Agent负责生成内容摘要分类Agent负责打标签。单次请求的Token消耗大致如下环节输入Token输出Token费用系数质检Agent120080输入输出摘要Agent2000300输入输出分类Agent60050输入输出单次请求总消耗3800输入Token 430输出Token 4230 Token。假设每百万输入Token价格3美元每百万输出Token价格6美元那么单次请求的Token成本大约是(3800 × 3 430 × 6) ÷ 1000000 ≈ 0.0114 0.00258 0.01398美元一天如果处理20万次请求那一天就是2796美元一个月就是约8.4万美元。这个数字一算出来你就明白为什么资源优化对智能体系统这么重要了。但这还不是全部。我还没有算缓存命中、上下文压缩带来的收益。如果给每个Agent配一层语义缓存重复内容直接命中Token消耗可以再降30%以上。2.3 资源优化的分层架构四层协同我在实际项目中把资源优化分成了四个层面每一层解决不同的问题。第一层请求层优化。主要做缓存、去重、批量处理。比如同一个问题短时间内反复出现可以直接缓存结果。这是成本最低、见效最快的优化手段。第二层模型层优化。根据任务复杂度选择不同规格的模型。简单分类任务用轻量模型复杂推理才用强模型而不是所有请求都上最贵的模型。没有一套系统需要所有请求都用最强的模型。这个选择和路由策略直接挂钩——先判断任务难度再决定该调用哪一档模型。第三层上下文层优化。做上下文压缩、摘要提取、滑动窗口裁剪。一个长对话从头到尾全部传给模型是最浪费的做法。维护一个关键信息摘要加上最近几轮完整对话效果远比全量传参好Token消耗能省一半。第四层调度层优化。生命周期管理、并发控制、优先级调度。比如控制同一模型的最大并发数量避免请求突增打爆配额给关键任务设置高优先级在资源紧张时优先保障。这四个层面不是孤立工作的。请求层的缓存决定是否能少调一次模型模型层的路由决定这次调用用什么价位的模型上下文层决定传多长的内容调度层决定什么时候调用、并发多少。只有四层同时起作用资源优化的收益才能最大化。3. 实操搭建一套带A2A通信的资源优化智能体系统3.1 场景设计与架构选型这一部分我拿一个真实做过的项目来拆解一个多Agent协作的智能工单处理系统。系统里包含四个Agent意图理解Agent负责判断工单类型和紧急程度知识检索Agent负责在内部知识库中查找解决方案生成回复Agent负责组装最终答案质检Agent负责检查回复质量。这四个Agent的协作关系是意图理解Agent先分析工单把分析结果发送给知识检索Agent知识检索Agent检索完把结果打包发给生成回复Agent生成回复Agent写完内容后交给质检Agent审核审核通过后对外输出。这个场景有一个明确的特点调用链是固定的不需要动态编排但每个环节都可能成为性能瓶颈。所以架构上我选了混合通信方案——链路内部用gRPC同步调用确保低时延但每个环节之前加一层Redis Streams做缓冲既能在高峰期削峰又能在某个Agent故障时让消息在队列里等一会儿而不至于直接超时失败。3.2 A2A消息协议设计一份能落地的消息结构A2A通信最怕消息结构五花八门。我一开始没统一规定消息格式每个Agent自行定义结果联调的时候光是字段映射就花了两天。第二次做就直接统一了一套消息结构后续再也没在这个问题上返工。我用的消息结构大致是这个样子{ task_id: task_20250321_0001, trace_id: trace_8f3a2b, source: intent_agent, target: knowledge_agent, type: task_request, payload: { intent: password_reset, user_context: {...}, query: 用户忘记密码怎么重置 }, priority: 2, deadline: 5000, timestamp: 1742529600123 }这里几个字段是反复推敲过的trace_id是排障的生命线。没有这个字段跨Agent的调用链根本没法追踪。有了trace_id哪怕消息经过了三个Agent也能用日志平台把整条链路拉出来。priority字段用于优先级调度。工单系统里有VIP客户消息和普通客户消息紧急程度差异很大。调度器读到这个字段后会在Token资源和计算资源紧张时优先保障高优先级任务。deadline是超时控制的关键。每个消息都有一个期望完成时间负责调度的Agent读这个字段如果预估处理时间超过deadline就提前降级或拒绝避免后面的Agent傻等。payload部分尽量只放必要的信息不附带无关的上下文字段。这一点对控制Token成本非常关键——A2A通信里传输的内容越长后续Agent把它拼进提示词时消耗的Token就越多。通信信道要精简传过去的应该是结构化数据而不应该是大段大段未处理的历史上下文。3.3 Agent服务发现与能力注册多Agent系统里服务发现是通信能不能跑通的先决条件。我在第一个版本里用的是直接写死目标地址的方式意图理解Agent直接调知识检索Agent的IP和端口。结果后期加了一个新的知识库Agent所有调用方的配置都要改一遍非常痛苦。现在改用注册中心来管理。每个Agent启动时向注册中心注册自己的服务名、地址、能力描述和负载状态。调用方只通过服务名获取一组可用实例然后由负载均衡策略选一个实例去调用。注册数据的结构大致是{ agent_name: knowledge_agent, version: v2.1.0, address: 10.0.3.15:9090, capabilities: [intent_lookup, similarity_search, doc_retrieval], status: healthy, load: 0.42, updated_at: 1742529600000 }关键字段是capabilities和load。load字段代表当前负载情况调度器读到这个数据后会把新任务分配给负载较低的Agent实例。这个是实现资源均衡的基础如果没有这个字段调度只能随机分配很容易出现一个实例忙到冒烟、另一个实例闲到摸鱼的情况。注册中心的选型上如果系统规模不大用Consul或者Etcd都够用规模大了我建议迁移到Nacos或者Kubernetes原生的服务发现机制但原理是一致的——服务注册、健康检查、负载感知。3.4 资源调度与限流实现Token桶与优先级队列通信链路跑通之后接下来就是我最想分享的部分资源调度。目标是让系统在高并发下还能稳定运行不被打爆。我采用了一个双层次的调度器外层用Token桶算法做全局限流内层用优先级队列做任务分类调度。Token桶算法的原理可以用一个水桶来类比桶里有一定数量的Token每个请求进来都要拿一个Token拿不到就排队或者拒绝。Token会以固定速率补充相当于水龙头往桶里加水。这个机制的好处是既能平滑突发流量又能限制长时间的平均速率。实际的调度逻辑写成伪代码大致是这样class TokenBucket: def __init__(self, capacity, refill_rate): self.capacity capacity self.tokens capacity self.refill_rate refill_rate # token/s self.last_refill time.time() def try_acquire(self): now time.time() self.tokens min(self.capacity, self.tokens (now - self.last_refill) * self.refill_rate) self.last_refill now if self.tokens 1: self.tokens - 1 return True return False容量和补充速率的设置需要根据实际流量来估算。公式是桶容量 单次业务峰值请求数 × (1 冗余系数)比如业务峰值时一秒钟会涌进500个请求那我一般把容量设成600到700留出20%到40%的冗余空间。补充速率等于系统稳定状态下的每秒处理能力比如下游Agent每秒最多处理300个请求那补充速率就设为300。内层的优先级队列实现起来也很直接。队列里每个消息带着priority字段出队时选择优先级最高的消息先出。但要注意一个坑如果高优先级任务永远都有低优先级任务会一直饿死。所以我会在每个低优先级任务进入队列时给它一个初始等待时间超过规定时间还没被处理就把它的优先级自动提升一级保证所有任务最终都会被处理只是时间早晚的问题。3.5 并发水位计算别拍脑袋定参数调度器最核心的一个参数是系统能扛住多少并发。这个数不是拍脑袋定的可以算出来。系统的并发上限取决于两个因素每个请求的处理时间以及单位时间内能新增多少请求。假设单个请求在系统里的平均处理时间是1.2秒我们希望系统同时处理的请求数不超过X那么每秒能处理的请求数就是 X / 1.2。如果业务要求每秒至少处理100个请求那么X至少要大于120。但这里还有个更重要的限制Token资源。假设单次请求平均消耗4000 Token模型支持的最大并发请求数是30那系统每秒最多消耗的Token就是30 × 4000 / 1.2 100000 Token每秒。如果模型配额只允许每秒消耗80000 Token那么并发上限就只能下调到24。所以我现在的做法是先算计算资源的限制再算Token配额的限制最后取两个值中的小者作为并发上限。计算资源的限制公式并发上限 平均请求处理时间下的资源利用率上限 × 可用计算资源 / 单请求计算资源消耗Token配额的限制公式并发上限 Token配额 / (单请求Token消耗 × 平均请求频率)这两个值取小就是系统的最终瓶颈。我遇到过一个项目计算资源明明很充裕但模型API配额限制了Token消耗导致整个系统的吞吐量上不去。这种问题如果不做量化分析很难定位到真正的瓶颈在哪里。4. 常见问题排查与调优实录4.1 消息风暴Agent之间互相循环调用我真实遇到过的最坑的问题就是消息风暴。当时质检Agent发现回复有质量问题会发消息让生成回复Agent重新生成生成回复Agent生成后再次发给质检Agent质检Agent再判如果还是不满意就再发一次。结果这两个Agent之间形成了一个循环消息量呈指数增长队列瞬间被打爆。排查的思路是看调用链日志里哪两个Agent之间出现了高频互调然后检查是不是有决策分支没有处理终止条件。修复方法有两个层面一是代码层面加上最大重试次数超过次数就直接转人工处理二是调度层加上消息去重和频控同一个trace_id在短时间内的重复消息直接丢弃或者合并。这个教训让我养成了一个习惯任何Agent发出消息前都先检查消息里带没带trace_id和重试计数器。没有这两个字段A2A通信就像一个没有刹车系统的车迟早要出事。4.2 队列积压与背压机制缺失另一个高频问题就是队列积压。异步消息架构下生产者的速度往往大于消费者的速度如果不对消费者做保护队列里的消息越积越多最终要么内存溢出要么消息超时集体作废。第一次遇到这个问题时我很困惑系统明明没有报错但响应速度越来越慢。后来查监控发现Redis Streams里的消息堆积量一直在涨消费者Agent的处理速度始终跟不上。根因是消费者Agent内部调LLM的并发数没有限制一旦并发过高模型API开始限流返回错误消费者不断重试反而把资源耗光了。解决方法是加背压机制。消费者在处理完一条消息之前不要从队列里拉取新消息同时限制消费者能拉取的最大消息数模型API返回限流错误时退避重试而不是蜂拥重试。说白了就是让消费者根据自己的真实处理能力来决定拉取多少消息而不是盲目接受队列灌入的全部数据。4.3 Token成本失控只见成本涨不见质量升第三个大坑是Token成本失控。有一次我在压测之后发现系统稳定运行一段时间后Token成本比预期翻了一倍。排查发现各个Agent在传递上下文时根本没有做裁剪意图理解Agent把用户原话全部传给了知识检索Agent知识检索Agent又把整段检索结果、相关文档段落全部打包传给了生成回复Agent层层传递Token自然水涨船高。这个问题通过上下文精简和摘要机制解决。在Agent链路里增加一道上下文处理环节每次传递前先做一次压缩把全量内容摘要成500字以内的要点只把关键信息传给下游。实测下来Token成本直接降了45%而任务完成质量没有明显下降。一个很实用的小技巧给每个Agent配置独立的Token预算上限超了预算就强制走摘要链路或者直接拒绝多余上下文。代码层面可以在Agent里加一个Token计数器和截断逻辑每次调用LLM之前先算好这次输入大概要花多少Token如果超过预算就启动摘要逻辑。这个机制每次都能帮我提前发现问题而不是等到月末账单出来才后悔。4.4 一次完整的调优实测列一组我实际调优前后的对比数据给还没有经历过这个过程的读者一个直观感受。指标调优前调优后变化单请求平均时延3.2秒1.4秒下降56%Token成本/万次请求86美元47美元下降45%系统峰值吞吐180请求/秒420请求/秒提升133%队列积压率12%0.3%大幅下降Agent可用性98.2%99.9%恢复更快、更稳这个调优不是哪一个单一改动带来的而是多个措施叠加的结果四层资源优化体系让Token消耗降下来了Token消耗降下来之后生成时延跟着下降时延下降带动了并发能力的提升并发能力提升之后队列积压也就基本消失了。整个链路是一个良性循环。5. 一些踩坑后的实操心得最后又想起来几个细节不适合单独开章节去讲但确实是实际使用过程中很关键的点。第一点日志和可观测性系统一定要在第一天就搭好不要等到出现问题再补。多Agent系统的排障难度远比单体应用高没有全链路追踪一个请求经过三个Agent出了问题根本没法定位是在哪个环节。我现在所有的A2A消息都强制带trace_id日志平台里一条trace_id能拉出整条调用链排障效率提高了不止一倍。第二点先跑通再优化别一开始就追求理想架构。我见过太多团队一上来就搞K8s、搞服务网格、搞完整的消息总线结果复杂度盖过了收益系统根本没跑起来。我第二次做这套系统时的经验是先把HTTP直连版本跑通确认业务逻辑没问题了再逐步引入服务发现、消息队列、调度器。每一步都有明确的价值验证而不是为了技术先进性而引入复杂度。第三点模型能力分层调度是这个系统里性价比最高的一环。不是所有请求都需要最贵的模型。把简单任务路由到轻量模型把复杂任务路由到强模型资源成本和响应速度都能有显著改善。这个思路适用于几乎所有Agent系统哪怕你的系统还很小也值得尽早把这个机制设计进去。第四点也是最后一点定期做一次Token消耗审计。建一个每周跑一次的巡检脚本统计所有Agent的Token消耗趋势。消耗异常的Agent要尽早排查别等到月底看账单才后知后觉。这个习惯帮我养成了对系统资源的持续关注也让我能在成本失控之前就及时调整。