Agent-Reach在我这边是一套多智能体协作框架的名字起初纯粹为了解决一个很具体的问题我手里同时维护着七八个基于大模型的脚本工具每个都是输入一段文本、调一次模型、吐一个结果的单体逻辑。功能单独跑没问题可一旦任务变成先解析会议纪要再抽取行动项接着检索相关资料补全背景最后生成日报并推送这种多步骤协作单个脚本就顶不住了。要么把所有逻辑硬塞进一个长长的prompt里上下文越长输出越容易跑偏要么自己写一堆胶水代码把脚本串起来但串起来之后的状态管理、失败重试、进度追踪全是坑。Agent-Reach就是奔着这个痛点去的——让不同的AI Agent之间能够以标准方式互相触达Reach彼此的能力把单点工具连成一张可编排的协作网。这个项目适合两类人看一类是已经在用大模型API做自动化但每次多步骤任务都要重复造轮子的开发者另一类是正在选型多Agent框架被LangGraph、AutoGen这类重框架搞得有点晕想要一个能讲清楚原理、能自己掌控细节的轻量方案的实践者。这篇文章不打算讲什么宏大架构就记录我实际设计、编码、踩坑、调优的全过程包括为什么选Redis Streams当通信底座、Reach协议长什么样、48小时跑通第一版时遇到的三个典型故障以及上线前在可观测性和人工兜底上补的那些课。1. 为什么需要Agent-Reach单体任务脚本的瓶颈到底在哪1.1 从一次失败的自动巡检说起触发我做这个项目的直接导火索是一个知识库巡检机器人。需求很简单每天自动把新增的工单内容接入知识库之前先做一轮质量检查——判断标题是否规范、正文是否完整、敏感词是否命中、相似历史工单是否已存在四个检查做完之后统一输出一份报告。第一版我用一个单体脚本实现一个prompt里把所有检查要求全写上让模型一次性返回JSON。前两周运行得还行后面就开始翻车。问题很典型检查项越多、上下文越长模型越容易顾此失彼。经常是敏感词检测正常了标题规范判断开始漏判你为了修正某个漏判往prompt里再补一句规则结果另一个维度又开始误报。我统计过一周四十多份巡检任务的输出整份报告可用的只有二十七份剩余的全是有某个维度明显失真需要人工修正的。这其实是单体Agent的天然瓶颈上下文是共享的所有子能力在一个模型调用里完成彼此没有边界。你希望标题规范这种规则判断足够严格但敏感词命中又需要模型有宽松一点的语义理解能力两种诉求塞进同一个prompt里就是在互相干扰。而且多步骤任务里只要有一个环节出错整条链路就得重跑浪费的token和时间都是倍数级的。1.2 Agent-Reach想解决的问题清单当时我拿了个本子把需求一条条列下来。列完之后发现真正要解决的其实不是哪个模型更强而是这几件基础设施层面的事职责隔离每个Agent只做一件事上下文独立prompt不用互相迁就。标准触达一个Agent完成任务后能按固定格式把结果递给下一个Agent而不是靠写死函数调用来串联。状态可见任何一个任务跑到哪一步当前卡在谁那里失败的原因是什么得能一眼看清。故障隔离某个Agent挂了或者输出不合法要能单独重试不能把整条链路都拖死。可观测每个Agent在一次任务里消耗了多少token、用了多长的上下文、模型返回是否稳定得能统计出来。这五条列完我心里就有数了这不是选个更聪明的模型能解决的事而是一个微型的任务调度与消息传递系统。Agent-Reach的定位随之清晰——一个面向单机可跑、多智能体协作的轻量框架通信层用现成的Redis编排层自己控制模型层只做统一接口抽象不加魔改、不加黑盒。2. 技术选型通信、编排、协议维度我是这样定的2.1 通信底座为什么选了Redis Streams而不是Celery或消息队列项目的第一轮技术争论集中在通信底座上。当时摆在我面前有三个选项Celery、RabbitMQ、Redis Streams还有个后来没采用的方案是直接用HTTP轮询。先排掉Celery。它的调度和执行模型很成熟但对我来说太重了。一是我这个场景没有定时任务和分布式worker的需求大部分时候一台机器就够跑二是Celery的结果后端、任务状态、broker配置一套全下来我得先搭三种存储的配置学习成本和运维成本都上来了。我的核心诉求是轻和看得懂。RabbitMQ的AMQP协议功能很全交换机、绑定、路由规则这套概念学一次能用很久但同样是复杂度问题而且我们不需要复杂的路由语义只需要简单的某个Agent处理某种类型的任务用消息队列的exchangequeue绑定属于大材小用。最后选了Redis Streams理由很实在它的消费者组模型和任务队列天然契合。Streams是Redis 5.0引入的数据结构每条消息自带递增ID天然支持消费者组、ACK、pending列表和消息认领机制。我想要的任务分发、处理确认、失败重投直接用消费者组的XREADGROUP、XACK、XAUTOCLAIM就能实现不用引入任何新组件。项目里本来就跑着RedisAPI统一度也高。更重要的是消息在Streams里是持久化的Redis重启后数据还在这比纯内存队列可靠得多。还有一个点很多人会忽略Streams消息ID是单调递增的本身就是天然的时间轴索引。我在排查问题时可以直接按消息ID范围去查某个时间段内系统到底派发了哪些任务、这些任务的先后顺序是什么这在RabbitMQ里反而不直观因为消息消费完就没了除非你额外做消息轨迹记录。2.2 编排层为什么没选LangGraph这种现成框架那段时间LangGraph正好开始火朋友圈里几篇介绍多Agent状态图的文章转发量都不错我也专门花了两天试用。说实话LangGraph的图编排思想和我想要的Agent触达很接近状态机、条件边、节点回退这些概念都对得上。但试用之后我还是决定自己写一个Agent-Reach的极简编排内核主要是这几个原因第一LangGraph当时的版本迭代非常快API一两个星期就换一次我的Agent逻辑还没写完调用接口就变了追版本追得心累。第二它是为图这个模型设计的节点和边的概念很强但我这个场景本质上是一条比较简单的流水线加少量条件分支为了复用图能力去引入整个框架收益不成正比。第三也是最关键的LangGraph把消息传递、状态持久化、用户交互这些细节都封装在内部出了问题我只能看它源码去猜而我自己做Agent-Reach从消息格式到重试逻辑都是我自己定的出任何问题我都能直接定位到代码行。对多数人来说用现成框架没问题但你要清楚它帮你封装了什么你就要为它帮你封装的东西付调试成本。我的风格是核心链路尽量掌控在自己手里封装越少越不容易黑盒。2.3 模型层统一接口抽象Agent-Reach里的每个Agent本质上都围绕一个大模型调用单元运转所以模型层必须做统一抽象。我在项目里定义了一个ModelAdapter接口核心方法只有一个async def complete( self, system_prompt: str, user_prompt: str, response_format: Optional[dict] None, temperature: float 0.2, ) - ModelResponse: 统一的模型调用入口。 response_format 是 JSON Schema能过则优先让模型按结构化输出返回。 适配器分别对接了OpenAI兼容协议和Anthropic协议实际运行中我发现一个很省事的规律现在国内外的模型服务商大部分都提供了OpenAI兼容接口/v1/chat/completions这一个适配器就能覆盖通义、智谱、DeepSeek这些主流选择。所以我只写了两个适配器一个叫OpenAICompatAdapter一个叫AnthropicAdapter选型时改一行配置就能切换模型供应商。统一接口的回报不只是换模型方便更重要的是Agent-Reach的每个Agent都只认这个接口prompt工程、温度参数、结构化输出格式都能统一管理。我可以在一个配置中心里同时控制所有Agent的温度默认0.2、审核类Agent温度降为0.1这在单体脚本里得每个都改一遍容易漏。3. Reach协议设计Agent之间如何说话才算有效触达如果说Redis Streams是Agent-Reach的血管那协议就是血液。多Agent协作的很多隐性Bug最终都可以追溯到一个问题两个Agent对消息里某个字段的含义理解不一致。所以Reach协议的设计我花的时间最多也最值得分享。3.1 消息信封每个触达请求带上上下文索引我设计了一个统一的消息信封结构不管是请求某个Agent执行任务还是回传执行结果都通过这个结构传递。核心字段如下字段类型说明message_idstring全局唯一消息ID用UUID4生成task_idstring一次业务任务从触发到结束的全局IDfrom_agentstring发送方名称如nlp.postaggerto_agentstring接收方名称如report.assemblerintentstring消息意图取值为request或response或heartbeatpayloadobject实际任务数据或执行结果parent_message_idstring上一跳消息ID用于链路追踪ttl_secondsint消息有效时间超过后接收方可直接丢弃created_atdatetime发送时间UTC格式这里有几个字段是经历过实战折磨之后才加的。第一个是parent_message_id最初没有这个字段排查问题的时候只能靠task_id去拼链路但同一任务里有并行分支光看task_id分不清先后。加上父消息ID之后整个触达链路可以还原成一棵消息树哪一步先触发哪一步清清楚楚。第二个是ttl_seconds这个字段一开始也没有。后来发现一个场景知识库意图识别Agent把问题分类成天气查询然后把请求触达给天气Agent天气Agent因为上游API不稳定负责了这条消息就一直挂在pending里被多次自动重投。加上TTL后当一条消息在队列里等待超过60秒消费者可以直接丢弃或者转人工队列避免死消息反复消费。3.2 任务状态机从created到completed的六种状态消息信封解决的是每一跳怎么通信而编排层还需要一个东西来管理整个任务的当前状态。Agent-Reach的每个任务贯穿以下六个状态created任务已入库尚未派发。dispatched已写入Redis Streams等待某个Agent消费。processing某个Agent已读取任务开始处理。suspended处理过程中依赖外部API超时或流控暂时暂停。completed任务成功走完全部环节。failed不可重试的失败或重试次数耗尽。状态转移规则我全部实现为一个Python状态机模块禁止随意跳转。比如completed状态严禁再被改回processing除非有人工干预接口来显式执行回滚否则任务一旦宣告成功对应的action就不会再被重复触发。这样做的好处是业务侧判断任务结果时不会出现状态矛盾这也是所有自动化系统最容易被忽略的角落——状态图画得挺好看代码里没有任何一个地方强制执行这些规则结果就是线上出现了任务状态是completed但后续Agent还在继续处理的灵异现象。3.3 超时、重试与幂等把模型偶尔抽风变成可管理的事件模型类的Agent最讨厌的地方在于同一个prompt同一份输入这次输出合法JSON下次可能就输出一段废话或者中途断连。我针对这个特征给Agent-Reach设计了分层处理策略第一层是调用超时。每个模型调用都设了读超时默认90秒这个值是我根据最大输入上下文20K token、输出最高8K token的耗时上限再放1.5倍余量得出的。超过就直接把任务置为suspended不占用Agent的并发槽位。第二层是服务端重试。遇到HTTP 429、5xx这些可重试错误采用指数退避加重试抖动。指数退避的基数设为2秒最多重试4次等待时间分别是2、4、8、16秒再加一个0到800ms的随机抖动。这里必须强调抖动的重要性没有抖动时多个Agent同时失败重试退避计时会趋于同步就会在恢复节点形成一波集中请求反而把上游打垮。第三层是业务幂等设计。光有重试不够重试的前提是同一个任务可以被安全地处理两次。我在设计时就规定每个Agent必须实现两个幂等输出路径一是结果按task_id存入Redis Hash写入动作本身是幂等的二是Agent在处理前会检查Redis里是否已有该task_id的处理结果有就直接读取返回不重新调用模型。这样一来消息偶发重复消费造成的副作用就被挡在业务逻辑之外了。3.4 优先级分发三个stream分别跑不同时效要求的任务项目刚跑起来的时候只有一个stream所有任务共享一个消费组。后来日报生成这种大任务和意图识别这种轻任务混在一起后者经常被前者阻塞体验很糟。Redis Streams本身不像RabbitMQ那样支持多级优先级队列我的解决方案很粗暴按时效级别拆成三个stream——high、normal、low每个stream对应一组消费者跑在独立进程中。high即时交互类比如用户在对话框里的意图识别要求P95延迟小于3秒。normal准实时类比如知识库入库前的质量检查能接受30秒级别的延迟。low批量离线类比如日报汇总、周报生成分钟级延迟完全能接受。每个stream的消费者进程数也可以独立配置high级别的进程数多配了两个。一开始三层stream的任务分配比例大概是2:5:3跑了一周观察后发现low级别的队列经常空转于是又把一部分low的消费者进程砍掉让位给high和normal。这个调整只能靠线上数据说话拍脑袋是没有意义的。4. 从零跑通Agent-Reach第一版骨架代码和三个坑4.1 最小可行骨架一个主控Agent加两个工作Agent我用来验证整个框架的最小闭环很简单主控Agent负责拆解用户一句话拆成查资料和写摘要两个子任务分别触达给调研Agent和摘要Agent最后主控汇总两份结果输出。主控Agent的核心逻辑大概是这样一个异步任务async def run_controller(task: dict): # 1. 拆解用户请求 plan await planner_agent.plan(task[query]) # 2. 按计划并行触达下游Agent subtask_ids [] for step in plan[steps]: subtask_id await dispatch( task_idtask[id], to_agentstep[agent], intentrequest, payload{query: step[query]}, ) subtask_ids.append(subtask_id) # 3. 等待子任务回包带超时控制 results await gather_with_timeout( subtask_ids, timeout120, on_timeoutescalate_to_manual, # 超时转人工 ) # 4. 汇总并回写 final await assembler_agent.assemble(results) await mark_task_completed(task[id], payloadfinal)dispatch函数的核心就三步生成UUID、写Redis Hashtask上下文、向目标Agent对应的stream XADD消息。表面上看很简单但这三步在初期踩的坑最多下面详细说。4.2 踩坑记录三种典型的Agent协作失败模式第一个坑是模型的JSON输出不稳定。调研Agent返回的内容应该是一个规范的JSON对象但模型偶尔会输出带Markdown代码块包装的JSON或者多个JSON拼在一个响应里。我一开始直接在解析处抛异常然后整条任务失败重试。后来加了一道JSON修复层先用正则抽取代码块内容再找JSON首尾大括号最后发现实在解析不了才报错。结果修复率大概提升了三成但最根本的解法还是切换到结构化输出模式在调用适配器时传response_format{type: json_object}同时把Agent的指令里写明必须输出JSON对象不得包含任何其他文字。两者叠加之后JSON解析失败率从本来的8%左右降到了不到1%。第二个坑是消费者组消息重复消费。现象是同一个任务偶尔会被处理两次下游发现重复写入数据。排查过程是这样的先看日志里的message_id确认两次处理的是同一条消息再看XACK调用时机发现我在Agent第一次写入结果后、还没完成后续处理时就XACK了结果Redis的pending列表里这条消息还在XAUTOCLAIM把超时未确认的消息重新投递给了另一个消费者。修复方案有两处一是把XACK改到整个处理流程的成功点确保只有完全处理完成才确认二是在Agent里加了幂等查重对应前面说的幂等设计。这个坑让我学到一个教训消费者组的语义是未确认即未完成重投是设计内的正常行为不能当意外来处理代码必须以可重复执行为前提。第三个坑是跨Agent的字段命名冲突。不同Agent各自往Redis Hash里存结果时字段名各写各的比如summary这个键调研Agent存的是调研摘要摘要Agent存的又覆盖成了正文摘要下游拿到的结果张冠李戴。我把统一存储规范收紧成每个Agent的结果必须存到task:{task_id}:result:{agent_name}这个Key下字段名统一由协议层预定义不允许Agent自定义顶层字段。改完之后再没出现过数据互相覆盖的事。4.3 从单体到Agent化的改造路径先不动业务只加网关Agent-Reach第一版上线时没有把所有脚本一锅端而是先做了一个协议适配网关给现有的每个单体脚本包一层Agent外壳脚本本身还是原来的函数外壳负责把入参包装成协议定义的payload把返回值包装成协议定义的response然后接入Redis Streams。这样一个星期内就把原来五个孤立脚本全部纳管到了Agent-Reach体系内业务代码几乎没动只是把调用方式从直接函数调用改成了发消息、等回包。路径是新的逻辑是旧的但生态建立起来了。这也是我想给所有做AI工程化的朋友的建议不要等框架完美了再迁移先找一个最低成本的切入口把系统跑起来再逐渐替换内部实现。Agent化改造的收益在早期不在模型更聪明而在每个环节的输入输出都标准化了、链路可追踪了、失败可重试了这三样对稳定性提升的效果立竿见影。5. 触达效率的量化用什么指标判断这套系统值不值Agent-Reach跑了一个多月后我开始认真收集数据不带数据的优化都是自嗨。我定义了三类核心指标也把实测数值贴出来作为参考基线。5.1 三类指标任务成功率、单跳延迟、端到端延迟第一类是任务触达成功率。定义是一个task从created到completed期间所有跨Agent触达消息都成功被目标Agent消费并返回response的比例。我统计的是消息层面的最终一致率即包括重试在内最终成功和最初失败的比值。第二类是单跳延迟。指一条消息从XADD进Stream到目标Agent完成处理并XACK的时间不包含业务处理本身可能触发的对上游模型的长时间等待。第三类是端到端P95延迟。从用户提交任务到最终结果可读取的完整耗时。我选了一批正常时段的典型数据统计窗口是一周的存量任务指标平均值P95备注任务触达成功率97.8%—失败主要为上游模型超时单跳延迟42ms96ms不含Agent内部模型调用耗时端到端延迟5.3s8.1s典型任务包含2次模型调用模型调用总耗时占比85%—通信层开销仅占15%不到这组数据告诉我的核心结论是Agent-Reach的通信层本身几乎没有瓶颈单跳延迟几十毫秒真正的耗时全在模型调用上。所以在做性能优化时我完全不需要去纠缠消息队列里积压了多少消息这种问题而是把火力集中在两件事上——减少模型调用次数比如让两个Agent合并成一个Agent去执行某两个紧耦合步骤以及降低单次模型调用的延迟换成更快的模型或缩短输入上下文。5.2 优化案例把两跳变一跳后延迟从9秒降到4秒我的日常巡检任务原本由两个Agent串行处理质检Agent先做检查然后触达格式Agent做排版两跳之间多了两次完整的Redis读写和一次模型调用。后来我把这个场景合并成了一个Agent内部第一步调用模型做检查第二步直接用规则引擎处理排版而不是再次调用模型。端到端P95从9.2秒降到4.1秒因为省去了一次模型调用。这个优化的成本极低但从效果上看比我在通信层做任何微调都值钱。这也是多Agent系统的一个反直觉点不是Agent拆得越细越好每一次触达都有成本只有当拆开后每个Agent的prompt显著变短、职责显著清晰、输出质量提升能覆盖通信和调用开销时拆分才是划算的。判断标准必须拿数据说话而不是凭感觉。6. 上线前必须补的课可观测性、人工兜底和迭代节奏Agent-Reach真正让我头疼的不是写Agent而是一个系统一旦跑起来你需要对它的看不见的行为负责。Agent不像普通函数你不能靠断点追踪每个环节所以上线前的最后一公里全在可观测性和人工兜底上。6.1 日志、追踪与指标三件套我在计数器、日志和链路追踪上分别做了对应实践。日志方面所有Agent统一采用结构化JSON日志每条日志必须带task_id、message_id、agent_name三个字段。这样做之后用grep task_idxxx可以瞬间捞出一条任务触达全链路上的所有日志不管是主控Agent派发的、调研Agent处理的还是最终汇总的全部串在一起。没用结构化日志之前排查一次失败任务要在终端翻半天还经常因为时间戳对不上看岔。链路追踪方面我实现了一个非常轻量的tracer核心是基于消息信封里的parent_message_id构建一棵消息树。在Web管理界面里勾选一个任务可以按时间展开每一个Agent收到的消息、发出的消息、调用模型的token消耗和耗时。做这个界面用了不到一百行Python代码加一个简单前端但价值极大——多Agent系统出故障时你需要的是这个故事是怎么发生的而不是最终结果是什么。指标方面用Redis自带的INFO和LATENCY命令起步重点关注三个数字Streams的length是多少代表积压量、消费者的pending条数是多少代表未确认或卡住的消息、Redis的瞬时CPU占用。这些数字变化趋势比绝对值重要length持续上涨说明消费者处理不过来pending持续上涨说明有Agent卡死没有XACK。6.2 优雅降级与人工接管让自动化系统有刹车再稳定的模型也有失控的时候我的经验是必须在设计阶段就预留人工接管通道。Agent-Reach里有一条规则凡是任务连续重试超过4次仍失败或者单条消息在pending队列里的存活时间超过5分钟系统自动把任务状态从failed或suspended转为manual_review同时向某个webhook推送一条待处理任务通知。运营同学可以在管理界面看到这个任务的完整消息轨迹可以选择重试、直接编辑结果、或者彻底废弃。这条通道看起来简单但真正的功夫在细节。比如自动转人工的通知必须包含一个一键重试按钮而且重试时要允许人工先修改payload里的关键参数而不是原样重放一次注定失败的任务。再比如人工处理完任务后状态机要允许从manual_review直接跳转到completed这条路不在常规状态机里属于显式开的一个口子。6.3 迭代节奏小步快跑每周只改一个行为最后聊一点管理层面的经验。Agent-Reach上线后我给自己定了一条规矩每周只改一个Agent的行为换模型风格、调prompt、改超时参数算一个行为然后跑一整周统计效果再决定下一步。多Agent系统最怕的就是同时调整多个Agent一旦整体效果下降你根本说不清是哪次调整导致的。单变量迭代的节奏看起来慢实际上因为每次改动都能获得干净的数据反馈整体进展速度反而快得多。这也算是我在做这套系统过程中最大的一个方法论收获。如果你也想搭一套类似的多Agent协作系统我的建议是先把职责边界想清楚再谈框架选型先把消息协议和状态机定死再写业务逻辑先跑通一个最小闭环再逐步纳管旧的单体脚本。把这三件事做完你会发现Agent-Reach真正的价值不在某一两个模型多聪明而在于整个触达链路变得可预测、可追踪、可重试、可接管——这些恰恰是自动化系统在真实业务里立足的底线。