上个月我把一个压了很久的需求彻底重做了让一个AI Agent独立完成“监控指标→生成日报→推送到企业群”整条流水线。第一版我图省事把查数据库、算聚合、做图表、组织文案四件事全塞进同一个Agent的System Prompt里结果最直观的表现是——任务越长越容易在某一个中间步骤上崩盘。后来改成多Agent分工新问题立刻冒出来这几个Agent彼此根本不认识更不知道谁擅长什么让A去找BB反而把请求推回给A。我折腾了一周最终沉淀下来的就是一套叫Agent-Reach的触达层。Agent-Reach这个名字里的“Reach”说的从来不是Agent有多聪明而是它能不能在整个工作流里“够得着”正确的协作对象和正确的上下文。它的核心价值可以拆成三件事把Agent的能力变成可见、可查、可校验的列表让请求按照语义和约束路由到正确的接收者在Agent之间做显式的上下文对齐而不是靠模型临场发挥。这篇文章我打算把整套思路和代码级实现完整写出来包括我是怎么设计注册中心、消息总线和能力描述协议的也把跑起来之后踩过的四个深坑原原本本交代一遍。无论你是正在纠结“单Agent还是多Agent”还是已经拆了多个Agent但发现它们根本没法稳定协作这篇都应该能给你一个可以直接落地的参考框架。1. 为什么说“Agent-Reach”解决的是连接而不是智商1.1 单体Agent跑全流程本质上是在赌上下文不失控先说说我为什么要把单体Agent拆掉。用单Agent执行完整业务链路时最容易踩的坑就是System Prompt越来越长。你把查询引擎怎么用、分析规则是什么、报告模板长什么样全部塞进去随着任务推进上下文里塞满了各阶段的中间结果。每次工具调用返回的数据都完整拼在对话尾部上下文总量就不再由信息价值决定而是由中间步骤的数量决定。我实际项目里有一个指标单Agent处理一份包含12个SKU的日报生成任务到了第八个SKU数字开始张冠李戴把A仓的销售额填到了B仓的报告里。当时我根本分不清是提示词问题还是数据问题因为整个上下文已经膨胀到快5万token模型注意力的分配早就是一团浆糊了。这里要澄清一下我并不是否定单体Agent的价值。对于“总结会议记录”这类上下文封闭的任务单Agent依然是最好的选择因为它的输入输出边界非常清晰。但跨环节业务是状态开放的你需要把状态经过多个工具、多个职责节点传递这时候把所有状态都堆在一个上下文里本身就是瓶颈。拆分责任是工程选择而不是模型能力不够的妥协。1.2 协作的瓶颈发现、路由、上下文对齐多Agent化之后第一个真实冲突点就是“我不知道该找谁”。我在代码里创建了一堆Agent实例每个都很能干但它们在运行时完全不知道彼此的存在。你可以在Coordinator的提示词里写“你有以下子Agent可用”但这就把协作完全押在了模型的理解力上。后来我把“触达”这件事拆成三个层面能力发现每个Agent提供一份结构化的“自我介绍”包括name、description、input_schema、output_schema注册到一个统一列表里。这个过程和搜索引擎爬虫提交站点地图没什么区别核心是让能力可以被检索。路由调度收到一个请求后决策模块需要判断这个请求该交给哪个Agent处理。这里既可以用规则匹配也可以用LLM做语义判断但关键在于必须有一个确定性的兜底不能纯靠模型猜。上下文对齐不同Agent对字段名、时间格式、单位、分页规则的预期往往不一样。比如一个Agent输出的date是“2024-08-01”另一个Agent接收时要求的是时间戳。对齐这件事必须显式做不能指望Agent自己猜出来。打个比方这套东西跟企业总机的分机表加前台转接没什么区别能力发现就是电话簿路由调度就是前台上下文对齐就是接线翻译。Agent-Reach做的不是训练Agent本身而是在协作体系里把这张电话网搭好让每个分机都够得着它该够得着的人。1.3 为什么要做成显式协议而不是靠提示词约定有人可能会问我直接在Coordinator的提示词里写“你有以下Agent可用”不也能实现触达吗表面看能但这里有个致命问题提示词约定是不可观测的。Agent什么时候选错了路由、选了之后有没有调用成功、调用失败之后有没有按预期降级这些过程全都藏在黑盒里出了问题只能靠猜。显式协议的价值在于每个协作动作都对应一条可查的日志谁提交了注册、路由是按什么规则匹配的、消息在哪个入站队列等了多久、上下文在哪一步做了字段转换。故障定位从“猜模型在想什么”变成了“查记录”。这也是我把Agent-Reach定位成基础设施而不是业务代码的原因它不需要任何Agent变聪明只需要让每个协作动作变成可观察、可追踪、可控制的操作。2. 一版就跑通的MVP注册中心、消息总线与能力描述协议技术选型上我用了Python加asyncio没有引入任何重量级框架。选它的原因很朴素asyncio的队列天然适合做Agent间的消息传递每个Agent的输入输出就是JSON消息调试起来特别直观。什么分布式消息中间件、服务网格在MVP阶段都不需要。2.1 能力描述文件让每个Agent先写“自我介绍”在Agent-Reach里注册的是能力不是Agent实例本身。每个Agent在启动时必须先向注册中心提交一份能力描述我用的格式大概长这样name: report_agent description: 根据指标数据生成日报/周报支持Markdown和HTML输出 version: 0.3.1 owner:>import asyncio import time from dataclasses import dataclass, field from typing import Optional, List dataclass class AgentInfo: name: str description: str input_schema: dict output_schema: dict tags: List[str] timeout_hint: int 60 last_heartbeat: float field(default_factorytime.time) healthy: bool True class Registry: def __init__(self): self._agents {} async def register(self, info: AgentInfo): self._agents[info.name] info async def unregister(self, name: str): self._agents.pop(name, None) async def heartbeats(self, name: str): if name in self._agents: self._agents[name].last_heartbeat time.time() self._agents[name].healthy True async def find_by_tag(self, tag: str) - Optional[str]: for info in self._agents.values(): if tag in info.tags: return info.name return None async def find_by_description(self, text: str) - Optional[str]: score_max, name_selected 0, None for info in self._agents.values(): hit sum(word in info.description for word in text.split()) if hit score_max: score_max, name_selected hit, info.name return name_selected if score_max 0 else None async def healthcheck(self, max_stale: int 30): now time.time() for name, info in self._agents.items(): info.healthy (now - info.last_heartbeat) max_stalefind_by_description的实现方式是把请求文本和每个Agent的description按词匹配计算命中数取最高分。有人看到这种简单做法可能会觉得太原始但我要强调它不依赖LLM结果完全可复现速度快到可以忽略不计。我在实际配置里让规则路由先跑如果规则找不到结果再让Coordinator用LLM做一次模糊匹配并且对LLM选出的结果做schema校验。规则兜底永远比直接推理更可靠这是多Agent架构里最值得坚持的原则。2.3 消息总线其实只需要一个能超时和重试的缓冲为了不让Agent之间直接互相调用搞出网状耦合我在中间加了一条简单的消息总线。每一条消息统一包成envelope格式如下{ msg_id: 2f9a4f00-1234-abcd-1234, from: coordinator, to: report_agent, trace_id: a1b2c3, type: request, payload: { ... }, credentials: { scope: [read:metrics, write:report] } }失败时type变为error在payload里带code和message。总线入口逻辑很单一接收消息、检查目标Agent是否注册且健康、放入对应队列、等待Agent完成或超时、写日志。超时判断不复杂用asyncio.wait_for包在响应队列的get()上超时后自动重试一次。但重试必须小心一个陷阱不是所有操作都幂等。比如report_agent生成报告第一次可能已经生成了半份结果你无脑重试对方就要面对两份半成品。我的做法是在payload里带msg_id目标Agent会先检查msg_id是否处理过处理过就直接返回上一次结果这比用各种分布式事务方案简单得多。3. 打通一条完整链路数据查询Agent与日报生成Agent的真实协作理论说得再多不如直接看一个真实链路。我拿业务方的一个典型需求举例每天想看到“销售额低于阈值10%的分仓SKU”然后基于这些数据生成一份日报并推送出去。3.1 场景拆解与角色分配以前在单体Agent里处理这个需求一个Prompt从头写到尾中间横跨查询、计算、报告生成、发送四个环节。现在我拆成两个Agentdata_query_agent只负责查询数仓返回固定字段的原始指标不做判断不组织文案。report_agent接收指标和日期字段负责写报告、生成图表、组装最终输出。拆分的理由很明确数据查询是强约束操作最好用规则或API完成结果必须可复验而报告生成是一次模型调用允许失败重试。两者混在一起模型在中间步骤稍微失一下真整个数据链路的可信度就完蛋了。3.2 路由链路与上下文转换用户从统一入口说了一句“生成昨天的日报”。Coordinator的处理流程是先用tags里的daily精确找到report_agent。report_agent发现自己需要原始数据但它不知道数据API在哪儿于是通过注册中心找到data_query_agent。report_agent发出一条request并在payload里带上明确的上下文约束timezone8、interval[昨天开始, 昨天结束)、granularitydaily、min_sales_threshold0.1。上下文转换的核心动作发生在这条request里把自然语言里的“昨天”解析成精确的start_date/end_date把“低于阈值10%”翻译成API能识别的min_sales_threshold0.1把返回值统一成双方约定的字段名sku, warehouse, sales_amount, threshold_ratio。转换必须是显式步骤不能依赖Agent自己发挥。我在实际操作中发现如果这里不显式传时区一个在北京早上九点跑的任务和一个在UTC时间跑的任务算出来的“昨天”可能差出整整一天。3.3 超时、降级与部分失败的处理数据查询Agent返回之后report_agent开始写报告写完要推送到企业群于是它再请求notifier_agent。这一步看起来很顺但真正考验系统的是失败场景。我在第2版里设置过查询Agent返回超时并重试仍失败时report_agent不再无限等下去而是直接生成一份降级报告明确标注“XX时段的销售数据缺失”流程继续向前走。这个设计在最初遭到了质疑有人说数据都不全了还生成什么报告我的看法是多Agent编排里的每一个环节都是可以独立降级的你追求的不应该是“所有环节都成功”而是“链路走到最后一定会有一个结论哪怕这个结论是部分失败”。当业务方半夜三点等着看报告时一份带缺失标注的报告比一个卡死到天亮的任务有价值得多。另外一个经验是编排链路高度尽量控制在三层以内——协调层、能力Agent层、数据服务层。层数一旦到四层以上trace信息形态就很容易失控每次排查问题都像在一团乱麻里找线头。4. 踩坑实录我在这套架构里翻过的四个大坑写这块之前我先声明Agent-Reach这版设计不是一次就稳的。我前后跑了两个月翻了四个让我印象非常深的坑每一个都值得单独拿出来说。4.1 死循环调用A做不了就推给BB又推回A第一个版本的System Prompt里我给每个Agent都加了一句“如果你不确定可以请求其他Agent帮助”。这句看似无害的话实际运行起来差点把系统拖垮。某天日志里出现了一条消息在两个Agent之间来回传播trace_id完全一样from和to交替出现直到我手动把它终止。跑了一会儿定位到根因report_agent要做一个翻译功能它不会于是调用assistant_agentassistant_agent也不知道数据准确抓取又推回给report_agent。两个Agent都没有处理能力但提示词给了它们“可以求助”的权限于是形成了死循环。修改方案是两个约束一起上第一注册中心路由时禁止“环形转包”即一个Agent只能响应tags和能力描述匹配的请求不能把属于自己职责的请求再推给其他同类Agent第二消息里强制带max_hop3超过跳数直接返回错误。从那以后我把系统里所有Prompt中的“尽量”“你可以尝试”这类模糊措辞全部清理了一遍——在多Agent架构里模糊的授权就是故障的温床。4.2 假死比崩溃更麻烦队列积压与重试风暴第二个大坑是假死。某天链路突然停下来所有消息都卡在队列里不报错、不返回、也没崩溃。一开始很难判断是LLM API在“思考”还是真的卡住了直到我查了API调用日志发现一个Agent在调用LLM时超时了抛了一个普通异常框架二话不说按“失败重试”处理。结果多个Agent同时交互都没有退避机制重试请求被放大成了三倍流量每个Agent都在反复撞同一个失败点。解决办法分两层第一层是给每个Agent的重试都加上指数退避和随机抖动比如第一次等3秒、第二次等9秒、第三次等27秒第二层是全局限制重试次数不能每个节点各自无限重试。假死的另一个表现是Agent内部在跑一个很长的工具调用队列迟迟等不到响应。后来我在Agent里增加了心跳机制它每隔30秒汇报一次当前阶段状态如果心跳停了才判定真死这个设计类似排队时服务员喊一句“还在办理请稍等”至少让你知道系统不是无响应。4.3 上下文在链路里无限膨胀这是让我最头疼的坑。第一版的时候每条消息都保存完整payload数据查询Agent返回了几千行结果report_agent拿到结果后把整份表格又写进了自己的chain日志然后再原封不动地传给notifier_agent。最后notifier的上下文里躺着三份重复的原始数据表模型已经完全分不清哪一份是最新的了。解决方法是做截断型规则核心就一句话每个Agent收到中间结果后只保留“结论摘要加结果引用”原始数据统一存在对象存储里消息里只放一个指向原文的链接。换句话说链路中的上下文像一个接力棒传出去的同时处理串就要缩短。我还在消息封包里加了token预算字段超过预算时原始payload替换为摘要再做一次语义压缩。举个具体例子原来传给notifier的payload是5000行原始销售数据现在变成{ summary: 销售额低于阈值10%的分仓共3个主要集中华南区最大缺口为华南-01仓, data_ref: s3://report-data/20240801/summary.json }这一改整条链路的token开销直接下降了70%以上而且每个环节要处理的噪音少了很多上下文灌入不正确数据的概率也降下来了。4.4 权限边界被忽略任何Agent都能调用所有能力Agent-Reach有一个隐含假设“只要Agent能发现就能调用”。这个假设在内网单租户环境跑着没毛病一旦放到多租户环境或者涉及用户数据风险立刻冒头。我遇到的实际案例是报表Agent用read:report的scope请求了write:user的能力结果真的调用成功了因为注册中心只校验了Agent名称没有校验scope。后来我改成每个能力描述里声明所需scope注册中心在路由阶段同时检查调用方的scope集合是否覆盖目标能力的scope要求不满足直接返回403。这里有个细节权限校验一定不要放在Agent内部代码里。Agent的代码逻辑可以被提示词影响你没法保证模型在运行过程中不会自主“绕过”一段看似多余的校验。放在注册中心和消息总线入口的强制过滤才是最可靠的因为那一段是纯规则判断不涉及推理。5. 什么时候真正适合引入Agent-Reach写到最后我必须说一句常被忽略的话并不是所有场景都该上多Agent架构Agent-Reach也不例外。如果你根本不需要多个Agent分工这个框架只会白白增加复杂度。5.1 用这把尺子决定任务能否被切分中间状态能否被显式传递如果出现下面两种情况之一你多半不需要Agent-Reach任务高度连贯、无法拆分。比如一段需要全局上下文才能做好的创意写作拆给多个Agent反而会丢失整体风格。中间状态很难用JSON表达。比如一个依赖实时交互和视觉反馈的复杂任务硬拆开会让进度和状态都无法传递。真正适合的场景是任务流相对固定中间状态可以结构化且每个环节允许独立重试和降级。拿数据分析加报告推送来说数据查询的结果是标准JSON报告生成的输出是字符串推送动作就一个回调——这种结构天然适合多Agent协作。5.2 三档接入方案不是所有团队第一步就要上完整的注册中心加LLM路由我建议按下面三档逐步演进方案接入改动适合团队路由确定性排查成本人肉路由只做分工脚本请求人工转发刚接触多Agent想验证分工收益100%人判断低因为步骤少规则路由MVP注册中心tags匹配规则兜底分工已明确希望减少人工介入中等依赖描述质量中日志可查LLM路由权限控制完整注册中心LLM模糊匹配scope校验请求语义多样需要灵活分发高但引入不确定性高需要品牌级追踪我个人的建议是先从人肉路由开始大概跑两周看清楚每个Agent的真实分工边界再升级到规则路由。直接跳到LLM路由是非常容易踩坑的你可能连路由错了都不知道。5.3 效果度量与回滚方法引入框架后我主要盯这几个指标端到端任务成功率。平均链路跳数与p95时延。失败定位时间也就是从故障发生到定位到具体环节需要多久。上下文字均值对比拆分前降了多少。重试率与降级率。这几项指标里失败定位时间是最能体现Agent-Reach价值的指标。以前单Agent出了问题我要靠反复调提示词去猜现在多Agent链路上出了问题我打开trace日志就能看到是哪一环返回了错误码。排查时间从小时级降到了分钟级这本身就是拆分最大的回报。另外架构上要保留回滚能力。一旦发现拆分后的端到端成功率反而不如原来的单体Agent我会一键把路由规则改成直连单个Agent。Agent-Reach存在的意义是让系统具备“按需协作”的能力而不是强迫你永远处在一个复杂的协作模式里。我现在的习惯是先在固定结构任务里用Agent-Reach比如周报、数据校验、定时巡检这类边界清晰、状态可传递的场景暂时不让它碰开放性特别强、错误代价特别高的动作。连接能力落地之后如果模型业务确实处理得不好至少我们能在几秒钟之内定位到问题出在哪一环而不是在漆黑一片里到处乱猜。这种“可定位”的价值说真的比多Agent本身还要值钱。