分布式A2A架构实战：从单体Agent到多Agent协作的演进与落地-北京尧图网络

1. 从单体到分布式Agent 基础架构为什么必须走这一步1.1 一个 Agent 扛不住的时候问题出在哪我最早做 Agent 项目的时候和大多数人一样先跑通一个单体 Agent 就觉得很满足了。一个进程里塞进规划器、工具调用、记忆模块、模型接口跑个 demo 完全没问题。但一旦把它放到真实业务里问题就像多米诺骨牌一样倒下来。最先崩的是并发。单个 Agent 实例处理一个请求要经历多轮推理循环每一轮都要调用模型、解析输出、执行工具、回填结果。这个过程短则几秒长则几十秒。当同时来几十个请求的时候单实例的吞吐量直接见底。你可能会想那就多开几个实例呗。但问题在于Agent 不是无状态的服务它带着上下文、记忆、会话状态简单地水平复制并不能解决问题反而会带来状态不一致的麻烦。然后是能力边界的问题。一个 Agent 什么都能干意味着它的系统提示词会越来越长工具列表会越来越臃肿模型在选择工具时的准确率会明显下降。我实测过一个 Agent 挂载超过 30 个工具之后工具选择的错误率会从个位数飙升到接近两成。这不是模型不行而是单体的设计本身就违背了“关注点分离”这个基本原则。再往后是可靠性。单体 Agent 一旦某个环节卡住比如某个外部 API 超时整个请求链路就挂在那里。你没法单独重启某个能力模块也没法针对某个环节做降级。这种“一荣俱荣、一损俱损”的结构在生产环境里是致命的。所以分布式 A2A 不是赶时髦而是 Agent 从 demo 走向生产时架构层面必须跨过的那道坎。1.2 A2A 到底解决的是什么问题A2A也就是 Agent-to-Agent核心思路是把一个“全能 Agent”拆成多个“专职 Agent”让它们之间通过标准协议互相协作。你可以把它理解成从“一个人干所有事”变成“一个团队分工协作”。这里的关键词是“标准协议”。如果只是简单地把 Agent 拆开然后让它们用私有接口互相调用那本质上还是紧耦合的分布式单体只是换了个形式。A2A 的价值在于定义了一套 Agent 之间发现、通信、协商、任务分发的规范让不同团队、不同技术栈、甚至不同厂商开发的 Agent 能够互操作。我打个比方。单体 Agent 就像一家什么都卖的杂货铺老板一个人进货、理货、收银、售后。生意小的时候没问题但客流一大就手忙脚乱。分布式 A2A 就像把杂货铺变成一条商业街有专门卖水果的、专门修鞋的、专门配钥匙的每家店只干自己最擅长的事顾客需要什么就去找对应的店。而 A2A 协议就是这条街上的“通用语言”和“路牌系统”保证顾客知道去哪找、店家之间也能互相推荐生意。这个类比里有个容易被忽略的点商业街的效率优势不仅来自分工更来自“发现机制”和“协作机制”。如果顾客不知道修鞋店在哪或者修鞋店需要配钥匙但找不到配钥匙的店那分工反而增加了摩擦。A2A 协议解决的正是这个层面的问题。1.3 哪些场景下分布式 A2A 是刚需不是所有 Agent 项目都需要上分布式 A2A。如果你的场景是单用户、低频次、任务简单的单体 Agent 完全够用强行分布式只会增加复杂度。但以下几类场景分布式 A2A 基本是绕不过去的。第一类是高并发场景。比如面向大量用户的智能客服、智能助手同时在线请求可能上千。单体 Agent 的吞吐量根本撑不住必须把不同能力拆成独立服务各自独立扩缩容。第二类是多能力协作场景。比如一个企业级 Agent 需要同时处理订单查询、库存管理、物流跟踪、售后处理这些能力分属不同系统、不同团队维护。把它们塞进一个 Agent 里既不现实也不合理拆成专职 Agent 通过 A2A 协作才是正解。第三类是异构技术栈场景。企业里往往存在多种技术栈有的 Agent 用 Python 写有的用 Java有的基于 Rust 做高性能推理。A2A 协议让这些异构 Agent 能够互通而不需要统一技术栈。第四类是需要独立演进和治理的场景。不同 Agent 的迭代节奏、安全要求、合规要求可能完全不同。分布式架构让每个 Agent 可以独立部署、独立升级、独立做安全策略不会因为一个模块的改动影响整个系统。1.4 复杂度从哪来分布式不是免费的午餐分布式 A2A 带来的复杂度是实打实的我把它归纳成四个层面。通信复杂度。Agent 之间要通信就要定义消息格式、序列化方式、传输协议、超时策略、重试机制。每多一个 Agent可能的通信路径就多一条调试难度指数级上升。状态复杂度。单体 Agent 的状态都在一个进程里读写直接操作内存。分布式之后状态可能分散在多个 Agent、多个节点、多个存储里一致性怎么保证、冲突怎么解决都是硬骨头。协调复杂度。多个 Agent 协作完成一个任务谁先谁后、谁依赖谁、某个环节失败了怎么补偿这些协调逻辑比单体里的顺序执行复杂得多。运维复杂度。分布式系统要监控的指标更多、要排查的问题更杂、要处理的故障模式更丰富。一个请求跨了五个 Agent最后失败了你得有能力把整条链路串起来看。但我想说的是这些复杂度不是分布式“制造”出来的而是原本就存在于业务需求里只是单体架构把它们掩盖了。当业务简单的时候掩盖没问题当业务复杂到一定程度这些复杂度必然要暴露出来分布式只是让它们显式化、可管理化。所以标题里说“复杂度与必然性”我理解这个“必然性”指的就是业务复杂度到了一定阈值分布式 A2A 就是唯一可行的架构选择没有捷径。2. 核心架构拆解分布式 A2A 的骨架怎么搭2.1 Agent 注册与发现让每个 Agent 都能被找到分布式 A2A 的第一块基石是注册与发现。你想想如果 Agent A 想调用 Agent B 的能力它首先得知道 B 在哪、B 能干什么、B 现在是否可用。这就需要一个注册中心。注册中心的核心数据结构是一张“能力清单”记录每个 Agent 的标识、能力描述、接口地址、健康状态、版本信息。Agent 启动时向注册中心注册自己关闭时注销运行期间定期发送心跳。注册中心负责维护这份清单的实时性。这里有个设计选择能力描述用什么格式。我见过两种做法。一种是用自然语言描述比如“这个 Agent 能处理订单查询”。另一种是用结构化的 schema比如定义好输入参数、输出格式、能力标签。我的经验是纯自然语言描述在 Agent 数量少的时候够用但一旦超过十几个模型在选择 Agent 时就会犯迷糊。结构化 schema 配合自然语言说明效果最好。模型先根据能力标签做粗筛再根据 schema 做精确匹配。发现机制也有两种模式。一种是主动查询Agent A 需要某个能力时去注册中心查“谁能干这个”。另一种是被动推送注册中心在能力清单变化时主动通知相关 Agent。生产环境里通常是两者结合常规调用走主动查询关键能力变更走被动推送。注意注册中心本身必须是高可用的。如果注册中心挂了整个 A2A 网络就瘫痪了。实践中至少部署三个节点用一致性协议保证数据同步。2.2 通信协议选型同步、异步还是流式Agent 之间的通信协议选型直接决定了整个系统的性能和可靠性上限。我把常见的几种模式列出来对比。通信模式适用场景优势劣势同步 HTTP短任务、强实时实现简单、调试方便阻塞等待、超时难处理异步消息队列长任务、高并发解耦彻底、削峰填谷链路追踪复杂、延迟增加流式传输需要中间结果用户体验好、可中断协议复杂、状态管理难事件驱动多 Agent 协作扩展性强、松耦合调试困难、顺序难保证我的建议是默认走异步消息队列对实时性要求高的走同步 HTTP对需要展示中间过程的走流式。不要试图用一种模式解决所有问题。异步消息队列的选择上我踩过的坑是不要用轻量级的队列去做重任务调度。轻量级队列在消息堆积、顺序保证、死信处理上往往不够用。Agent 任务的特点是执行时间长、失败重试多、需要追踪状态这些都对消息中间件提出了更高要求。流式传输这块如果 Agent 需要边推理边输出SSE 或者 WebSocket 都是常见选择。SSE 更简单单向推送够用WebSocket 双向通信更灵活但连接管理更复杂。我一般优先选 SSE除非确实需要双向交互。2.3 任务编排与状态机谁先谁后、失败了怎么办多个 Agent 协作完成一个任务编排逻辑是核心。我见过两种主流做法。一种是中心化编排。有一个 Orchestrator Agent 负责拆解任务、分发给下游 Agent、收集结果、处理异常。这种模式逻辑清晰容易追踪但 Orchestrator 容易成为瓶颈和单点。另一种是去中心化编排。每个 Agent 自己决定下一步调用谁通过共享的状态存储来协调。这种模式扩展性好但调试起来非常痛苦因为任务的执行路径不是预先确定的。我的实践建议是核心业务流程用中心化编排探索性任务用去中心化编排。核心业务流程需要可预测、可审计、可回滚中心化编排更合适。探索性任务比如研究型 Agent路径本来就不确定去中心化更灵活。状态机是编排的底层支撑。每个任务从创建到完成会经历一系列状态待处理、处理中、等待依赖、成功、失败、重试中、已取消。状态之间的转换必须有明确的触发条件和守卫条件。我强烈建议把状态机显式建模而不是散落在各个 Agent 的代码里。显式状态机的好处是状态可查询、转换可审计、异常可恢复。这里有个关键设计幂等性。分布式环境下消息可能重复投递Agent 可能重复执行。每个 Agent 的操作必须是幂等的或者至少要有去重机制。我通常会在任务 ID 上加一层去重表同一个任务 ID 的处理结果只记录一次。2.4 记忆与上下文管理分布式下的状态一致性Agent 的记忆是它区别于普通服务的关键。单体 Agent 的记忆管理相对简单都在一个进程里。分布式之后记忆怎么存、怎么共享、怎么保证一致性就成了大问题。我把 Agent 记忆分成三层。会话记忆是当前对话的上下文生命周期短访问频繁。长期记忆是跨会话的知识积累生命周期长访问相对少。共享记忆是多个 Agent 之间需要共享的状态比如任务进度、中间结果。会话记忆我一般放在 Redis 里读写快支持过期。长期记忆用向量数据库支持语义检索。共享记忆比较麻烦因为它涉及多个 Agent 的并发读写。共享记忆的一致性我试过几种方案。悲观锁简单但性能差Agent 执行时间长锁持有时间也长容易造成阻塞。乐观锁性能好但冲突处理复杂需要 Agent 自己处理版本冲突。最终我倾向于基于版本号的乐观并发控制配合重试机制。每个共享状态带一个版本号Agent 读取时记下版本号写入时检查版本号是否变化变了就重新读取再处理。提示共享记忆不要设计得太细粒度。粒度过细会导致锁竞争激烈粒度过粗会导致不必要的冲突。我一般按“任务”或“会话”作为共享记忆的边界。2.5 安全边界Agent 之间的信任怎么建立分布式 A2A 里Agent 之间的信任关系是个容易被忽视但极其重要的问题。单体 Agent 里所有能力都在一个进程里信任是隐式的。分布式之后Agent A 调用 Agent BA 怎么知道 B 是可信的B 怎么知道 A 有权限调用自己。认证层面我一般用双向认证。每个 Agent 有自己的身份凭证调用时互相验证。凭证的发放和轮换由统一的身份服务管理。不要用简单的 API KeyAPI Key 一旦泄露就是灾难而且轮换麻烦。授权层面需要定义清楚“哪个 Agent 能调用哪个 Agent 的哪个能力”。我通常用基于角色的访问控制给每个 Agent 分配角色角色绑定能力权限。权限检查放在被调用方调用方只负责携带身份信息。数据层面Agent 之间传递的数据可能包含敏感信息。我建议在协议层面支持字段级加密敏感字段在传输前加密只有有权限的 Agent 能解密。这样即使消息被截获敏感信息也不会泄露。审计层面每次 Agent 之间的调用都要记录谁调的、调了什么、什么时候调的、结果如何。审计日志要独立存储不能被 Agent 自己修改。出了问题审计日志是唯一的真相来源。3. 实操落地从零搭一个分布式 A2A 系统3.1 环境准备与技术栈选型假设我们要搭一个企业级的分布式 A2A 系统支持订单查询、库存管理、物流跟踪三个专职 Agent外加一个编排 Agent。我先把技术栈列出来。组件选型理由注册中心Nacos 或 Consul成熟稳定支持健康检查和服务发现消息队列Kafka 或 RabbitMQ支持持久化、顺序保证、死信队列状态存储Redis PostgreSQLRedis 做会话和缓存PG 做持久化向量存储Milvus 或 Qdrant长期记忆的语义检索通信协议gRPC 消息队列gRPC 做同步调用MQ 做异步任务编排引擎自研状态机业务逻辑复杂通用引擎不够灵活监控Prometheus Grafana指标采集和可视化链路追踪OpenTelemetry跨 Agent 的调用链追踪这个选型不是唯一的但每个选择背后都有考量。比如注册中心选 Nacos 而不是 Eureka是因为 Nacos 同时支持服务发现和配置管理减少一个组件。消息队列选 Kafka 而不是 RabbitMQ是因为 Kafka 的持久化和顺序保证更强适合 Agent 任务这种长执行、需要重试的场景。环境准备阶段我建议先用 Docker Compose 把所有依赖跑起来验证连通性。不要一上来就上 Kubernetes调试成本太高。等本地跑通了再考虑容器编排。# docker-compose.yml 片段 services: nacos: image: nacos/nacos-server:latest ports: - 8848:8848 redis: image: redis:7-alpine ports: - 6379:6379 postgres: image: postgres:15 environment: POSTGRES_PASSWORD: agent_pass ports: - 5432:5432 kafka: image: bitnami/kafka:latest ports: - 9092:90923.2 Agent 注册与心跳机制的代码实现注册逻辑的核心是Agent 启动时把自己的能力清单写到注册中心然后起一个后台线程定期发心跳。心跳超时注册中心就把这个 Agent 标记为不可用。import requests import threading import time class AgentRegistry: def __init__(self, registry_url, agent_info): self.registry_url registry_url self.agent_info agent_info self.heartbeat_interval 10 # 秒 def register(self): resp requests.post( f{self.registry_url}/agents/register, jsonself.agent_info ) if resp.status_code ! 200: raise RuntimeError(f注册失败: {resp.text}) print(fAgent {self.agent_info[id]} 注册成功) def heartbeat_loop(self): while True: try: requests.post( f{self.registry_url}/agents/heartbeat, json{id: self.agent_info[id]} ) except Exception as e: print(f心跳发送失败: {e}) time.sleep(self.heartbeat_interval) def start(self): self.register() t threading.Thread(targetself.heartbeat_loop, daemonTrue) t.start()这里有个细节心跳间隔不能太短否则注册中心压力大也不能太长否则故障发现慢。我一般设 10 秒心跳30 秒超时。也就是说一个 Agent 挂掉后最多 30 秒会被标记为不可用。能力清单的结构我建议包含这些字段Agent ID、能力标签列表、接口地址、版本号、健康检查地址、最大并发数。能力标签用标准化的词汇不要各写各的。比如“订单查询”就统一用order.query不要有的写“查订单”有的写“订单查询”。3.3 跨 Agent 调用的完整链路演示假设用户发起一个“查询我的订单并告诉我物流状态”的请求。编排 Agent 收到请求后需要先调订单 Agent 查订单拿到订单号后再调物流 Agent 查物流。class OrchestratorAgent: def __init__(self, registry_client, mq_producer): self.registry registry_client self.mq mq_producer def handle_request(self, user_id, request_text): # 第一步创建任务写入状态存储 task_id self.create_task(user_id, request_text) # 第二步发现订单 Agent order_agent self.registry.discover(order.query) if not order_agent: return self.fail_task(task_id, 订单服务不可用) # 第三步异步调用订单 Agent self.mq.send(agent.order.query, { task_id: task_id, user_id: user_id, callback: orchestrator.order_result }) return {task_id: task_id, status: processing} def on_order_result(self, message): task_id message[task_id] if message[status] ! success: return self.fail_task(task_id, message[error]) order message[data] # 第四步发现物流 Agent logistics_agent self.registry.discover(logistics.track) if not logistics_agent: return self.fail_task(task_id, 物流服务不可用) # 第五步调用物流 Agent self.mq.send(agent.logistics.track, { task_id: task_id, order_id: order[order_id], callback: orchestrator.logistics_result }) def on_logistics_result(self, message): task_id message[task_id] if message[status] ! success: return self.fail_task(task_id, message[error]) # 第六步汇总结果完成任务 self.complete_task(task_id, { order: message[order], logistics: message[data] })这条链路里每个 Agent 都是独立的服务通过消息队列异步通信。编排 Agent 不阻塞等待而是通过回调处理结果。这样做的好处是订单 Agent 和物流 Agent 可以独立扩缩容某个环节慢了不会拖垮整个链路。但异步也带来了复杂度。任务状态需要持久化否则编排 Agent 重启后回调就丢了。我一般把任务状态存在 PostgreSQL 里每个状态转换都写一条记录。这样即使系统重启也能从状态存储里恢复未完成的任务。3.4 分布式锁在 Agent 协作中的实际应用分布式锁在 A2A 系统里用得很多但用错的地方也很多。我举几个实际场景。场景一共享资源的互斥访问。多个 Agent 可能同时需要更新同一个订单的状态。如果不加锁两个 Agent 同时读到“待处理”各自改成“处理中”最后谁覆盖谁不确定。这时候需要一把锁保证同一时刻只有一个 Agent 能更新这个订单。场景二任务去重。同一个任务可能被重复投递Agent 需要保证只处理一次。可以用任务 ID 作为锁的 key拿到锁的 Agent 才处理处理完释放锁。场景三限流。某个外部 API 有调用频率限制多个 Agent 共享这个配额。可以用分布式锁配合计数器控制单位时间内的调用次数。import redis import uuid class DistributedLock: def __init__(self, redis_client, key, ttl30): self.redis redis_client self.key flock:{key} self.ttl ttl self.token str(uuid.uuid4()) def acquire(self, timeout10): end time.time() timeout while time.time() end: # SET NX EX 是原子操作 if self.redis.set(self.key, self.token, nxTrue, exself.ttl): return True time.sleep(0.1) return False def release(self): # 用 Lua 脚本保证检查 token 和删除的原子性 lua if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end self.redis.eval(lua, 1, self.key, self.token)这里有几个坑我必须提醒。第一锁的 TTL 要大于业务执行时间否则业务还没执行完锁就过期了其他 Agent 就能拿到锁互斥就失效了。第二释放锁必须检查 token否则可能释放了别人的锁。第三如果业务执行时间不确定需要实现锁续期机制在业务执行期间定期延长 TTL。注意分布式锁不是银弹。能用无锁方案解决的优先用无锁方案。比如用数据库的唯一约束做去重比分布式锁更简单可靠。锁只在确实需要互斥的场景下用。3.5 监控与链路追踪让问题无处遁形分布式 A2A 系统最怕的就是“出了问题不知道哪出的”。一个请求跨了五个 Agent最后失败了你得有能力把整条链路串起来看。链路追踪的核心是Trace ID。请求进入系统时生成一个全局唯一的 Trace ID之后每次 Agent 之间的调用都携带这个 ID。每个 Agent 在处理时把自己的 Span 信息开始时间、结束时间、状态、关键参数上报到追踪系统。这样你就能看到一个请求在哪些 Agent 上花了多少时间、哪个环节失败了。from opentelemetry import trace tracer trace.get_tracer(a2a.agent) def handle_task(task_id, trace_id): with tracer.start_as_current_span(handle_task) as span: span.set_attribute(task_id, task_id) span.set_attribute(trace_id, trace_id) try: result do_work(task_id) span.set_attribute(status, success) return result except Exception as e: span.set_attribute(status, error) span.set_attribute(error.message, str(e)) raise监控指标方面我重点关注这几个每个 Agent 的请求量、成功率、P95 延迟、错误率消息队列的堆积量、消费延迟注册中心的 Agent 在线数分布式锁的等待时间和超时次数。这些指标能覆盖大部分故障场景。告警策略上不要什么都告警否则告警疲劳。我一般设三级P0 是系统不可用立即电话告警P1 是核心功能受损企业通讯工具告警P2 是性能下降邮件告警。P0 和 P1 的阈值要保守宁可误报不可漏报P2 的阈值可以宽松些避免噪音。4. 踩坑实录分布式 A2A 的常见问题与排查4.1 Agent 调用超时与重试的连锁反应这是我踩过最狠的坑。一个 Agent 调用下游 Agent 超时触发重试重试又超时再重试。如果多个 Agent 同时重试下游 Agent 的负载瞬间翻倍本来只是慢现在直接被打挂。这就是典型的重试风暴。排查这类问题的思路是先看监控确认是哪个 Agent 的请求量异常飙升再看链路追踪确认重试是从哪个环节开始的最后看日志确认超时的具体原因。解决重试风暴我总结了几条经验。第一重试必须带退避不能立即重试。指数退避加随机抖动是标配。第二重试要有上限不能无限重试。我一般设最多三次。第三重试要区分错误类型网络超时可以重试参数错误重试也没用。第四要有熔断机制下游 Agent 连续失败到一定阈值直接熔断不再调用给它恢复的时间。import random import time def retry_with_backoff(func, max_retries3, base_delay1): for attempt in range(max_retries): try: return func() except RetryableError as e: if attempt max_retries - 1: raise # 指数退避随机抖动 delay base_delay * (2 ** attempt) random.uniform(0, 1) time.sleep(delay) except NonRetryableError: raise4.2 状态不一致的典型场景与修复状态不一致在分布式 A2A 里太常见了。我举几个典型场景。场景一任务状态和实际执行不一致。编排 Agent 把任务标记为“处理中”但实际上下游 Agent 已经处理完了只是回调消息丢了。结果任务永远卡在“处理中”。场景二共享记忆的读写冲突。两个 Agent 同时更新同一个共享状态后写的覆盖了先写的先写的更新丢失。场景三跨 Agent 的事务问题。订单 Agent 扣了库存物流 Agent 创建运单失败但库存已经扣了需要回滚。修复状态不一致核心思路是对账和补偿。对账是定期扫描状态存储找出长时间处于中间状态的任务主动查询实际执行情况修正状态。补偿是当某个环节失败时执行反向操作回滚已完成的步骤。我一般会实现一个对账 Agent每隔几分钟跑一次扫描所有超过阈值时间还在“处理中”的任务逐个核查。核查方式根据任务类型不同而不同有的是查下游 Agent 的状态有的是查数据库记录。提示对账 Agent 本身也要幂等。对账过程中可能重复处理同一个任务必须保证重复对账不会产生副作用。4.3 消息丢失与重复消费的处理消息队列不是绝对可靠的。网络抖动、Broker 重启、消费者崩溃都可能导致消息丢失或重复。Agent 系统必须能容忍这两种情况。消息丢失的防护我一般做三层。第一层是生产者确认消息发送后要等 Broker 确认收到没确认就重发。第二层是持久化消息写入磁盘后才算成功。第三层是消费者手动确认处理完业务逻辑后才确认消息处理失败就不确认消息会重新投递。重复消费的防护核心是幂等。每个消息带一个唯一 ID消费者处理前先查这个 ID 是否处理过处理过就直接跳过。幂等表可以用 Redis 或数据库实现关键是查询和写入要原子。def consume_message(message): msg_id message[id] # 用 SET NX 做幂等检查 if not redis.set(fconsumed:{msg_id}, 1, nxTrue, ex86400): print(f消息 {msg_id} 已处理过跳过) return try: process(message) except Exception: # 处理失败删除幂等标记允许重试 redis.delete(fconsumed:{msg_id}) raise这里有个细节幂等标记的过期时间要足够长至少覆盖消息可能重投的时间窗口。我一般设 24 小时。4.4 常见问题速查表问题现象可能原因排查方向解决方案Agent 调用超时下游负载高、网络慢、死锁看下游指标、链路追踪加超时、退避重试、熔断任务卡在处理中回调丢失、状态未更新查状态存储、消息队列对账 Agent 修正状态状态不一致并发写冲突、事务未回滚查版本号、审计日志乐观锁、补偿事务消息重复消费消费者崩溃、确认丢失查幂等表、消费日志幂等处理、去重表注册中心 Agent 掉线心跳超时、网络分区查心跳日志、网络调整心跳参数、网络修复分布式锁失效TTL 过期、误释放查锁日志、业务耗时锁续期、token 校验重试风暴无退避、无上限查请求量曲线指数退避、熔断链路追踪断链Trace ID 未传递查各 Agent 日志统一传递 Trace ID4.5 几个我踩过的独家坑坑一注册中心的健康检查太激进。我一开始把健康检查间隔设成 5 秒超时 3 秒。结果网络稍微抖一下Agent 就被标记为不可用流量切走然后又切回来来回震荡。后来改成 10 秒间隔、30 秒超时稳定多了。健康检查的参数要根据实际网络质量调整不能照搬文档。坑二消息队列的顺序保证被忽略。Agent 任务有时候需要保证顺序比如先创建订单再更新订单。我一开始用普通队列消息顺序不保证导致更新订单的消息先到创建订单的消息后到更新失败。后来改用带分区键的队列同一个订单的消息路由到同一个分区顺序才有保证。坑三分布式锁的 TTL 设太短。有个业务处理要 20 秒我把锁 TTL 设成 10 秒。结果业务还没处理完锁就过期了另一个 Agent 拿到锁两个 Agent 同时操作数据乱了。后来改成 TTL 大于业务最大耗时并且加了锁续期。坑四链路追踪的采样率设太高。全量采样导致追踪系统压力巨大反而影响了业务。后来改成自适应采样正常请求采样 1%错误请求全采样。既保证了问题可追踪又不影响性能。坑五Agent 版本升级没有灰度。直接全量升级新版本有 bug整个系统挂了。后来改成灰度发布先升级一个实例观察一段时间没问题再逐步扩大。分布式系统里任何变更都要灰度。5. 复杂度与必然性的再思考5.1 什么时候该上分布式 A2A我经常被问我的 Agent 项目要不要上分布式 A2A。我的回答是看三个指标。并发量。如果峰值并发超过单实例能承受的上限就必须分布式。单实例的并发上限取决于模型调用延迟、工具执行时间、内存占用。我一般实测单实例能稳定处理 10 到 20 个并发请求超过这个数就要考虑拆分。能力数量。如果 Agent 需要挂载的工具超过 20 个或者需要协作的能力模块超过 5 个单体 Agent 的工具选择准确率会明显下降这时候拆分是必要的。团队规模。如果多个团队需要独立开发和维护不同的能力模块分布式 A2A 能让各团队独立演进减少协调成本。三个指标满足任意两个我就建议上分布式 A2A。只满足一个可以先观望用单体加一些优化手段撑一撑。5.2 复杂度控制的几个原则分布式 A2A 的复杂度是必然的但可以控制。我总结了几条原则。原则一能少一个 Agent 就少一个。每多一个 Agent就多一条通信链路、多一个故障点、多一份运维成本。拆分要基于真实的业务边界不要为了分布式而分布式。原则二协议要统一实现可异构。A2A 协议统一保证互操作性。但每个 Agent 的内部实现可以自由选择技术栈不要强求统一。原则三状态尽量集中逻辑尽量分散。状态集中存储便于一致性和审计。逻辑分散到各个 Agent便于独立演进。原则四可观测性优先。分布式系统的调试难度远高于单体必须在架构设计阶段就把监控、追踪、日志做进去不能等出了问题再补。原则五渐进式演进。不要一开始就设计一个完美的分布式架构。从单体开始遇到瓶颈再拆拆的时候保持接口稳定。我见过太多项目一开始就上微服务结果复杂度爆炸项目直接烂尾。5.3 我个人的经验体会做了这么多 Agent 项目我最大的体会是分布式 A2A 的难点不在技术而在边界划分。技术方案再复杂都有成熟的组件和模式可以套用。但怎么把一个业务拆成合理的 Agent 边界这个没有标准答案只能靠对业务的深入理解。我见过拆得太细的一个简单任务要跨十几个 Agent通信开销比业务逻辑还大。也见过拆得太粗的跟单体没区别分布式的好处一点没享受到。合理的边界应该是每个 Agent 有明确的职责Agent 之间的交互次数可控单个 Agent 的复杂度在可维护范围内。另一个体会是分布式 A2A 的成熟度取决于运维能力。架构设计得再好运维跟不上系统照样不稳定。监控、告警、对账、灰度、回滚这些运维能力必须同步建设。我一般建议团队在分布式改造之前先把运维体系搭起来否则就是给自己挖坑。最后一个体会不要低估状态管理的难度。分布式系统里状态是最容易出问题的地方。我现在的做法是能无状态就无状态必须带状态的状态存储和业务逻辑严格分离状态变更全部走统一的状态管理服务。这样虽然增加了一层抽象但换来的是可控性和可调试性。这个方向后续还可以扩展的点很多比如 Agent 之间的协商机制、动态能力发现、跨组织的 A2A 联邦每一个都值得单独拿出来聊。但核心思路是一样的用标准协议连接专职 Agent用工程手段管理分布式复杂度让系统在业务增长时能够平滑扩展。

分布式A2A架构实战：从单体Agent到多Agent协作的演进与落地

相关推荐

基于华为昇腾与Xing4.0-29B的私有化编程数学教学Agent实战

自研数据标注工具实战：基于AiLabel.js实现目标检测数据集高效标注

Vue项目接入华视身份证读卡器：从浏览器安全到WebSocket中间服务的完整实践

更多资讯

开源 WMS 合规交付实务：JeeWMS 仓库管理系统在 GPL-3.0 下的源码台账与交付清单

HOG+SVM目标检测原理与工业级实战指南

WPF树状表格实现：DataGrid多列与层级缩进的最小工程

焊接结构疲劳评估实战：nCode DesignLife从应力到寿命的完整闭环

随机振动疲劳分析怎么做？nCode DesignLife频域法全流程详解

Space Bunny：匿名模型服务与OpenAI兼容协议接入实践

今日更新

CMOS设计三大物理效应：体效应、沟道长度调制与亚阈值导电

微信小程序在线阅读毕设全解析：Java后端+MySQL+避坑实战

Android音频问题五层分析法：从App到Codec的系统化诊断

本周精选

CMOS设计三大物理效应：体效应、沟道长度调制与亚阈值导电

微信小程序在线阅读毕设全解析：Java后端+MySQL+避坑实战

Android音频问题五层分析法：从App到Codec的系统化诊断

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？