Agent-Reach 这个名字最初其实是我本地项目里随手起的代号。当时我在折腾一套多智能体协作系统遇到的问题让我印象特别深单个Agent做得再好一旦把多个Agent拼在一起协作就变得很脆弱。有的Agent拿到了工具却不会传参数有的Agent在等待别的Agent返回结果时把上下文窗口直接塞满了最让人崩溃的是两个Agent为了一个小问题互相踢皮球来回调用了几十轮。我把这些现象复盘了一遍发现核心矛盾其实是一个词触达。Agent没有可靠的触达方式——触达该用的工具、该拿的数据、该协作的下一个Agent。所以我抽了一个独立的编排层取名Agent-Reach。这篇文章把Agent-Reach的设计思路、最小实现和排障经验写出来给正在做多Agent编排、工具接入或者企业内部AI助手的同学参考。1. Agent-Reach是什么为什么我在多智能体协作里卡了壳1.1 单Agent好做多Agent难在“触达”现在随便一个开源框架都能让单个Agent玩得很好接上大模型、挂上几个工具它就能完成不少任务。单Agent的思路其实非常简单一个大模型当大脑外面挂一圈工具模型决定调哪个工具、传什么参数、什么时候收手。这种模式几乎没有协作成本所以开发体验很好。但生产环境里真正难的是多Agent协作。我打个比方单个Agent就像一家全能小店店员知道自己货架上有啥、仓库在哪。可当多个Agent放在一起就好像一个商场里有几十个专柜顾客要买东西时需要的其实是导购——告诉他去几楼、找哪个柜台。多Agent系统的问题也一样任务不知道交给谁、工具不知道去哪调、信息不知道和谁同步。这些问题的共同点都指向同一个词“触达能力不够”。我在一个实际项目里手上捏着N个Agent分别负责文档解析、信息检索、代码补全、报告生成。单测每个都不错可一旦把它们组合起来跑一个完整业务流程就会频繁出现“任务派不下去”“派下去没人接”“接了之后不回复”的情况。我一度以为是模型能力不行后来才发现缺的是一个能让Agent之间互相“找到对方”的中间层。于是Agent-Reach的第一版就走上了这条路不重新发明Agent而是把Agent之间、Agent与工具之间的触达关系管起来。1.2 我理解的“Reach”有三层含义第一层是触达工具。Agent要用工具但没必要让每个Agent看到全部工具。代码生成Agent不该看到日志清理工具文档处理Agent也不该拿到运维治理接口。Agent-Reach让工具统一收口到一个独立层里按Agent的能力目标动态注入避免“工具一多就乱套”。第二层是触达伙伴。一个任务往往需要多个Agent接力完成。Agent-Reach维护着一张“Agent能力地图”路由层根据任务要求找到最合适的Agent并且知道这个Agent当前忙不忙、有没有能力承接。这样一来Agent之间不再需要互相猜身份而是直接通过调度层建立协作关系整个系统的可预测性会好很多。第三层是触达目标。任务从进来、拆解、分配到最终结果返回整个过程必须有清晰的状态流转。一个任务交给Agent-AA回复说做完了但结果还没被合并进原对话这就是目标丢失。Agent-Reach对这种状态做了显式记录每一层都知道任务正在谁手里、还差什么、下一步要找谁。这三层合在一起就是我对Agent-Reach的定位一个让多Agent系统能“找得到、调得到、收得回”的编排层。如果你觉得这个定义有点抽象没关系后面几段我会把每个部分拆开讲再附上一套可运行的最小实现。1.3 关键的设计取舍第一个取舍是中心化编排还是完全自组织。我最终选了中心化因为它可控、可观测。完全自组织的Agent之间相互协商在任务规模小的时候很灵活但一旦超过几十个Agent问题排查起来会非常痛苦。中心化编排就像路口装了红绿灯虽然多了一道调度环节但至少不会出现所有车都挤在路口互相僵持的局面。Agent内部保持自治外部行为由编排层统一约束两个世界的优点都拿到了。第二个取舍是传递数据还是传递引用。早期我图省事所有中间结果直接拼到上下文里再传给下一个Agent结果就是上下文迅速膨胀成本跟着起飞。后来改成“引用传递按需取数”上一个Agent只传一个结果ID下一个Agent需要具体内容时再根据ID去取。虽然多了一次网络开销但上下文稳定了很多排障时也更清楚谁拿了什么数据。第三个取舍是工具全量注册还是按需注入。全量注册开发时最省事但到了线上就是一个隐患——Agent可能调用到不属于它权限范围的工具。我后来改成按能力清单匹配工具Agent注册能力时同时声明它需要哪些工具路由层在分发任务时把相关工具挂到本次执行环境里。这个思路在后面实现部分会再展开。2. 核心模块拆解注册、路由、工具与记忆写Agent-Reach最让我花时间的是四个模块注册中心、任务路由、工具执行层、上下文管理。它们各自解决一个层面的问题下面逐个拆开讲。这节偏概念但每一个设计点后面都有对应的代码实现建议结合第3章一起看。2.1 Agent注册与能力描述Agent-Reach里每个Agent上线时都要做一件事注册自己的能力描述。能力描述是路由层判断“这个任务能不能交给它”的依据所以一定要写得结构化。我一开始让Agent用自然语言描述能力结果路由判断一会儿能用一会儿不能用后来全部改成结构化声明效果立刻稳定下来。一份能力描述至少包含三块agent_id、name、capabilities。capabilities里的每一项都要写明能力名、输入参数、输出类型。下面是我项目里一个文档分析Agent的注册信息{ agent_id: doc_agent, name: 文档分析Agent, capabilities: [ { name: 文档解析, inputs: {file_path: string, max_pages: integer}, outputs: ExtractedText }, { name: 摘要生成, inputs: {text: string, style: string}, outputs: SummaryText } ], required_tools: [file_reader, text_splitter] }这个结构有几个设计点值得注意。第一能力名是路由匹配的主键取名尽量短、语义唯一“文档解析”就不要同时出现在另一个Agent头上否则路由会产生歧义。第二inputs的Schema要写完整路由层靠它做参数校验缺字段或类型不对的任务会直接淘汰。第三required_tools声明了Agent执行上述能力时需要用到的工具这个声明决定了工具注入范围也是权限控制的第一道闸门。注册信息的存储方面我在第一版里用的是内存字典Agent实例启动时上报接口设计成幂等的。同一个agent_id反复注册只保留最后一次避免Agent重启时产生脏数据。项目规模大了之后可以换到Redis或者配置中心但核心接口可以保持不变。2.2 任务路由与调度策略路由层是Agent-Reach的心脏。它的输入是一个结构化任务输出是这个任务应该交给哪个Agent处理。我在项目里把它拆成三个步骤按顺序执行。第一步是能力匹配。通过能力名反查Agent列表找到有能力处理该任务的候选Agent。关键字匹配速度快但泛化能力差。用户说“帮我查下天气”和任务里的“天气查询”语义完全相同关键字匹配却可能漏掉所以我加了一层轻量语义匹配用向量相似度兜底确保近义表达不会落空。第二步是参数匹配。任务里携带的inputs字段要和Agent能力声明的inputs做类型检查类型不兼容的直接淘汰。比如任务里传了file_path但Agent声明的是url即使能力名匹配也不能进入候选集。这一步能拦掉大量“Agent看懂了但没法执行”的尴尬情况。第三步是负载均衡。多个候选Agent都能干这个任务时我选当前负载最低的一个。负载的计算方法很简单每个Agent维护一个pending队列队列长度就是它的负载值。调度完成后在任务的完成回调里把负载减回去这样新任务不会总是砸向最强的Agent而是更平均地分摊给所有可用的执行者。下面是一段简化版的调度逻辑展示整体思路真实版本比这个多了一些异常处理和审计埋点class TaskRouter: def __init__(self, registry): self.registry registry def route(self, task): candidates self.registry.match(task.capability, task.inputs) if not candidates: return None def _load(agent_id): return self.registry.get_agent(agent_id)[pending] best min(candidates, keylambda item: _load(item[0])) self.registry.increase_pending(best[0]) return best[0]真实的业务里我在路由前还加了一层“任务校验”任务必须带trace_id、capability、inputs缺一项就返回400。宁可让调用方补参数也不要靠猜。2.3 统一工具调用层多Agent系统里最脏最乱的部分通常是工具调用。有的Agent直接调外部API有的直接查数据库有的自己读文件每个Agent都各接一套。一旦出问题连工具正在被谁调用都定位不到。Agent-Reach把所有工具收敛到一个统一执行层对外只暴露一个接口execute(tool_name, params)。这个接口的职责非常简单找到工具实现、校验参数、执行、返回结构化结果。Agent不需要知道工具底层是HTTP请求还是本地函数只需要传对参数然后接收结果就行。这里其实和业内最近常提的“统一工具协议”思路是相通的工具描述用统一的Schema执行入口用统一的请求对象。我自己的做法没有上很重的框架而是用一个轻量的ToolExecutor把工具注册起来注册时用Schema描述参数。核心代码长这样class ToolExecutor: def __init__(self): self._tools {} def register_tool(self, name, handler, input_schema): self._tools[name] { handler: handler, input_schema: input_schema, } def execute(self, tool_name, params): if tool_name not in self._tools: raise ToolNotFoundError(tool_name) schema self._tools[tool_name][input_schema] validated self._validate(params.copy(), schema) return self._tools[tool_name][handler](**validated)我把参数校验特意提到执行之前因为大模型给出的参数经常多一个或者少一个。一旦少了一个必填字段直接塞给handler抛的异常会很难定位。统一校验之后错误信息可以写成“参数缺file_path”这种明确的提示Agent在下一次调用时自己就能纠正过来。另一个细节是工具返回结果一定要做结构化包装。不要只返回一串字符串而是返回类似{status: ok, data: {...}}的结构。Agent判断“工具调用是否成功”会更直接也不用在长文本里找结果。这个细节帮我节省了大量调试时间。2.4 上下文管理与记忆分层上下文管理听起来不性感但实践里是最容易踩坑的模块。Agent-Reach里面向多Agent一套上下文处理我分了三个层级会话级上下文保存一次对话的所有消息由所有参与Agent共享设置硬上限。超了就早期消息汇总成摘要放回上下文头部这个机制后面会细讲。任务级上下文保存一次具体任务内的中间结果只在任务过程中存在任务完成即丢弃。持久层长期记忆用于跨会话检索当Agent发现当前任务和之前某次任务相关时会主动去查询。层级生命周期存储位置用途会话级上下文一次对话内存/Redis多Agent协作的基础信息任务级上下文单个任务内存临时变量避免任务间数据污染长期记忆跨会话向量库/JSON辅助Agent做同类判断3. 实操落地从零搭一个Agent-Reach最小实现光讲概念不够直观这里我直接给出一个可以跑起来的最小实现。没法贴完整工程但关键部分的结构是能照着搭的。整个实现大概只需要几个文件适合拿来做原型验证。3.1 环境准备与依赖选择我自己的开发环境是Python 3.11依赖选了三个Redis 7、FastAPI、OpenAI兼容接口的SDK。为什么用Redis因为它的List结构非常适合做pending队列而且所有Agent实例都能看到同一份状态避免了单机内存不一致的问题。FastAPI用来暴露Agent-Reach的HTTP入口因为多Agent系统天然是一个分布式结构用HTTP接口比进程内函数调用更接近生产状态。大模型推理部分则通过环境变量配置base_url和api_key这样本地跑和上生产都方便。启动环境没有太多坑但有一个细节Redis的键名一定要带服务名前缀比如agent_reach:pending避免和别的服务冲突。这个习惯能让你在排查问题的时候省下不少时间。3.2 注册中心实现注册中心我放在一个叫registry.py的模块里。代码不长但边界情况都考虑到了import time from typing import Dict, Optional class AgentRegistry: def __init__(self): self._agents: Dict[str, dict] {} def register(self, agent_id: str, name: str, capabilities: list, required_tools: list, endpoint: str) - None: self._agents[agent_id] { name: name, capabilities: capabilities, required_tools: required_tools, endpoint: endpoint, pending: 0, updated_at: int(time.time()), } def unregister(self, agent_id: str) - None: self._agents.pop(agent_id, None) def get(self, agent_id: str) - Optional[dict]: return self._agents.get(agent_id) def match(self, capability: str, inputs: dict) - list: candidates [] for agent_id, info in self._agents.items(): for cap in info[capabilities]: if cap[name] ! capability: continue missing [k for k in inputs.keys() if k not in cap[inputs]] if missing: continue candidates.append((agent_id, cap, info)) return candidatesregister直接以agent_id为键赋值幂等覆盖旧数据Agent重启时不会产生脏数据。unregister用pop而不是先判断再删因为pop对不存在的键返回None天然支持重复调用。match函数里顺带做了参数校验任务传入的字段在能力声明里必须都有这是路由正确性的基础。3.3 路由调度与任务生命周期管理路由层除了能力匹配和选择Agent还要管理任务的整个生命周期。我加了一个简单状态机pending、running、completed、failed。任务创建时进入pending调度成功后进入running收到Agent的回调后变成completed或failed。任务状态我放在Redis里而不是内存里因为Agent可能是多实例部署的调度进程和Agent执行进程是分离的。只有内存的话重启调度进程时所有任务状态就全丢了。放到Redis即使调度进程重启数据依然在任务还能接续处理。import json class TaskLifecycle: def __init__(self, redis_client): self.redis redis_client def create(self, trace_id: str, capability: str, inputs: dict): task { trace_id: trace_id, capability: capability, inputs: inputs, status: pending, agent_id: None, created_at: int(time.time()), } self.redis.set(fagent_reach:task:{trace_id}, json.dumps(task)) return task def assign(self, trace_id: str, agent_id: str) - None: self.redis.hset(fagent_reach:task:{trace_id}, mapping{ status: running, agent_id: agent_id, }) def finish(self, trace_id: str, status: str, result: str) - None: self.redis.hset(fagent_reach:task:{trace_id}, mapping{ status: status, result: result, })3.4 工具执行器与一个具体工具工具执行器的核心逻辑在2.3已经给出这里补一个具体例子。假设我要让Agent能查天气那就注册一个get_weather工具。真实场景中工具实现会调用第三方天气API这里简化为本地函数保证大家都能跑通。def get_weather(location: str, date: str) - dict: return { location: location, date: date, temperature: 18-24, condition: 晴转多云, } executor.register_tool(get_weather, get_weather, { location: {type: string, required: True}, date: {type: string, required: True}, })这个例子很小但展示了一个关键模式任何外部能力都可以包装成“输入参数输出JSON”的标准工具。工具注册完之后Agent调用它只需要传参数而不需要关心天气API的地址、鉴权方式、参数格式。也就是说工具背后的实现变了Agent端的调用代码完全不用改这层抽象的价值会在工具频繁变更时体现得很明显。3.5 完整示例意图拆解与Agent接力最后跑一个真实的端到端例子。用户说“帮我查一下今天的天气然后生成一段50字的车间安全提醒。”意图Agent先分析出两个能力天气查询、文本生成。Agent-Reach路由层收到两个子任务。第一个子任务capabilityweather_queryinputs{location: 北京, date: 2026-01-15}。路由选到weather_agentweather_agent通过ToolExecutor调用get_weather拿到结构化天气数据。它把结果返回给编排层编排层把结果挂到任务上下文中。第二个子任务capabilitytext_generationinputs{content: 天气数据, style: 安全提醒}。路由选到writer_agentwriter_agent基于天气数据生成一段简短的提醒文案。最后编排层把writer_agent的输出作为最终结果返回给用户。这个流程里没有任何一个Agent知道全局细节weather_agent不知道后面还会有文本生成writer_agent也不知道天气数据是从哪来的。每个Agent只处理自己负责的那一段再把结果交回给编排层拼装。我把这种模式叫“接力式编排”——拆解大任务各Agent接力完成各自的片段编排层负责递棒和监督。对复杂任务来说这种模式比硬塞给一个Agent做到底更稳定也更方便扩容。4. 常见问题与排障技巧实录这章写我实际项目里踩过的坑每一条都附带排查路径和解决方案可以直接对照着用。很多问题不看真实日志很难想到是哪个环节出的错这里有我自己摸出来的方法。4.1 共享上下文爆掉的坑症状很典型对话进行到第5轮每个Agent的响应开始变慢甚至直接报上下文超限。我去查日志发现共享上下文里存下了每一次工具调用返回的完整文本。原因很简单上下文是只增不减的。多个Agent共用一份上下文A的中间结果、B的工具返回、C的长文本都往里面塞十次调用之后轻松超过几万token。我最初以为模型回答是大头结果发现工具API返回的JSON经过格式化以后占了60%以上的token这非常反直觉。排查方法是在编排层给每一条入上下文的消息打一个token数日志然后观察哪个环节引入的token最大。解决方案有三个一是中间结果不进共享上下文而放进任务级上下文只有摘要结果进会话语义层二是给共享上下文设置硬上限超了就用摘要替换历史三是工具返回的原始文本先做截断比如只保留前500字符。这三招下来上下文膨胀的问题基本被压住了。4.2 Agent之间互相踢皮球踢皮球这个场景很有意思。A、B两个Agent围绕一个任务来回调用每次都在回复“这个该由你来处理”。日志里能看到同一条任务链的trace_id在A和B之间反复出现就像两个人在群里互相艾特谁也不干活。根因一般是路由匹配条件太宽A和B的能力声明有重叠而且任务没有设置最大流转跳数。依赖模型自己判断“什么时候该停止移交”在真实场景里太不可控了。我的对策分三步。第一每个任务设置最大流转跳数我用的上限是5超过就强制走兜底Agent。兜底Agent哪怕只是返回“需要人工介入”也比无限循环好得多。第二路由记录任务经过的Agent链已经出现过的agent_id不允许再次出现在同一个任务链里从结构上消除回环。第三给能力声明增加“唯一归属”约束同一能力名只允许一个Agent声明如果有多个Agent都声明在注册阶段就要纠正不纠正就不让注册成功。一个特别实用的排查技巧给每条任务打上trace_id日志里看到同一个trace_id在不同Agent之间来回横跳基本就可以断定流转异常了。4.3 工具调用超时与重试另一个高频事故是Agent在工具调用时卡住10分钟不返回结果。查日志发现它调用的外部API一直没有响应而模型就这么干等着这个HTTP请求完成整个任务被拖死。问题在于工具调用没有超时限制失败后也没有重试机制。真实场景里外部API的响应时间非常不稳定必须显式控制。我用asyncio.wait_for来实现超时简单直接import asyncio async def call_with_timeout(executor, tool_name, params, timeout10): try: return await asyncio.wait_for( executor.execute_async(tool_name, params), timeouttimeout, ) except asyncio.TimeoutError: return {status: error, error: timeout}工具默认超时10秒重试一次。第一次超时后等待2秒再试第二次还是超时就直接返回结构化错误。这里有个细节需要特别注意不要让模型拿着“超时”信息去自由发挥比如自行猜测天气结果。那样会引入幻觉数据用户拿到错误信息反而以为是真的。错误结果里要明确写“服务暂时不可用”让Agent把它转述给用户。4.4 权限边界模糊与工具滥用有一次我在调试时发现一位同事的Agent竟然调用了另一个团队的内部接口。虽然不是恶意但已经明显越权了。根因是最初把所有工具都放进了全局工具列表结果不写代码的Agent启动后也带着一大串工具声明相当于给了员工一把万能钥匙期望他自己不乱开门。对策是严格建立“工具-能力”绑定关系每次Agent注册时声明required_tools注册中心校验这些工具是否在该Agent的允许名单里运行时编排层根据任务能力把允许的工具挂到执行环境其他工具一概不能访问。高危工具还要额外加二次确认例如发送邮件、修改数据这类操作需要用户先确认Agent才能执行。这个过程中我总结出的原则是工具权限不能靠模型自觉要靠架构边界。模型可以把工具参数编得头头是道但如果它根本没有这个工具的调用权那再会编也没有用。问题直接表现优先排查点上下文爆掉响应慢、超限报错token日志中间结果是否误入共享上下文Agent循环调用同一trace_id在两个Agent间反复出现跳数限制Agent链去重工具超时Agent长时间无响应是否设置超时是否重试权限越权Agent调用了不该调的工具required_tools是否与能力绑定运行时是否过滤这套东西我从零搭到现在最深的体会是Agent编排的核心不在模型选择而在把“触达”这件事理清楚。只要让每个Agent知道它该用什么工具、该找谁合作、该把结果交给谁大部分协作痛都会消失。Agent-Reach目前还在我的内部环境里跑着下一步我准备把路由权重改为可学习的再给任务加上更细的审计日志让每个任务从哪来到哪去都清晰可查。如果你也在折腾多Agent欢迎来交流踩坑的细节随时可以细聊。 Agent-Reach 这个名字最初其实是我本地项目里随手起的代号。当时我在折腾一套多智能体协作系统遇到的问题让我印象特别深单个Agent做得再好一旦把多个Agent拼在一起协作就变得很脆弱。有的Agent拿到了工具却不会传参数有的Agent在等待别的Agent返回结果时把上下文窗口直接塞满了最让人崩溃的是两个Agent为了一个小问题互相踢皮球来回调用了几十轮。我把这些现象复盘了一遍发现核心矛盾其实是一个词触达。Agent没有可靠的触达方式——触达该用的工具、该拿的数据、该协作的下一个Agent。所以我抽了一个独立的编排层取名Agent-Reach。这篇文章把Agent-Reach的设计思路、最小实现和排障经验写出来给正在做多Agent编排、工具接入或者企业内部AI助手的同学参考。1. Agent-Reach是什么为什么我在多智能体协作里卡了壳1.1 单Agent好做多Agent难在“触达”现在随便一个开源框架都能让单个Agent玩得很好接上大模型、挂上几个工具它就能完成不少任务。单Agent的思路其实非常简单一个大模型当大脑外面挂一圈工具模型决定调哪个工具、传什么参数、什么时候收手。这种模式几乎没有协作成本所以开发体验很好。但生产环境里真正难的是多Agent协作。我打个比方单个Agent就像一家全能小店店员知道自己货架上有啥、仓库在哪。可当多个Agent放在一起就好像一个商场里有几十个专柜顾客要买东西时需要的其实是导购——告诉他去几楼、找哪个柜台。多Agent系统的问题也一样任务不知道交给谁、工具不知道去哪调、信息不知道和谁同步。这些问题的共同点都指向同一个词“触达能力不够”。我在一个实际项目里手上捏着N个Agent分别负责文档解析、信息检索、代码补全、报告生成。单测每个都不错可一旦把它们组合起来跑一个完整业务流程就会频繁出现“任务派不下去”“派下去没人接”“接了之后不回复”的情况。我一度以为是模型能力不行后来才发现缺的是一个能让Agent之间互相“找到对方”的中间层。于是Agent-Reach的第一版就走上了这条路不重新发明Agent而是把Agent之间、Agent与工具之间的触达关系管起来。1.2 我理解的“Reach”有三层含义第一层是触达工具。Agent要用工具但没必要让每个Agent看到全部工具。代码生成Agent不该看到日志清理工具文档处理Agent也不该拿到运维治理接口。Agent-Reach让工具统一收口到一个独立层里按Agent的能力目标动态注入避免“工具一多就乱套”。第二层是触达伙伴。一个任务往往需要多个Agent接力完成。Agent-Reach维护着一张“Agent能力地图”路由层根据任务要求找到最合适的Agent并且知道这个Agent当前忙不忙、有没有能力承接。这样一来Agent之间不再需要互相猜身份而是直接通过调度层建立协作关系整个系统的可预测性会好很多。第三层是触达目标。任务从进来、拆解、分配到最终结果返回整个过程必须有清晰的状态流转。一个任务交给Agent-AA回复说做完了但结果还没被合并进原对话这就是目标丢失。Agent-Reach对这种状态做了显式记录每一层都知道任务正在谁手里、还差什么、下一步要找谁。这三层合在一起就是我对Agent-Reach的定位一个让多Agent系统能“找得到、调得到、收得回”的编排层。如果你觉得这个定义有点抽象没关系后面几段我会把每个部分拆开讲再附上一套可运行的最小实现。1.3 关键的设计取舍第一个取舍是中心化编排还是完全自组织。我最终选了中心化因为它可控、可观测。完全自组织的Agent之间相互协商在任务规模小的时候很灵活但一旦超过几十个Agent问题排查起来会非常痛苦。中心化编排就像路口装了红绿灯虽然多了一道调度环节但至少不会出现所有车都挤在路口互相僵持的局面。Agent内部保持自治外部行为由编排层统一约束两个世界的优点都拿到了。第二个取舍是传递数据还是传递引用。早期我图省事所有中间结果直接拼到上下文里再传给下一个Agent结果就是上下文迅速膨胀成本跟着起飞。后来改成“引用传递按需取数”上一个Agent只传一个结果ID下一个Agent需要具体内容时再根据ID去取。虽然多了一次网络开销但上下文稳定了很多排障时也更清楚谁拿了什么数据。第三个取舍是工具全量注册还是按需注入。全量注册开发时最省事但到了线上就是一个隐患——Agent可能调用到不属于它权限范围的工具。我后来改成按能力清单匹配工具Agent注册能力时同时声明它需要哪些工具路由层在分发任务时把相关工具挂到本次执行环境里。这个思路在后面实现部分会再展开。2. 核心模块拆解注册、路由、工具与记忆写Agent-Reach最让我花时间的是四个模块注册中心、任务路由、工具执行层、上下文管理。它们各自解决一个层面的问题下面逐个拆开讲。这节偏概念但每一个设计点后面都有对应的代码实现建议结合第3章一起看。2.1 Agent注册与能力描述Agent-Reach里每个Agent上线时都要做一件事注册自己的能力描述。能力描述是路由层判断“这个任务能不能交给它”的依据所以一定要写得结构化。我一开始让Agent用自然语言描述能力结果路由判断一会儿能用一会儿不能用后来全部改成结构化声明效果立刻稳定下来。一份能力描述至少包含三块agent_id、name、capabilities。capabilities里的每一项都要写明能力名、输入参数、输出类型。下面是我项目里一个文档分析Agent的注册信息{ agent_id: doc_agent, name: 文档分析Agent, capabilities: [ { name: 文档解析, inputs: {file_path: string, max_pages: integer}, outputs: ExtractedText }, { name: 摘要生成, inputs: {text: string, style: string}, outputs: SummaryText } ], required_tools: [file_reader, text_splitter] }这个结构有几个设计点值得注意。第一能力名是路由匹配的主键取名尽量短、语义唯一“文档解析”就不要同时出现在另一个Agent头上否则路由会产生歧义。第二inputs的Schema要写完整路由层靠它做参数校验缺字段或类型不对的任务会直接淘汰。第三required_tools声明了Agent执行上述能力时需要用到的工具这个声明决定了工具注入范围也是权限控制的第一道闸门。注册信息的存储方面我在第一版里用的是内存字典Agent实例启动时上报接口设计成幂等的。同一个agent_id反复注册只保留最后一次避免Agent重启时产生脏数据。项目规模大了之后可以换到Redis或者配置中心但核心接口可以保持不变。2.2 任务路由与调度策略路由层是Agent-Reach的心脏。它的输入是一个结构化任务输出是这个任务应该交给哪个Agent处理。我在项目里把它拆成三个步骤按顺序执行。第一步是能力匹配。通过能力名反查Agent列表找到有能力处理该任务的候选Agent。关键字匹配速度快但泛化能力差。用户说“帮我查下天气”和任务里的“天气查询”语义完全相同关键字匹配却可能漏掉所以我加了一层轻量语义匹配用向量相似度兜底确保近义表达不会落空。第二步是参数匹配。任务里携带的inputs字段要和Agent能力声明的inputs做类型检查类型不兼容的直接淘汰。比如任务里传了file_path但Agent声明的是url即使能力名匹配也不能进入候选集。这一步能拦掉大量“Agent看懂了但没法执行”的尴尬情况。第三步是负载均衡。多个候选Agent都能干这个任务时我选当前负载最低的一个。负载的计算方法很简单每个Agent维护一个pending队列队列长度就是它的负载值。调度完成后在任务的完成回调里把负载减回去这样新任务不会总是砸向最强的Agent而是更平均地分摊给所有可用的执行者。下面是一段简化版的调度逻辑展示整体思路真实版本比这个多了一些异常处理和审计埋点class TaskRouter: def __init__(self, registry): self.registry registry def route(self, task): candidates self.registry.match(task.capability, task.inputs) if not candidates: return None def _load(agent_id): return self.registry.get_agent(agent_id)[pending] best min(candidates, keylambda item: _load(item[0])) self.registry.increase_pending(best[0]) return best[0]真实的业务里我在路由前还加了一层“任务校验”任务必须带trace_id、capability、inputs缺一项就返回400。宁可让调用方补参数也不要靠猜。2.3 统一工具调用层多Agent系统里最脏最乱的部分通常是工具调用。有的Agent直接调外部API有的直接查数据库有的自己读文件每个Agent都各接一套。一旦出问题连工具正在被谁调用都定位不到。Agent-Reach把所有工具收敛到一个统一执行层对外只暴露一个接口execute(tool_name, params)。这个接口的职责非常简单找到工具实现、校验参数、执行、返回结构化结果。Agent不需要知道工具底层是HTTP请求还是本地函数只需要传对参数然后接收结果就行。这里其实和业内最近常提的“统一工具协议”思路是相通的工具描述用统一的Schema执行入口用统一的请求对象。我自己的做法没有上很重的框架而是用一个轻量的ToolExecutor把工具注册起来注册时用Schema描述参数。核心代码长这样class ToolExecutor: def __init__(self): self._tools {} def register_tool(self, name, handler, input_schema): self._tools[name] { handler: handler, input_schema: input_schema, } def execute(self, tool_name, params): if tool_name not in self._tools: raise ToolNotFoundError(tool_name) schema self._tools[tool_name][input_schema] validated self._validate(params.copy(), schema) return self._tools[tool_name][handler](**validated)我把参数校验特意提到执行之前因为大模型给出的参数经常多一个或者少一个。一旦少了一个必填字段直接塞给handler抛的异常会很难定位。统一校验之后错误信息可以写成“参数缺file_path”这种明确的提示Agent在下一次调用时自己就能纠正过来。另一个细节是工具返回结果一定要做结构化包装。不要只返回一串字符串而是返回类似{status: ok, data: {...}}的结构。Agent判断“工具调用是否成功”会更直接也不用在长文本里找结果。这个细节帮我节省了大量调试时间。2.4 上下文管理与记忆分层上下文管理听起来不性感但实践里是最容易踩坑的模块。Agent-Reach里面向多Agent一套上下文处理我分了三个层级会话级上下文保存一次对话的所有消息由所有参与Agent共享设置硬上限。超了就早期消息汇总成摘要放回上下文头部这个机制后面会细讲。任务级上下文保存一次具体任务内的中间结果只在任务过程中存在任务完成即丢弃。持久层长期记忆用于跨会话检索当Agent发现当前任务和之前某次任务相关时会主动去查询。层级生命周期存储位置用途会话级上下文一次对话内存/Redis多Agent协作的基础信息任务级上下文单个任务内存临时变量避免任务间数据污染长期记忆跨会话向量库/JSON辅助Agent做同类判断3. 实操落地从零搭一个Agent-Reach最小实现光讲概念不够直观这里我直接给出一个可以跑起来的最小实现。没法贴完整工程但关键部分的结构是能照着搭的。整个实现大概只需要几个文件适合拿来做原型验证。3.1 环境准备与依赖选择我自己的开发环境是Python 3.11依赖选了三个Redis 7、FastAPI、OpenAI兼容接口的SDK。为什么用Redis因为它的List结构非常适合做pending队列而且所有Agent实例都能看到同一份状态避免了单机内存不一致的问题。FastAPI用来暴露Agent-Reach的HTTP入口因为多Agent系统天然是一个分布式结构用HTTP接口比进程内函数调用更接近生产状态。大模型推理部分则通过环境变量配置base_url和api_key这样本地跑和上生产都方便。启动环境没有太多坑但有一个细节Redis的键名一定要带服务名前缀比如agent_reach:pending避免和别的服务冲突。这个习惯能让你在排查问题的时候省下不少时间。3.2 注册中心实现注册中心我放在一个叫registry.py的模块里。代码不长但边界情况都考虑到了import time from typing import Dict, Optional class AgentRegistry: def __init__(self): self._agents: Dict[str, dict] {} def register(self, agent_id: str, name: str, capabilities: list, required_tools: list, endpoint: str) - None: self._agents[agent_id] { name: name, capabilities: capabilities, required_tools: required_tools, endpoint: endpoint, pending: 0, updated_at: int(time.time()), } def unregister(self, agent_id: str) - None: self._agents.pop(agent_id, None) def get(self, agent_id: str) - Optional[dict]: return self._agents.get(agent_id) def match(self, capability: str, inputs: dict) - list: candidates [] for agent_id, info in self._agents.items(): for cap in info[capabilities]: if cap[name] ! capability: continue missing [k for k in inputs.keys() if k not in cap[inputs]] if missing: continue candidates.append((agent_id, cap, info)) return candidatesregister直接以agent_id为键赋值幂等覆盖旧数据Agent重启时不会产生脏数据。unregister用pop而不是先判断再删因为pop对不存在的键返回None天然支持重复调用。match函数里顺带做了参数校验任务传入的字段在能力声明里必须都有这是路由正确性的基础。3.3 路由调度与任务生命周期管理路由层除了能力匹配和选择Agent还要管理任务的整个生命周期。我加了一个简单状态机pending、running、completed、failed。任务创建时进入pending调度成功后进入running收到Agent的回调后变成completed或failed。任务状态我放在Redis里而不是内存里因为Agent可能是多实例部署的调度进程和Agent执行进程是分离的。只有内存的话重启调度进程时所有任务状态就全丢了。放到Redis即使调度进程重启数据依然在任务还能接续处理。import json class TaskLifecycle: def __init__(self, redis_client): self.redis redis_client def create(self, trace_id: str, capability: str, inputs: dict): task { trace_id: trace_id, capability: capability, inputs: inputs, status: pending, agent_id: None, created_at: int(time.time()), } self.redis.set(fagent_reach:task:{trace_id}, json.dumps(task)) return task def assign(self, trace_id: str, agent_id: str) - None: self.redis.hset(fagent_reach:task:{trace_id}, mapping{ status: running, agent_id: agent_id, }) def finish(self, trace_id: str, status: str, result: str) - None: self.redis.hset(fagent_reach:task:{trace_id}, mapping{ status: status, result: result, })3.4 工具执行器与一个具体工具工具执行器的核心逻辑在2.3已经给出这里补一个具体例子。假设我要让Agent能查天气那就注册一个get_weather工具。真实场景中工具实现会调用第三方天气API这里简化为本地函数保证大家都能跑通。def get_weather(location: str, date: str) - dict: return { location: location, date: date, temperature: 18-24, condition: 晴转多云, } executor.register_tool(get_weather, get_weather, { location: {type: string, required: True}, date: {type: string, required: True}, })这个例子很小但展示了一个关键模式任何外部能力都可以包装成“输入参数输出JSON”的标准工具。工具注册完之后Agent调用它只需要传参数而不需要关心天气API的地址、鉴权方式、参数格式。也就是说工具背后的实现变了Agent端的调用代码完全不用改这层抽象的价值会在工具频繁变更时体现得很明显。3.5 完整示例意图拆解与Agent接力最后跑一个真实的端到端例子。用户说“帮我查一下今天的天气然后生成一段50字的车间安全提醒。”意图Agent先分析出两个能力天气查询、文本生成。Agent-Reach路由层收到两个子任务。第一个子任务capabilityweather_queryinputs{location: 北京, date: 2026-01-15}。路由选到weather_agentweather_agent通过ToolExecutor调用get_weather拿到结构化天气数据。它把结果返回给编排层编排层把结果挂到任务上下文中。第二个子任务capabilitytext_generationinputs{content: 天气数据, style: 安全提醒}。路由选到writer_agentwriter_agent基于天气数据生成一段简短的提醒文案。最后编排层把writer_agent的输出作为最终结果返回给用户。这个流程里没有任何一个Agent知道全局细节weather_agent不知道后面还会有文本生成writer_agent也不知道天气数据是从哪来的。每个Agent只处理自己负责的那一段再把结果交回给编排层拼装。我把这种模式叫“接力式编排”——拆解大任务各Agent接力完成各自的片段编排层负责递棒和监督。对复杂任务来说这种模式比硬塞给一个Agent做到底更稳定也更方便扩容。4. 常见问题与排障技巧实录这章写我实际项目里踩过的坑每一条都附带排查路径和解决方案可以直接对照着用。很多问题不看真实日志很难想到是哪个环节出的错这里有我自己摸出来的方法。4.1 共享上下文爆掉的坑症状很典型对话进行到第5轮每个Agent的响应开始变慢甚至直接报上下文超限。我去查日志发现共享上下文里存下了每一次工具调用返回的完整文本。原因很简单上下文是只增不减的。多个Agent共用一份上下文A的中间结果、B的工具返回、C的长文本都往里面塞十次调用之后轻松超过几万token。我最初以为模型回答是大头结果发现工具API返回的JSON经过格式化以后占了60%以上的token这非常反直觉。排查方法是在编排层给每一条入上下文的消息打一个token数日志然后观察哪个环节引入的token最大。解决方案有三个一是中间结果不进共享上下文而放进任务级上下文只有摘要结果进会话语义层二是给共享上下文设置硬上限超了就用摘要替换历史三是工具返回的原始文本先做截断比如只保留前500字符。这三招下来上下文膨胀的问题基本被压住了。4.2 Agent之间互相踢皮球踢皮球这个场景很有意思。A、B两个Agent围绕一个任务来回调用每次都在回复“这个该由你来处理”。日志里能看到同一条任务链的trace_id在A和B之间反复出现就像两个人在群里互相艾特谁也不干活。根因一般是路由匹配条件太宽A和B的能力声明有重叠而且任务没有设置最大流转跳数。依赖模型自己判断“什么时候该停止移交”在真实场景里太不可控了。我的对策分三步。第一每个任务设置最大流转跳数我用的上限是5超过就强制走兜底Agent。兜底Agent哪怕只是返回“需要人工介入”也比无限循环好得多。第二路由记录任务经过的Agent链已经出现过的agent_id不允许再次出现在同一个任务链里从结构上消除回环。第三给能力声明增加“唯一归属”约束同一能力名只允许一个Agent声明如果有多个Agent都声明在注册阶段就要纠正不纠正就不让注册成功。一个特别实用的排查技巧给每条任务打上trace_id日志里看到同一个trace_id在不同Agent之间来回横跳基本就可以断定流转异常了。4.3 工具调用超时与重试另一个高频事故是Agent在工具调用时卡住10分钟不返回结果。查日志发现它调用的外部API一直没有响应而模型就这么干等着这个HTTP请求完成整个任务被拖死。问题在于工具调用没有超时限制失败后也没有重试机制。真实场景里外部API的响应时间非常不稳定必须显式控制。我用asyncio.wait_for来实现超时简单直接import asyncio async def call_with_timeout(executor, tool_name, params, timeout10): try: return await asyncio.wait_for( executor.execute_async(tool_name, params), timeouttimeout, ) except asyncio.TimeoutError: return {status: error, error: timeout}工具默认超时10秒重试一次。第一次超时后等待2秒再试第二次还是超时就直接返回结构化错误。这里有个细节需要特别注意不要让模型拿着“超时”信息去自由发挥比如自行猜测天气结果。那样会引入幻觉数据用户拿到错误信息反而以为是真的。错误结果里要明确写“服务暂时不可用”让Agent把它转述给用户。4.4 权限边界模糊与工具滥用有一次我在调试时发现一位同事的Agent竟然调用了另一个团队的内部接口。虽然不是恶意但已经明显越权了。根因是最初把所有工具都放进了全局工具列表结果不写代码的Agent启动后也带着一大串工具声明相当于给了员工一把万能钥匙期望他自己不乱开门。对策是严格建立“工具-能力”绑定关系每次Agent注册时声明required_tools注册中心校验这些工具是否在该Agent的允许名单里运行时编排层根据任务能力把允许的工具挂到执行环境其他工具一概不能访问。高危工具还要额外加二次确认例如发送邮件、修改数据这类操作需要用户先确认Agent才能执行。这个过程中我总结出的原则是工具权限不能靠模型自觉要靠架构边界。模型可以把工具参数编得头头是道但如果它根本没有这个工具的调用权那再会编也没有用。问题直接表现优先排查点上下文爆掉响应慢、超限报错token日志中间结果是否误入共享上下文Agent循环调用同一trace_id在两个Agent间反复出现跳数限制Agent链去重工具超时Agent长时间无响应是否设置超时是否重试权限越权Agent调用了不该调的工具required_tools是否与能力绑定运行时是否过滤这套东西我从零搭到现在最深的体会是Agent编排的核心不在模型选择而在把“触达”这件事理清楚。只要让每个Agent知道它该用什么工具、该找谁合作、该把结果交给谁大部分协作痛都会消失。Agent-Reach目前还在我的内部环境里跑着下一步我准备把路由权重改为可学习的再给任务加上更细的审计日志让每个任务从哪来到哪去都清晰可查。如果你也在折腾多Agent欢迎来交流踩坑的细节随时可以细聊。