做过一段时间的 Agent 项目手头最让我满意的一个就是叫 Agent-Reach 的东西。简单说它是一套负责“让多个 AI Agent 彼此找得到、叫得动、干得完活”的轻量调度中枢。如果你也在搞多智能体协作或者被“Agent 之间互相扯皮、工具越权调用、任务跑一半就沉默”这些问题折磨过那 Agent-Reach 这套设计思路和代码你应该能直接用上。这篇文章不聊虚的我把当时怎么拆需求、怎么选方案、核心代码长什么样、以及掉过的坑都整理出来希望能帮你少走几步弯路。1. 项目拆解Agent-Reach 到底在解决什么问题1.1 核心需求多 Agent 协作时的“触达”难题先说个背景。之前团队里并行跑着好几个独立 Agent有专门做信息检索的有做文档生成的还有负责调用内部工单系统的。单看每一个能力都挺完整。可一旦让它们协作问题就来了找不到人文档生成 Agent 需要一份最新的市场数据它不知道该找“检索 Agent”还是“数据库 Agent”叫不动人就算知道找谁也没有统一的调用方式有的是 HTTP 接口有的是进程内函数还有的是写 RabbitMQ 消息干一半就断检索 Agent 查完数据生成 Agent 刚好超时崩溃整个流程白跑没有过程记录任务失败后没人知道到底卡在哪一步只能靠猜。如果把问题抽象一下本质上就是“触达”出了问题任务触达不到正确的 AgentAgent 触达不到该用的工具调用方触达不到执行结果。Agent-Reach 这个名字就是这么来的——它的核心职责就是把这些“触达”路径统一管起来。1.2 方案选型为什么不自研重编排框架市面上其实有不少成熟的 Agent 编排方案像 LangGraph、AutoGen、CrewAI 这些我当时都评估过。最后放弃不是因为它们不好而是项目环境比较特殊团队内部有不少老系统没有办法全部迁移到某个框架的标准协议里集成方习惯用简单的 HTTP/JSON不希望被某个框架的复杂抽象绑死我们需要非常精细地控制“谁能调谁”这个在通用框架里配置起来反而啰嗦。所以我决定自研一个轻量中枢。核心原则就一条只做路由和状态管理不碰 Agent 的内部实现。Agent 还是原来那个 Agent只是外部统一通过 Agent-Reach 来呼叫。这个决策带来的好处非常直接老系统不需要改造业务逻辑只要包一层适配接口就能接进来。项目启动前我先列了一个对比表把自研和选型的账算清楚了再动手对比项通用编排框架自研 Agent-Reach接入成本需要按框架规范重写交互层包一层 HTTP/消息适配即可权限控制依赖框架自带的角色体系可在路由层自定义任意规则故障定位框架内部链路日志较重轻量票据日志可逐条追踪定制速度受框架版本迭代影响团队自己掌控维护成本升级容易带出兼容问题代码量少约 2000 行现在回头看这个选择是对的。后面所有功能迭代比如加“任务配额”、“熔断限流”都是在路由层直接加代码就完成了不需要等上游框架更新。2. 系统架构与核心设计思路2.1 三层结构接入层、调度层、执行层Agent-Reach 的整体结构我分成了三层每一层只干一件事接入层Entry负责接收外部请求统一把 HTTP、消息队列、定时器触发转成内部的标准任务票据TaskTicket调度层Dispatcher核心逻辑所在负责根据任务票据选择目标 Agent、检查权限、执行分发、跟踪状态、处理失败重试执行层Executor真正的 Agent 运行环境执行具体任务最后把结果回写到调度层。刚开始的时候我差点把调度逻辑直接塞进接入层里后来实际写了一版发现每个入口都要重复处理超时、重试、异常代码瞬间就膨胀了。改成三段式之后新增一种触发方式只需要写一个“把外部请求变成 TaskTicket”的小适配器调度逻辑完全不用动。这个分层还有一个隐藏好处接入层和执行层可以分别独立部署。当时我们就把执行层单独放在另一组 GPU 机器上接入层放在公网出口机器中间通过 Redis Stream 传递票据互不阻塞。2.2 消息总线和 Agent 注册中心的建模各层之间我不搞“点对点直连”而是统一走一个轻量消息总线。这个总线一开始用的是 Redis Stream后来量大了也兼容过 Kafka但核心数据结构没变过。消息总线负责两件事保存 TaskTicket 及其状态变更传递 Agent 的执行结果TaskResult。与此同时还有一个 Agent 注册中心保存了所有 Agent 的元数据。每个 Agent 注册时上报的信息大概是这个样子的{ agent_id: doc_generator_01, name: 文档生成Agent, skills: [markdown, docx, chart], endpoint: http://executor-local:9001/invoke, timeout_ms: 30000, max_concurrency: 5, permissions: [read_market_data, write_report] }调度层的路由逻辑本质上就是“根据任务票据里的目标技能去注册中心里匹配 agent 的 skills”再叠加一层权限与配额校验。这个建模方式相当于给每个 Agent 做了一个“能力名片”路由时不关心 Agent 是谁只关心它能干什么这样后续加新 Agent、下线旧 Agent对调用方完全透明。2.3 关键设计决策为什么用“任务票据”而不是直接调用这是整个 Agent-Reach 设计里最重要的一张牌。所有人一开始都会本能地认为调用一个 Agent 就是“请求-响应”搞那么复杂干嘛一开始我也是这么写的但第一个星期就踩了坑一个执行时间 40 秒的 Agent直接把 HTTP 连接挂超时对端还在跑调用方已经放弃两边状态不一致。所以我引入了 TaskTicket 的概念它本质上是一个“任务契约”里面记录任务整个生命周期所需的全部信息{ ticket_id: f8d9aa31-b2bd-47ba-9a8f-1f2be2c3d001, type: sync_call, source: api_gateway, target_skills: [market_data], payload: {query: 2024年行业报告, top_k: 5}, timeout_ms: 30000, retry_policy: {max_retries: 2, backoff_ms: 1000}, status: pending, created_at: 1700000000, updated_at: 1700000000, parent_ticket_id: null }为什么票据带来这么大的改进因为它让任务变“可寻址”了调用方可以凭 ticket_id 随时查询状态不需要一直占着连接调度层可以做补偿超时了改状态失败了按策略重试任务之间可以精确记录父子关系形成一棵调用树排查问题的时候一眼就看出谁等了谁、谁失败导致的连锁反应。用生活类比的话以前的直连调用像是当街喊人帮忙喊完就堵在那等用了 TaskTicket 之后相当于去快递站下了单整个包裹在哪、走到哪一步都有记录哪怕中转站机器坏了也能靠运单号重新调度。3. 实操过程从零搭建 Agent-Reach3.1 环境准备与基础数据结构下面这部分是实际操作我按 Python 3.10 FastAPI Redis Stream 来讲。这是最省事、也最容易复现的组合。环境准备好之后先把基础数据结构定义好这是后面一切逻辑的地基。# dataclasses.py from dataclasses import dataclass, field from typing import Any, Dict, List, Optional import uuid import time dataclass class TaskTicket: ticket_id: str field(default_factorylambda: str(uuid.uuid4())) type: str sync_call source: str unknown target_skills: List[str] field(default_factorylist) payload: Dict[str, Any] field(default_factorydict) timeout_ms: int 30000 retry_policy: Dict[str, Any] field( default_factorylambda: {max_retries: 2, backoff_ms: 1000} ) status: str pending # pending | routed | running | succeeded | failed | timeout created_at: float field(default_factorytime.time) updated_at: float field(default_factorytime.time) parent_ticket_id: Optional[str] None agent_id: Optional[str] None result: Optional[Dict[str, Any]] None error_info: Optional[str] None这里我给每个字段都写清楚了默认值实际项目里可以少写很多样板代码。尤其是 ticket_id 自动生成和 createdAt 用time.time()省了后面大量手工赋值。3.2 核心代码注册中心、任务路由、重试与超时Agent 注册中心我用了最简单的 dict threading.Lock没有引入复杂的外部存储。对于几千个 Agent 以内的规模完全够用而且重启恢复也容易扫一遍各 Executor 的监听端口就能重建。# registry.py import threading from typing import Dict, Optional class AgentRegistry: def __init__(self): self._agents: Dict[str, dict] {} self._lock threading.Lock() def register(self, agent_info: dict): agent_id agent_info[agent_id] with self._lock: self._agents[agent_id] agent_info print(f[registry] agent {agent_id} registered) def unregister(self, agent_id: str): with self._lock: self._agents.pop(agent_id, None) def match(self, skill: str) - Optional[dict]: with self._lock: for agent in self._agents.values(): if skill in agent.get(skills, []): return agent return None路由是调度层的重头戏。核心逻辑并不复杂根据 target_skills 从注册中心逐个找候选 Agent再调用执行器。真正要小心的是重试和超时这里有两个容易写错的地方。第一个是超时时间不能只算单次执行要把重试的累计时间也考虑进去。否则可能出现“总量 40 秒重试 3 次每次 15 秒”的情况外部调用方早就等疯了。第二个是重试必须判断异常类型不能无脑重试。我见过不少项目把“参数校验失败”也重试了白白浪费 Agent 资源。Agent-Reach 里我做了异常分类只有AgentBusyError、AgentTimeoutError这样的瞬时异常才触发重试业务逻辑错误直接返回失败不做无意义补偿。下面这个调度函数就是完整实现# dispatcher.py import time import threading from typing import Optional from dataclasses import dataclass import requests class AgentBusyError(Exception): pass class AgentTimeoutError(Exception): pass class BusinessLogicError(Exception): pass dataclass class DispatcherResult: success: bool result: Optional[dict] None error: Optional[str] None def call_agent(agent: dict, ticket: TaskTicket) - dict: url agent[endpoint] resp requests.post(url, jsonticket.payload, timeoutticket.timeout_ms / 1000) resp.raise_for_status() return resp.json() def dispatch(ticket: TaskTicket, registry: AgentRegistry) - DispatcherResult: for skill in ticket.target_skills: agent registry.match(skill) if not agent: continue ticket.agent_id agent[agent_id] max_retries ticket.retry_policy[max_retries] backoff_ms ticket.retry_policy[backoff_ms] for attempt in range(max_retries 1): try: result call_agent(agent, ticket) return DispatcherResult(successTrue, resultresult) except (AgentBusyError, AgentTimeoutError) as e: if attempt max_retries: time.sleep(backoff_ms / 1000 * (2 ** attempt)) continue return DispatcherResult(successFalse, errorstr(e)) except Exception as e: return DispatcherResult(successFalse, errorfbusiness error: {e}) return DispatcherResult(successFalse, errorno agent matched skills)实测这种写法非常稳。重试退避我用了指数退避1 秒、2 秒、4 秒这样避免重试风暴。像“同时收到 100 个任务某个 Agent 全部超时”的情况下固定退避反而会加剧拥堵指数退避能相对平缓下来。3.3 工具注册与权限校验Agent 要真正干活往往需要触达外部工具或数据源。Agent-Reach 里也管了这件事。每个工具在注册时都有清晰的访问权限定义Agent 想调用工具“调度层”会先检查当前 Agent 是否在白名单里。不夸张地讲这层校验救了整个项目。当时有个 Agent 本来只应该读市场数据结果因为可以直接调内部写接口发生了一次误操作。加上权限校验之后Agent 与工具的对应关系必须是路由层显式注册的再没出现过越权调用。# tool_registry.py class ToolRegistry: def __init__(self): self._tools {} self._allowlist {} def register_tool(self, tool_name: str, handler: callable, allowed_agents: list): self._tools[tool_name] handler self._allowlist[tool_name] set(allowed_agents) def invoke(self, tool_name: str, agent_id: str, *args, **kwargs): allowed self._allowlist.get(tool_name, set()) if agent_id not in allowed: raise PermissionError(fagent {agent_id} cannot access {tool_name}) return self._tools[tool_name](*args, **kwargs)建议工具的注册信息尽量和 Agent 注册信息一起维护。当时我们从只维护 Agent 白名单到后来把“Agent 能力工具权限数据源范围”全部集中到一个配置文件里维护成本反而更低了。因为排查问题只需要看一个地方而不是满项目地找权限配置。3.4 可观测性设计日志追踪、会话链路多 Agent 协作项目里最大的痛苦是“不知道任务到底走到哪了”。Agent-Reach 用了一套“链路日志”方案灵感来自分布式追踪里的 trace 概念但没有引入额外依赖就是约定日志格式。具体做法每次进入调度层就生成一个 TraceID整个任务链路中所有日志都带上这个 ID。例如[TRACEf8d9aa31-b2bd-47ba-9a8f-1f2be2c3d001] [AGENTretriever] [ACTIONinvoke_tool] toolmarket_data statusok cost_ms120 [TRACEf8d9aa31-b2bd-47ba-9a8f-1f2be2c3d001] [AGENTdoc_generator] [ACTIONgenerate] statusrunning cost_ms24000在日志系统里按 TraceID 一搜整条流程就像故事线一样展开。很多“看起来 Agent 在乱跑”的诡异现场其实都是因为链路日志不完整查清链路后多半都能定位到具体节点。我还做了一个很轻量的小页面直接读 Redis 里的票据状态按 ticket_id 展示“当前状态、被路由到哪个 Agent、总共耗了多少时间、失败原因在哪个环节”。这个页面大约 100 行代码但团队排查效率提升了非常多强烈建议你也搞一个。4. 常见问题与排查技巧实录4.1 Agent 互相踢皮球路由命中但执行失败最常见的问题调度层已经匹配到 Agent但 Agent 返回“这事我干不了”任务就这样断掉。原因通常是注册中心记录的 skills 太宽泛或者 Agent 内部的提示词对它真实能力理解偏了。排查路径分两步。第一步看路由日志里是否真的把任务发给了这个 Agent第二步看 Agent 返回的错误码是不是business error。如果是业务错误调度层不会重试这是合理行为关键在于注册中心里 abilities 的设置。我当时把检索 Agent 的 skills 从简单地写[market_data]细化成了[market_data, industry_report_2024]并且给每个 skill 增加了一个description字段做模糊匹配命中率立刻提升。这里有个小技巧Agent 注册信息里的 skills 描述得越“像人会说的话”路由匹配的准确率越高别用过于抽象的词。4.2 死循环与任务风暴如何加熔断和配额多 Agent 协作还有一个容易忽视的风险Agent A 调 Agent BAgent B 又反过来调 Agent A如果触发条件设置不当就变成互相调用直到资源耗尽。我们在 Agent-Reach 里做了两层保护。第一层是任务深度限制。TaskTicket 里加了一个depth字段每次进入新的子任务就depth 1超过限制比如 5 层直接拒绝。这个数字不是拍脑袋定的当时我们真实业务链路最长是“检索→分析→生成→校验”四层我留了一级余量压测下来 5 层最合理。第二层是按 Agent 的并发配额与熔断。注册中心的max_concurrency字段不只是个数字调度层会实时统计该 Agent 正在执行的任务数超过配额就返回AgentBusyError触发重试逻辑。连续失败次数超过阈值例如 10 次就把该 Agent 临时熔断不再路由新任务过 30 秒再自动恢复。这套组合拳上线后任务风暴的概率基本降到零。最明显的一次效果是外部网关误配了一个定时任务一下子上万个任务打进来系统只是正常拒绝了超配部分没有把执行层的 Agent 全部拖垮。4.3 上下文太长导致响应超时滑动窗口策略第二个高频问题是大语言模型 Agent 的上下文太长导致响应时间飙升最终触发超时。一开始我们简单地把timeout_ms调大结果一条链路整体 SLA 越来越差。后来我换了个思路与其被动调大超时不如主动控制上下文长度。在 Agent-Reach 的调度层里加了一个“上下文窗口”预处理逻辑任务票据里允许调用方指定window_policy如果历史上下文太长就先做摘要压缩剪掉不相关的参考片段再发给执行层。实测效果非常明显40 个长上下文的检索任务平均耗时从之前的 25 秒降到 11 秒超时率从 15% 降到 2% 以下。这里建议你将超时和压缩阈值联动起来看而不是各调各的。4.4 排查技巧速查表最后把我实际排障过程中常用的定位路径整理成一个速查表贴出来给大家备用现象优先排查点手段任务卡在 pending 很久注册中心有没有匹配到 Agent查路由日志的 match 结果Agent 被调用但无响应并发配额是否占满查看该 Agent 的实时并发数任务秒失败是否业务逻辑错误被误判为可重试异常看错误类型分类多个 Agent 循环调用任务深度是否超限看 TraceID 的父子链路调用方等不到结果HTTP 连接超时与任务超时是否匹配对比接入层 timeout 和票据 timeout工具越权调用工具白名单是否按 Agent 配置检查 ToolRegistry 的 allowlist这套速查表是那两个月最值钱的产出之一。新人排查问题时不用再瞎猜照着表格先定位再拿日志验证基本十分钟内能锁定问题。最后再分享一个调试技巧Agent-Reach 里我自己最满意的一个实践是本地调试时专门写了一个“MockAgent”。这个 MockAgent 不是真的调用大模型而是根据请求参数返回预设结果还能自定义延迟、随机故障、特定错误码。每次改完路由代码我先用 MockAgent 把“超时、忙、业务错误、正常返回”这四类情况都压一遍确认调度层的重试和熔断逻辑都符合预期再接真实生产 Agent 联调。你别小看这个步骤。很多看起来诡异的线上问题实际上在本地就能稳定复现——只要 MockAgent 能精确模拟成百上千次并发下的故障很多隐藏 bug 都会提前爆出来。这一套下来后面即使再出问题我心里也有数先跑一遍 MockAgent 场景能复现就先定位不能复现再怀疑外部环境排查效率至少翻一倍。Agent-Reach 这个项目整体不大但把“触达、路由、状态、权限、可观测”这件小事想透、做透之后我最大的感受是多 Agent 系统的复杂度从来不是模型本身而是它们之间的连接方式。如果你正被这类问题困扰不妨把今天这套任务票据 注册中心 分层调度的骨架抄回去试试大概率会有惊喜。