Agent-Reach：多Agent协作框架的调度、通信与可观测性实践

前阵子帮朋友调一个客服自动化的项目里面的Agent一直在“假装工作”——看起来回答得头头是道实际上经常忘了最初的需求上下文一长就开始胡言乱语。我意识到这不是模型选型的问题而是把太多职责塞给了一个Agent。后来我搭了一套多Agent协作框架也就是今天要聊的Agent-Reach。Agent-Reach解决的是多Agent系统里最朴素也最容易被忽视的问题多个智能体怎么找到彼此、怎么交接任务、怎么保证整个链路不丢上下文、不重复执行、不循环空转。它本身不是一个模型而是一套协作骨架把调度、通信、记忆、幂等、可观测性这些工程问题一次性框住。如果你正在做多Agent应用被Agent间通信混乱、状态不同步、任务重复执行折磨过这篇文章值得你花十分钟看完。我会从架构设计、消息协议、代码实现到实际踩坑的完整链路都摊开来讲。1. Agent-Reach的来由单Agent跑不动长链路任务1.1 拆掉“万能Agent”这个幻想很多人第一次接触Agent的时候脑子里想的都是“一个超级智能体什么都能干”。想法很美实际操作下来你会发现单Agent处理短任务确实顺手但一旦面对长链路、多步骤、多知识域的任务问题就出现了。我举一个真实场景一个电商售后的工单用户说“我昨天买的耳机左耳没声音想换货但发票丢了”。这个需求至少要拆成好几步——先查订单再判断是否在质保期然后确认换货政策还要处理发票丢失的后续流程最后生成回复并归档。如果放在一个Agent里它需要在一次推理里同时完成查库、政策匹配、情绪感知、话术生成、流程判断上下文窗口很快就被塞满模型开始遗忘最前面的订单信息回答自然就飘了。这就是单Agent的硬伤职责太多导致指令纠缠、上下文溢出导致记忆衰减、一步出错很难定位是哪个环节的问题。你会发现所有错误都变成一个黑盒你根本不知道它到底是在查订单时错的还是在写话术时错的。1.2 多Agent协作的真正难点不在“模型”而在“触达”既然单Agent不行那就拆成多个Agent。一个查订单一个查政策一个生成话术一个做质检看起来逻辑清晰。但拆完之后你马上会撞上另一个墙这些Agent之间怎么通信先别笑这真的是很多团队第一次搭多Agent系统时卡住的地方。你定义了一个订单Agent和一个话术Agent任务进来之后谁来调度订单Agent的结果怎么交给话术Agent如果话术Agent发现信息不足它能不能主动找订单Agent要两个Agent同时处理同一个任务时怎么保证不重复这些问题的本质不是模型智能不够而是“触达”机制缺失——每个Agent像是一座孤岛没有路没有邮差没有共同语言。我把这套机制叫做Agent-Reach直译过来就是“智能体的触达范围”。它要回答三个问题谁能被触达、触达消息长什么样、触达之后如何确认成功。这三个问题解决不了再多Agent也是各干各的甚至互相帮倒忙。1.3 我给Agent-Reach定下的三个设计原则在动手写代码之前我先给自己立了三条规矩后面所有设计决策都围绕它们展开。第一每个Agent只负责一个语义边界清晰的角色。订单Agent只回答订单相关的问题政策Agent只输出政策结论谁也别越界。这不是限制模型的灵活性而是为了可维护性——边界清晰才能独立测试、独立升级、独立回滚。第二所有通信必须有协议、有记录、可重放。Agent之间不能直接开一个私聊窗口所有消息都要经过消息总线带版本号、带trace_id、带幂等键。没有记录就没有排障依据没有幂等键就一定会重复执行。第三调度器做兜底决策Agent之间有条件的协商。任务分给谁、什么时候推进下一步由调度器统一决定。Agent之间有澄清需求时允许有限轮次的协商但不能无限循环下去。这三条原则后来被证明是这套系统能活下来的关键。尤其是第三条我在第5章会详细讲没有轮次上限的协商系统最终一定会演变成两个Agent互相刷屏的调用风暴。2. 架构与通信Agent-Reach的消息协议怎么设计2.1 五个核心组件各管一段Agent-Reach的整体架构不复杂核心是五个组件各管一段不重叠。组件职责备注Orchestrator调度器任务分解、DAG推进、状态记录全局唯一无状态化设计Worker工作Agent真正执行某个语义角色的模型逻辑可水平扩展注册后由调度器分配任务Message Bus消息总线Agent与调度器之间的所有消息通道同时承载任务流和事件流Registry注册中心记录每个Agent的能力声明、状态、心跳调度器做路由时的“通讯录”Memory Store记忆存储长期记忆、会话隔离、向量检索分共享区和隔离区这五个组件拆开的逻辑是调度器不干活只分活Worker不决策只执行消息总线不存业务逻辑只保证消息可靠地从一个点到另一个点注册中心是信息的来源记忆存储是上下文的归宿。职责越单一出问题时越容易定位。以一次完整任务为例用户提交一个售后工单Orchestrator把它拆成三个子任务——查询订单、核对政策、生成回复。调度器根据Registry里注册的Agent能力描述把查询订单子任务发给订单Agent的消息队列订单Agent消费消息、执行、把结果写回消息总线。Orchestrator监听结果事件确认任务完成后推进到下一个节点继续下发政策核对任务。整个链路就像一条流水线每个工位只做自己的事流转逻辑全部在调度器手里。2.2 消息封套一次触达要带哪些字段消息协议是整个Agent-Reach最值得细细说的地方。我的原则是宁可消息长一点也不要让接收方靠猜。每个消息统一用一个封套包裹字段固定语义明确。# agent_reach/protocol.py from dataclasses import dataclass, field from enum import Enum from typing import Any import uuid from datetime import datetime, timezone class MessageType(str, Enum): REGISTER agent.register TASK_ASSIGN task.assign TASK_ACK task.ack TASK_RESULT task.result EVENT_PUBLISH event.publish NEGOTIATE agent.negotiate HEARTBEAT agent.heartbeat dataclass class Envelope: msg_id: str field(default_factorylambda: uuid.uuid4().hex) msg_type: MessageType MessageType.TASK_ASSIGN source_agent: str target_agent: str # 空表示由调度器按能力路由 trace_id: str task_id: str created_at: str field(default_factorylambda: datetime.now(timezone.utc).isoformat()) version: int 1 payload: dict[str, Any] field(default_factorydict)这个封套里每个字段都不是摆设。msg_id用于消息层面的去重task_id对应的是一个业务子任务同一个任务被重试时task_id不变这是幂等判断的锚点trace_id从最外层入口生成贯穿所有子任务一次用户请求产生的全部消息都能用它串成一条线排障时靠它过滤日志就够了。还有一个容易被忽略但很重要的字段version。消息协议会演进如果哪天需要在payload里加一个新字段version字段能让你在接收方做兼容分发老Worker看到新版本号可以明确拒绝而不是瞎解析新Worker看到老版本号也能继续按老逻辑处理。这个设计是我踩过一次“协议变更导致全线报错”的坑之后才补上的。2.3 运输层选型为什么是Redis Stream消息的运输层我对比过三个方案最后选了Redis Stream这里说一下理由。Kafka是最先考虑的吞吐量大、持久化强但对我们这种十几个Agent、单日几万任务的规模来说运维成本不成比例。Kafka要维护broker集群、topic分区和消费组引入之后整个部署就重了团队里其他人接手成本也高。RabbitMQ是很多人首选的消息中间件路由能力和稳定性都不错。问题在于消费确认和重试的语义偏基础需要自己在业务层做大量封装。我要的是“消息进了队列就尽量别丢Agent挂了自动跳过处理完写回结果”这套完整语义RabbitMQ得拼好几个插件和配置才能达到不省心。Redis Stream反而是最贴合我们需求的操作简单一个队列就是一个Stream key消费者组天然支持消息有唯一ID消费组能记录每个消费者的进度。更重要的是Redis作为基础设施团队早已在运维Agent-Reach不需要再引入一个新的中间件。用Redis Stream还有一个隐性优势调试方便直接redis-cli看一眼队列里的pending消息就能定位不需要打开Kafka的管理界面。当然Redis Stream的持久化不如Kafka消息积压太多会挤占内存。我的处理方式是严格限制下游消费速度每个Agent消费完必须尽快ACK配合消息体里只放必要数据、不放文件内容实测跑下来没什么压力。2.4 注册与能力路由Agent如何被“看见”Agent-Reach里的一切触达前提是Agent先被注册中心“看见”。每个Worker启动时会向Registry发送一条REGISTER消息内容是我是谁、我能干什么、我当前的负载状态、我监听哪个队列。# agent_reach/registry.py async def register(agent_name: str, skills: list[str], queue_name: str): async with redis.pipeline() as pipe: pipe.hset(fagent:{agent_name}, mapping{ name: agent_name, skills: json.dumps(skills), queue: queue_name, status: idle, ts: datetime.now(timezone.utc).isoformat(), }) pipe.hset(agent:index, queue_name, agent_name) # 给skills建立倒排索引 for skill in skills: pipe.sadd(fskill:{skill}, agent_name) await pipe.execute()路由规则我用了最简单可靠的两级匹配先按技能词精确匹配匹配不到再由调度器调用模型做一次语义匹配。比如订单Agent注册时声明了skills为[订单查询, 订单详情, 物流信息]调度器拿到“查询购买记录”这个需求先看技能词命中“订单查询”直接路由如果命不中调度器再用LLM把任务描述和所有Agent的skills列表做一次相似度排序选得分最高的。精确匹配快、便宜、稳定语义匹配慢、贵、但能兜底。两者结合性价比最高。3. 任务编排与交接从拆活到干活的完整链路3.1 把大任务变成DAG按拓扑序推进多Agent协作不能靠“一个Agent干完传给下一个”这种线性思维真实业务里子任务之间有并行关系也有依赖关系。拿售后工单举例查询订单和查询政策这两个子任务没有依赖可以并行但生成回复必须等这两个结果都回来之后才能启动。这本质是一个DAG有向无环图。Agent-Reach里我定义了一个简单的DAG描述格式调度器只负责理解这张图并推进节点。节点有两种状态机pending未开始、running执行中、done完成、failed失败。每个节点有dependencies字段代表前置依赖集合。调度器维护一个待执行队列当某个节点依赖的全部节点都进入done状态它才被触发。# agent_reach/dag.py workflow { def: { nodes: [ {id: query_order, role: order_agent, dependencies: []}, {id: query_policy, role: policy_agent, dependencies: []}, {id: gen_reply, role: reply_agent, dependencies: [query_order, query_policy]}, {id: quality_check, role: review_agent, dependencies: [gen_reply]}, ] } }这个设计的核心价值在于调度器根本不需要知道业务逻辑它只需要做一个机械的“依赖检查 分发 回收结果”。每个节点执行完调度器收到TASK_RESULT消息把节点状态改为done再把依赖解除的后续节点投递出去。业务逻辑全部沉淀在DAG定义里调度器保持纯工程组件这让后续新增业务场景时只加DAG不用改调度器。3.2 交接协商与轮次上限DAG模式解决的是“流程确定”的任务。但真实业务里经常出现信息不全、需要Agent主动确认的情况。比如话术Agent发现政策Agent返回的结果里没有写明“发票丢失能否换货”它该怎么做直接猜一个政策结论风险太大硬着头皮继续写基本就是编。Agent-Reach给每个Agent开放了一个受限的协商通道。Agent收到任务后如果发现自己掌握的信息不足可以通过NEGOTIATE消息发起澄清请求我需要关于字段X的补充信息请政策Agent重新回答。这个请求会转发给调度器调度器根据目标语义路由将追问挂到原任务上重新派发给对应的Agent。但协商必须有上限。我的做法是每一轮NEGOTIATE消息携带一个轮次计数同一个task_id下最多允许两轮澄清超过后自动升级调度器把两边Agent的中间结果打包直接抛给兜底策略——要么走人工审核流程要么基于已有信息做保守处理。我在第5章会展开讲那个没有轮次上限时发生的循环调用风暴这里先不剧透但记住这句话任何允许Agent主动发起交互的机制都必须给它一个明确的终止条件否则它会在你看不见的地方空转耗钱。3.3 幂等与重试触达一次就做成功一次多Agent系统里最隐蔽的坑就是同一个任务被重复执行。触发原因五花八门Worker处理超时调度器以为它挂了重新下发消息总线回执丢失发送方重试消费端代码逻辑里出现重试机制。如果你的任务只是“生成一段文本”重复执行最多浪费点token但如果Agent执行的是“调用退款接口”这种有副作用的动作重复执行就是事故。Agent-Reach的幂等方案很朴素但有效每个任务进入执行队列之前调度器生成全局唯一的task_idWorker执行前先查幂等表key就是task_id查到了直接ACK跳过没查到才执行执行完写入幂等表再返回结果。# agent_reach/worker.py def handle_message(msg: dict): task_id msg[task_id] if redis.sismember(idempotent_keys, task_id): logger.info(duplicate task skipped: %s, task_id) return None # ACK即可不重复执行 result execute_business_logic(msg[payload]) redis.sadd(idempotent_keys, task_id) return result需要注意的是幂等表不能只放在Worker内存里。多个Worker实例可以水平扩展内存态的幂等表在实例之间不同步起不到作用。Agent-Reach统一把幂等表放在Redis里Worker再多也共享同一份去重状态。代价是每次任务多一次Redis查询但从工程可靠性角度这笔开销是绝对值得的。3.4 上下文记忆该共享的共享该隔离的隔离多Agent协作绕不开一个深层问题每个Agent的上下文从哪来我见过很多团队的方案是“把对话历史和中间结果全部塞给下一个Agent”这个方案在前三个任务里能跑到第五个任务基本就废了——信息越来越多模型输入越来越长最后超出上下文窗口。Agent-Reach把记忆分成两层管理。第一层是会话级短期记忆按session_id隔离存最近几轮对话和最近产生的关键中间结果。第二层是业务级长期记忆存历史工单的处理模板、用户画像、政策要点这类跨会话复用的信息用向量存储按需检索。每个Agent的任务payload里调度器只注入“当前任务需要的最小上下文集”而不是把所有历史和中间结果全量塞进去。这句话是重点最小上下文集而不是全量上下文。你只需要让话术Agent知道订单结论、政策结论和用户原话不需要让它知道这个工单在DAG里走了几个节点、谁在几秒前ACK过。上下文给多了模型会迷失在无关信息里还白白增加token成本。4. 落地实现Agent-Reach的最小可运行版本4.1 技术栈清单与选型理由Agent-Reach这套骨架不绑定特定语言但我的参考实现是Python核心依赖如下。组件选型选型理由服务框架FastAPI异步、轻量Agent的HTTP接口和健康检查都方便消息通道Redis Stream前面已详述运维成本低、语义够用长期记忆Qdrant向量检索性能好支持Docker单机部署LLM调用层LiteLLM统一封装OpenAI、Claude等厂商接口切换模型只改配置DAG推进自研状态机场景简单不需要引入Airflow这类的重型DAG调度器有一点要提前说清楚的不是“哪个技术最好”而是“不要为了引入组件而引入组件”。我见过不少人做类似系统上来就上Kafka加ZooKeeper再配一套Flink结果Agent只有三个任务量五分钟跑完整个集群的运维成本比业务本身还高。Agent-Reach的设计哲学是先用最少组件跑通瓶颈在哪再补哪。Redis队列压不住的时候再去横向扩展Worker而不是一开始就架一堆重型中间件。4.2 调度器核心代码任务分发怎么做调度器的核心是一个循环接收外部请求、找到可运行的DAG节点、把子任务塞进对应Agent的队列、监听结果事件推进DAG。# agent_reach/orchestrator.py import json import uuid class Orchestrator: def __init__(self, redis, workflow_registry): self.redis redis self.workflows workflow_registry def submit(self, flow_name: str, session_id: str, inputs: dict): graph self.workflows.build(flow_name, inputs) trace_id uuid.uuid4().hex ready [n for n in graph.nodes if not n.dependencies] for node in ready: self._dispatch(node, trace_id, session_id, graph) return trace_id def _dispatch(self, node, trace_id, session_id, graph): task { trace_id: trace_id, session_id: session_id, task_id: f{trace_id}:{node.id}, action: node.action, payload: node.inputs, retry: 0, } queue fqueue:{node.assigned_role} self.redis.xadd(queue, json.dumps(task).encode()) graph.mark_running(node.id) async def consume_results(self): # 监听结果队列更新节点状态推进后继节点 while True: _, entries await self.redis.xread( {queue:results: $}, block500, count10 ) for msg in entries: result json.loads(msg[1][1]) trace_id result[trace_id] task_id result[task_id] node_id task_id.split(:, 1)[1] self.workflows.on_result(trace_id, node_id, result[status]) self._schedule_dependents(trace_id, node_id)这段代码刻意保持简练真正能说明问题的是_dispatch里的设计取舍。task_id的格式是“trace_id 冒号 节点ID”这一个约定就同时满足了三个需求按session过滤、按trace串联、按节点定位生命周期。你不需要在消息体里额外维护一套复杂的关联关系光看task_id就能知道这条消息属于哪次请求、哪个DAG节点。调度器本身是无状态的除了Redis里存的DAG状态它不保存任何本地状态。这意味着调度器可以多实例部署任何一个实例挂了换一个新的起来从Redis恢复状态继续跑不会中断流程。无状态是分布式系统的万金油工程习惯代价只是多几次Redis读写的毫秒级开销但对于调度器这种核心组件可靠性优先于那几毫秒。4.3 Worker侧代码Agent怎么消费任务Worker是真正跟模型打交道的部分它做的事情比较简单从自己的队列里取消息、做幂等检查、调用模型、写回结果。但有几个细节我觉得比代码本身更值得注意。Worker消费消息时需要记录自己处理每条消息的耗时和token消耗。这不是为了炫数据而是定位问题的关键服务质量下降往往是某条消息的执行时长突然飙高、token量突然超限。有了按task_id粒度记录的执行指标你能在报表里第一眼看出是哪个环节出了问题。# agent_reach/worker.py async def run(self): stream fqueue:{self.agent_name} group self.agent_name while True: try: res await self.redis.xreadgroup(group, self.agent_name, {stream: }, count1, block5000) if not res: continue raw res[0][1][0][1] msg json.loads(raw) if await self._is_duplicate(msg[task_id]): await self.redis.xack(stream, group, res[0][1][0][0]) continue result await self._invoke_model(msg) await self._publish_result(msg, result) await self.redis.xack(stream, group, res[0][1][0][0]) except Exception as exc: await self._handle_failure(exc)Worker的异常处理有一个关键教训不要无条件重试。刚开始我把所有异常都catch住然后放回队列尾部延迟重试看起来稳妥实际上会把问题放大——如果异常是“API key过期”这类系统性故障所有Worker都会陷入“消费-失败-回队-再消费”的空转循环消息处理速率瞬间变零日志却疯狂刷屏。后来我改成根据异常类型分流网络抖动类延迟5秒重试模型侧限流类延迟30秒重试参数错误、模型不存在这类必错任务直接写入失败队列交给人工或下游兜底处理。4.4 可观测性没有trace_id就别谈排障如果你只从这篇文章里带走一个工程习惯我希望是所有多Agent系统必须在一开始就引入贯穿全链路的trace_id。Agent-Reach的每个消息都带trace_id业务日志统一结构化输出这个字段。排障时的常规动作就是拿到一个异常工单号查日志里对应的trace_id一条链路从上到下所有Agent的执行记录全出来了。哪个节点耗时异常、哪条消息被重试、哪一步结果为空都清清楚楚。我还做了一个轻量的看板按trace_id汇总每个节点的耗时和token数。这个看板帮我们抓到了不少隐蔽问题比如某个Agent平时只要1秒响应某天突然变成8秒查下去发现是它读的长期记忆里混进了大量无关向量检索耗时暴增。没有trace_id级别的链路追踪这种问题只能靠运气发现。5. 踩坑实录三个典型故障的完整排查链路5.1 循环调用风暴同一trace_id下的无限套娃这是Agent-Reach上线后遇到的第一个严重故障。某次压测时我下了30个并发工单然后发现系统响应时间从正常的8秒飙升到120秒日志量在一个小时内增长了40倍。按trace_id过滤日志后我看到同一个trace_id下面产生了调用序列order_agent - policy_agent - order_agent - policy_agent无限重复就像两个人在互相踢皮球。排查链路是这样的先看协商轮次计数发现日志里NEGOTIATE消息的negotiation_round字段永远都是0。再看代码找到了根因——发起协商时把轮次记录写在消息的payload里但接收方解析payload时把它当成普通业务参数重新构造新协商请求时没有继承这个字段导致轮次计数在每次传递中丢失。双方Agent都以为这是第一次协商于是无限循环下去。修复方案是在协议层把negotiation_round提升到Envelope的顶层字段与business payload彻底解耦并在消费端强制校验同一task_id下收到第3轮协商请求直接拒绝上报调度器走兜底。这个教训的核心是任何跨Agent传递的请求它的控制字段轮次、跳数、超时必须放在协议顶层不能混进业务payload里否则总会有某个环节把它当成无效数据丢弃。5.2 会话串扰Agent张冠李戴的根因另一个让人头秃的问题发生在同时跑多个会话的时候。我起了三个并发的对话测试每个会话的用户画像完全不同但Agent B处理会话2时居然引用了会话1里的订单信息。第一反应是向量检索串了吗检查了Qdrant里存的向量session_id前缀没有问题。又开始怀疑是不是模型缓存导致的后来看到Worker的代码才发现问题不在模型而在Worker的事件循环Agent服务是一个异步进程多个会话的消息在同一个Worker实例里被并发消费而业务代码里有一块用模块级字典实现的临时缓存key没有加session_id两个会话的中间结果撞进了同一个槽位。定位过程依赖的依然是trace_id但这次的关键不是链路追踪而是链路追踪带来的“并发边界”视角。过滤同一个Worker实例在相同时间窗内处理的两条消息能明显看到session_id不同但缓存位置相同的现象。修复方案不复杂所有缓存key强制带session_id前缀模块级可变状态全面清除业务数据一律走Redis或者Worker内部无状态化。会话串扰这种bug最恶心的地方在于它不是每次都触发只在并发量足够大、时序刚好撞上的时候出现而且表现就是“张冠李戴”非常容易误导你往模型能力的方向排查。如果多Agent系统里出现“模型回答似乎不听话”的怪异情况先别急着换模型去查你的共享状态是否被并发踩踏了。5.3 上下文溢出导致的静默失败第三个坑涉及的场景是处理超长文档。某次任务是让Agent C从一份20万字的合同里提取关键条款传给后面的回复Agent生成摘要。任务提交后C没有任何报错结果队列里也没有它的消息。调度器等了30秒超时把任务标记为failed。我最初的判断是模型服务有问题但查看调用日志后发现Agent C调用模型时返回了“context_length_exceeded”的错误而错误被Worker的通用异常处理吞掉了——它尝试重试了两次每次都是同一个上下文溢出错误最终把消息丢进了失败队列没有上报调度器。这个问题的根因有两个层面。第一层是模型侧的20万字的文档直接全部塞进prompt神仙模型也顶不住。第二层是我们这边的异常处理逻辑把所有错误都当成“可重试的临时故障”没有对上下文长度这类必错错误做识别和分流。修复分两步。第一步改异常分类模型抛出context_length_exceeded时不再重试而是立刻走“内容压缩通道”——用一个小模型先把原文按章节压缩成摘要再抽取条款结构最后才把压缩后的内容传给Agent C。第二步是给调度器加失败原因透传Agent失败时发送TASK_RESULT消息的statusfailed和reason_code字段看板直接展示。这样至少后续人能一眼看到是上下文溢出还是模型超时而不是对着超时记录瞎猜。上下文管理是Agent应用里最容易被低估的工程问题。很多人觉得模型上下文窗口越来越大了就不需要精细管理了但真实业务里文档量级的增长永远比模型窗口增长快。20万字不行未来200万字的文档呢唯一靠得住的方案还是分层处理先检索、再压缩、最后才给大模型喂相关内容而不是指望模型一口吞下整个宇宙。6. 触达效果的度量与适用边界6.1 有效触达率成功率的另一个视角系统稳定之后我开始思考怎么评估Agent-Reach这套协作框架的好坏。传统指标是任务成功率但它只看最终结果掩盖了过程中的大量细节。我自己更关心的是“有效触达率”定义为目标Agent实际收到任务、正确理解需求、并在规定时间内返回有效结果的比率。为什么强调“有效”因为Agent返回一个空结果、一个格式错误的结果、或者一个“我无法处理”的文案从消息层面看都是成功消费了消息但对业务来说都是无效触达。我在评估日志里加了一个字段来判断结果有效性——由质检Agent对下游Agent的产出做一次轻量校验不通过就标记为invalid。实际统计下来系统整体消息触达成功率是97.4%但扣掉无效结果后有效触达率降到了93.1%。那4个百分点的差距就是系统里真正需要优化的价值空间。6.2 延迟与成本的实测调优Agent-Reach内部压测的数据我整理了一下给读者一个直观参考。测试场景是电商客服类工单5个Agent协作端到端需要经过订单查询、政策核对、回复生成、质检四个DAG节点。Agent数量任务数链路平均延迟有效触达率单任务平均token消耗320012.4s88.5%420052008.2s93.1%510072006.9s95.6%6800从3个Agent增加到5个延迟反而下降了因为原来由一个Agent串行干的活被两个Agent并行执行了。但7个Agent时任务吞吐提高带来的收益开始被Agent间握手开销抵消延迟只降了1.3秒token成本却涨了1700。这说明Agent数量不是越多越好最佳数量取决于任务的并行度而不是盲目追求角色细分。成本调优方面我踩过最深的坑是把所有Agent都配置成同一个大模型。后来改成按角色分级订单查询和政策核对这种结构化任务用小模型处理准确率不降成本降了三分之二只有回复生成和质检这种需要复杂语义理解的环节才动用大模型。这个调整让整体token成本下降了42%而有效触达率只降了0.3个百分点。6.3 什么场景不该用它说完了用途和效果我想认真聊聊Agent-Reach的边界因为不是所有场景都适合上多Agent架构。单轮问答场景不需要它。用户问一句“订单到哪了”直接调一个查单函数返回结果比任何Agent编排都更快更便宜。非要去搞一个多Agent系统反而把简单问题复杂化延迟增加还引入了一堆额外的失败点。固定流程的批量处理也不适合。比如每天凌晨跑一次数据清洗任务逻辑一百年不变这种场景用定时脚本或者工作流引擎就够了Agent的灵活性反而成为负担——它的自由发挥会破坏固定流程的确定性。Agent-Reach真正适合的是任务链路长、涉及多个知识域、中间结果会互相影响、需要灵活处理异常情况的场景。换句话说如果你连自己都不确定任务下一步该做什么需要多个Agent协作探索式的推进那才值得用这套协作框架。回到我这段时间的实践一个特别深刻的体会是多Agent系统很多时候不是被“单Agent能力不够”打败的而是被“多个Agent之间没有可靠的触达机制”拖垮的。我见过无数个Demo展示效果惊艳但都只跑通了一条精心设计的路径换个输入就全线崩溃。Agent-Reach这套骨架让我在应对复杂真实业务时保持从容但它的作用不是让系统变聪明而是让系统变得可控。先把可控性做好再去谈智能。最后分享一个我自己的小习惯每次新增Agent之前我都会先问自己一句——这个Agent如果凭空消失了系统会不会变得更差如果答案是“不会”那说明它本来就不该存在。多Agent不是堆得越多越好而是每个Agent都要有自己的不可替代性。