1. 项目概述Agent-Reach解决的是什么问题先说结论Agent-Reach是一个面向多智能体Multi-Agent场景的轻量级调度与触达框架核心解决的是“当你有一堆各自为政的Agent时如何让它们安全、有序、高效地协作”这件事。这两年LLM应用卷到现在单Agent已经不太够用了。一个像样的业务系统里往往需要多个Agent各管一摊有的负责理解用户意图有的负责查数据有的负责调用工具有的负责生成回复。问题随之而来——这些Agent之间怎么互相发现任务怎么路由上下文怎么传递状态怎么同步如果没有一个统一调度层Agent之间就是一团乱麻每个Agent都在重复造轮子而且很容易出现任务丢失、上下文错乱、互相踢皮球的情况。Agent-Reach就是在这样的背景下出现的。你可以把它理解成一个“Agent界的消息中心路由网关”每个Agent只需要把自己注册进来声明自己能干什么、需要什么、产出什么剩下的协作细节全部交给Reach层处理。调用方不需要关心目标Agent部署在哪、用什么协议通信只要说“我要做什么”Reach会帮你找到合适的Agent并把任务送过去。这套东西适合谁如果你是做LLM应用开发的工程师或者正在从单Agent迁移到多Agent架构再或者手上有多个Agent服务但找不到一个干净的协作方式Agent-Reach这套设计和代码可以直接抄作业。我下面会从设计思路、核心模块、完整搭建过程、路由策略、问题排查五个方面把整个项目拆开讲尽量做到拿过来就能用。2. 整体设计思路为什么先定协议再谈协作2.1 多Agent协作的真实痛点在讲Agent-Reach的架构之前先聊聊多Agent系统的几个常见死法这样你才知道为什么这套方案要这么设计。第一Agent之间没有统一的“语言”。有的Agent暴露的是HTTP接口有的是gRPC有的干脆就是个Python函数。你用的时候得挨个适配每接一个新Agent就要写一堆胶水代码。第二任务没有明确的归属。一个用户请求来了先让谁处理某个Agent搞不定怎么办超时怎么办完全没有机制基本上靠硬编码写死流程。第三上下文被割裂。多个Agent协作处理一个任务时每个Agent都只看到自己那一小段历史没有全局视角结果就是答非所问或者重复劳动。第四状态管理靠人肉。谁处理到哪一步了哪个Agent挂了需要重试全靠开发者自己心里记系统规模一大必然失控。Agent-Reach的设计思路就是针对这四类问题用一套协议层路由层上下文层状态层的四层结构来兜住。2.2 四层架构的设计取舍整个Agent-Reach分为四层层级职责解决的核心问题协议层统一Agent接入的接口规范Agent之间语言不通、接口各异路由层根据任务属性和Agent能力进行匹配分发任务不知道交给谁处理上下文层维护跨Agent的会话级上下文传播上下文被割裂、信息孤岛状态层记录任务生命周期和Agent健康状态任务状态不可见、失败无法恢复这四层看着简单每层的设计都踩过不少坑我拆开说。2.3 为什么选中心化调度而不是去中心化架构选型上我没有用去中心化的点对点方案比如让Agent之间互相直接通信而是选择了中心化的调度层。理由很简单在多Agent场景里中心化调度带来的可控性远远大于它引入的单点风险。点对点方案听起来很酷但实际用起来你会面临几个很现实的问题每个Agent都要维护一份全局服务发现列表新Agent上线要广播、下线要通知任务路径不可追踪出了问题很难复现还有最要命的——没有统一的兜底机制A Agent把任务丢给B AgentB处理不了又丢给C最终任务在哪个环节丢了没人知道。中心化调度把所有这些复杂度收敛到一个点上。Agent-Reach作为唯一的“任务入口”所有请求都经过它进行能力匹配、路由分发、状态记录。带来的好处很直接任务链路完全透明每个环节耗时、参数、结果都有据可查Agent之间不需要互相感知对方的存在只管接活干完活交回来就行。单点问题怎么解决Agent-Reach在设计上做了有状态的调度节点支持多实例部署共享存储Invoker模式也支持故障转移后面实操部分我会详细说。3. 核心模块拆解注册、路由、传播、状态3.1 Agent注册与能力声明RegistryAgent-Reach里最基础的概念是Agent注册表。任何Agent要接入第一步是在注册表里登记自己的基础信息和能力描述。一个标准的Agent注册信息包含以下字段{ agent_id: weather-bot, name: 天气查询助手, version: 1.2.0, capabilities: [weather.query, weather.forecast], input_schema: { city: {type: string, required: true}, date: {type: string, required: false} }, output_schema: { temperature: {type: number}, condition: {type: string} }, endpoint: agent://weather-bot/run, timeout_ms: 3000, priority: 10 }这个设计的核心思路是把Agent的“能力”收敛为语义化能力标签。任务侧描述“我要查天气”会被解析为weather.query这个能力标签路由器直接按标签匹配而不是靠大模型去理解Agent的描述文本。这个设计踩过一个坑一开始我试图用自然语言描述Agent能力比如“这个Agent可以查询天气也可以查未来几天的天气预报”让路由器用LLM去判断该给谁。后来发现这种方案有两个问题一是LLM判断有延迟一次路由要好几百毫秒二是判断结果不稳定同一句话两次解析可能分给不同Agent。后来改成schema化能力标签LLM辅助解析的方式用户请求先由一个轻量级的意图理解器解析成能力标签再去注册表里精确匹配。解析器偶尔判断不准的情况由降级策略兜底这部分后面细说。3.2 任务路由与匹配算法Router路由层是Agent-Reach的核心也是最值得细看的地方。我把路由设计成一个分阶段的过程第一步意图解析。调用方传入的自然语言请求或者结构化任务首先被解析为意图对象。结构化任务好办直接读字段自然语言请求需要走一次LLM调用输出格式固定为{intents: [weather.query], params: {...}}。第二步能力匹配。拿意图能力标签去注册表里做匹配。这里有个细节匹配不是简单的字符串相等我实现了简单的标签相似度匹配。比如任务侧说weather.today注册表里有weather.query和weather.forecast两个Agent都没有完全相等的标签这时候走标签别名表today别名指向query匹配成功。标签命名规范和别名表的维护是个长期工程。我的建议是每个新Agent注册时先查一遍现有标签列表能复用的绝不新建命名风格统一用“领域.动作”格式。这部分可以在后续通过维护一个JSON配置文件来做。第三步排序与选择。多个Agent都能处理同一个任务时按以下优先级排序标签匹配度精确匹配优于别名匹配→ Agent优先级priority字段→ 当前负载正在处理的请求数→ 历史成功率。最后加一个随机因子避免流量永远打给同一实例。这个排序逻辑看起来简单但实际价值非常大。之前我踩过一个坑没有按负载排序结果两个天气Agent一个忙死一个闲死忙的那个由于任务是同步阻塞模式新请求直接排长队整体P95延迟从200ms飙到了3s。加上负载排序之后链路平稳很多。第四步超时与降级。路由匹配不到任何Agent时触发降级链路尝试模糊匹配标签子串包含、同域匹配→ 尝试默认兜底Agent如果注册了的话→ 返回明确错误“当前没有Agent可以处理该请求”。这里有个设计原则我在项目里反复强调过宁可明确报错也不要乱找一个不合适的Agent硬接。有一次我把降级策略配置成了“匹配不到就随机挑一个Agent”结果查天气的任务被随机丢给了一个负责订机票的Agent返回了一堆航班信息作为回复用户直接懵了。后来我把随机降级改成“返回可理解错误”用户体验反而更好。3.3 跨Agent上下文传播Context Propagation多Agent协作最棘手的问题就是上下文。举个例子用户先问“北京今天天气怎么样”再问“那明天呢”如果第二个请求被路由给了另一个Agent、且它看不到前一轮的对话历史它根本不知道“那明天呢”指的是什么地方的明天。Agent-Reach的实现方案是维护一个会话级上下文存储。每个会话有一个全局上下文IDAgent在处理任务时可以读取当前上下文的全部信息、把自己的产出追加到上下文、标记哪些上下文片段是私有的不传给下游。具体数据结构长这样class AgentContext: def __init__(self, session_id): self.session_id session_id self.messages [] self.state {} self.metadata {} def append(self, role, content, agent_idNone): self.messages.append({ role: role, content: content, agent_id: agent_id, timestamp: time.time() }) def attach_state(self, key, value, scopeglobal): if scope global: self.state[key] value else: self.state.setdefault(f{scope}:{key}, value)上下文传播的模式我实现了两种全量传播把整个会话历史都传给下游Agent。适合需要全局理解的场景缺点是Token消耗大、有隐私泄露风险。摘要传播关键状态传播用LLM把上下文压缩成摘要再附上结构化状态信息。适合链路较长的场景省Token但可能丢细节。实际使用中我的建议是混合使用前三轮对话用全量传播超过一定轮数自动切换为摘要模式。这个阈值我在项目里默认配置的是8轮具体可以根据你的模型上下文窗口和业务需要调。3.4 任务状态机与可观测性State Observability没有状态管理的多Agent系统出了问题基本靠猜。Agent-Reach为每个任务维护了完整的状态机PENDING → ROUTING → DISPATCHED → RUNNING → SUCCEEDED ↘ TIMEOUT → RETRYING → SUCCEEDED ↘ FAILED → TERMINATED每个Agent处理完任务后回传结果状态机推进一步。所有状态变更都会写入日志附带的trace_id贯穿整个调用链。这个状态机最重要的价值是可观测性。我在项目里给每个任务生成了trace_id所有涉及该任务的日志、上下文变更、路由决策都会带上这个ID。排查问题时只需要拿到用户那边的trace_id就能完整还原整个链路。可观测性这块我额外加了两个细节一个是指标埋点记录每个Agent的平均耗时、成功率、P95延迟另一个是链路可视化面板按trace_id聚合展示每个环节的耗时分布。这两个功能在项目真正上线后帮了大忙有好几次线上问题就是靠链路面板一眼定位到是某个Agent响应慢导致的级联超时。4. 从0到1搭建Agent-Reach4.1 环境准备与依赖安装Agent-Reach核心是一个Python包使用FastAPI提供调度服务通过Redis做状态存储和上下文缓存。之所以选Python是因为生态里跟LLM相关的SDK最齐全选型成本最低。依赖分两部分。核心运行依赖pip install fastapi uvicorn redis httpx pydantic可选的Agent SDK依赖——如果你接的是LangChain生态的Agent还需要装langchain库。我项目里用的是OpenAI的函数调用风格Agent所以额外装了openai。安装完依赖之后快速创建一个最小可运行的调度服务# main.py from agent_reach import ReachServer, AgentRegistry, Router app ReachServer( registryAgentRegistry(redis_urlredis://localhost:6379/0), routerRouter(redis_urlredis://localhost:6379/0) ) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8787)如果你不想装Redis项目也支持纯内存模式构造AgentRegistry和Router时不传redis_url即可。内存模式适合本地调试线上跑建议还是用Redis因为要支持多实例调度。4.2 注册你的第一个Agent搭建好调度服务后第一步是往注册表里登记Agent。我建议把Agent的注册信息做成独立配置文件这样Agent上线时可以走配置文件交付而不是手动调用注册接口。项目里我用的方式是声明式注册文件# agents/weather_agent.yaml agent_id: weather-bot name: 天气查询助手 version: 1.2.0 capabilities: - weather.query - weather.forecast endpoint: http://127.0.0.1:9001/run timeout_ms: 3000 priority: 10然后加载注册python -m agent_reach.cli register --file agents/weather_agent.yaml配置文件方式的好处是Agent的接入信息可以走版本管理回滚、审计都方便。注册完成后可以通过python -m agent_reach.cli list查看当前所有在线Agent。4.3 实现Agent的运行接口每个Agent只需要实现一个统一接口接收任务请求返回执行结果。协议采用HTTP POST请求体和响应体都是JSON。拿天气Agent举例实现起来长这样# weather_agent.py from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class TaskRequest(BaseModel): task_id: str intent: str params: dict context: dict class TaskResponse(BaseModel): task_id: str result: dict status: str success app.post(/run) def run(request: TaskRequest): city request.params.get(city, 北京) date request.params.get(date, 今天) # 这里是实际调用天气API的逻辑 weather_data query_weather(city, date) return TaskResponse( task_idrequest.task_id, resultweather_data )这里有个实践心得Agent接口的参数一定要用Schema约束好不要接受任意字段。一开始我为了省事让Agent接口的params设计成dict类型随便传结果路由层传错了参数类型下游Agent崩溃了还不知道是上游的问题。后来所有Agent接入必须声明input_schema和output_schema路由层在分发前做一次参数校验传错直接拦截。4.4 发送第一个跨Agent任务调度服务跑起来、Agent也注册好了下一步就是发送任务。import requests response requests.post( http://127.0.0.1:8787/v1/tasks, json{ session_id: session-001, request: 帮我查一下北京明天的天气, metadata: {source: playground} } ) task_id response.json()[task_id] print(task_id)发送任务后Agent-Reach会完成意图解析、能力匹配、路由分发、结果回传。任务可能是异步的你可以通过轮询或者Webhook拿结果。import time while True: result requests.get(fhttp://127.0.0.1:8787/v1/tasks/{task_id}) status result.json()[status] if status in (SUCCEEDED, FAILED): print(result.json()) break time.sleep(0.5)这个轮询模式在调试阶段完全够用。生产阶段建议走Webhook回调的方式避免大量轮询请求打到调度服务上。5. 路由策略内核调度算法与安全细节5.1 意图解析的工程实现细节前面简单提了意图解析的流程这里展开讲讲工程细节。我实现了一个轻量级的意图解析器流程分为两步def parse_intent(request: str, llm_client, registry) - Intent: # 第一步用LLM把自然语言解析为结构化意图 prompt f 你是意图解析器。请将用户请求解析为能力标签列表。 可选能力标签如下 {registry.list_capabilities()} 用户请求{request} 输出格式JSON {{intents: [标签1, 标签2], params: {{}}}} llm_output llm_client.complete(prompt) intent parse_json_llm_output(llm_output) # 第二步标签规范化 normalized_intents [registry.normalize_label(i) for i in intent[intents]] return intent这里有个很实际的工程注意点LLM输出的JSON格式不一定合法。有的模型会多输出几行解释文字有的会在JSON外面加反引号包裹。项目里我实现了一个宽容的JSON解析器先试json.loads失败则用正则提取JSON块再失败就调用LLM重试一次。这看起来是个小地方但实际运行中这是意图解析器报错的大头来源之一。另外一个设计选择是意图解析不是每次请求都要走LLM。我加了一层缓存对相同或相似的请求文本缓存解析结果。实现上用简单的编辑距离做相似度判断比如两次请求的文本编辑距离小于5直接复用上一次的解析结果。能省不少API调用。5.2 路由匹配的顺序与权重路由匹配的执行顺序是我花了不少精力调整的部分。最初版本只有一个简单的规则精确匹配标签找不到就报错。后来实际跑起来发现完全不够用用户的一句话经常解析出多个意图标签每个标签又不是都能精确匹配。现在的路由匹配完整流程是第一优先精确匹配。任务意图标签完全等于Agent能力标签这是最可靠的情况。第二优先别名匹配。通过标签别名表把任务意图转换为已有能力标签。比如任务意图是weather.today别名表里today→query于是匹配到weather.query。第三优先模糊匹配。基于标签的领域前缀匹配。比如weather.current没有Agent认领但weather.*有Agent注册可以尝试路由过去。这个策略我默认是关闭的因为它有误路由风险需要业务方确认后再打开。第四优先默认Agent。注册表里可以设置一个或一组默认兜底Agent处理所有无法匹配的请求。这个我建议是接一个专门负责“澄清”的Agent它不负责实际执行任务而是生成追问回复让用户把需求说得更具体。每条路由决策都会打日志记录匹配过程和最终选择。调试路由问题时候看路由日志比看代码靠谱得多。5.3 失败重试与超时熔断不管路由做得再好线上总会遇到下游Agent挂掉或者响应超时的情况。Agent-Reach内置了失败重试和超时熔断机制。配置项如下# reach_config.yaml retry_policy: max_retries: 2 # 最大重试次数 backoff_base_ms: 500 # 退避基数 retry_on_status: [500, 502, 503, 504] # 可重试的HTTP状态码 retry_on_timeout: true circuit_breaker: failure_threshold: 5 # 连续失败5次触发熔断 open_timeout_ms: 30000 # 熔断持续时间 half_open_max_calls: 2 # 半开状态下允许试探的请求数这里我踩过的坑值得单独说一下不要对所有失败都无脑重试。HTTP 400这种参数错误重试一万次也是失败只会放大下游压力HTTP 429说明是限流重试时应该加更大的退避时间。之前我把retry_on_status配置成了空的结果默认全部重试下游服务本来就雪崩了还被重试流量反复捶打。后来明确配置只重试5xx和超时情况立刻好转。熔断机制的触发条件是连续失败次数超过阈值。熔断打开后后续请求直接快速失败不进入路由分发等熔断时间过了进入半开状态放少量请求试探下游成功则关闭熔断失败则重新打开。这个机制没有大篇幅代码但逻辑设计对系统稳定性的提升是非常显著的。5.4 安全设计上下文的权限管控多Agent协作时上下文数据是共享的但不同Agent能读到的信息应该不一样。项目里做了一层上下文权限控制。实现方式是给上下文状态打上scope标记Agent注册时可以声明自己可访问的scopeclass AgentContext: def __init__(self, session_id, agent_access_scopeNone): self.session_id session_id self.state {} self.agent_access_scope agent_access_scope or set() def read_state(self, key, agent_id): scope, _, name key.partition(:) if scope and scope not in self.agent_access_scope: return None return self.state.get(key)每个Agent在上下文中写入数据时声明这个数据的可见范围。比如用户画像类的数据只允许特定的Agent读取执行记录类的数据全局可读。这样可以防止敏感信息在整个Agent链路上无节制传播。6. 常见问题与排查技巧实录6.1 Agent注册后路由不到这是最频繁遇到的问题。Agent明明注册成功了list命令也能看到但发任务就是报“无匹配Agent”。排查思路按顺序来第一步确认能力标签一致。注册的是weather.query任务解析出来的意图是weather.today标签不匹配自然路由不到。用python -m agent_reach.cli capabilities看一下当前注册表中所有Agent的能力标签对比意图解析日志。第二步确认Agent存活状态。注册表里Agent的状态字段是否显示为active。Agent下线时没有妥善处理注销注册表里可能残留inactive状态的记录匹配时会直接跳过。第三步确认Agent的endpoint可以访问。从调度服务所在机器直接curl Agent的endpoint确认网络通。这个坑经常发生在Agent和调度服务不在同一网段端口被防火墙挡住的情况。第四步查看路由日志。Agent-Reach会记录完整路由决策日志包括每一步匹配的结果。实测下来90%的路由问题在第一步就能定位。6.2 跨Agent上下文丢失现象是任务链路里某个Agent拿到的是空上下文或者只有部分历史信息。这个问题的常见原因有几种。一种是上下文存储的key设计不合理——我用Redis存上下文时最初key只用了session_id结果不同任务复用同一个会话时出现了上下文错位。后来改成session_id task_id做key保证每个任务看到的是该任务启动时的上下文快照。另一种是摘要模式丢数据。上下文切换成摘要模式后某些Agent需要的关键状态没有写进摘要导致后续Agent拿到不完整信息。解决方式是让Agent显式声明哪些状态字段是必须传播的摘要生成时强制包含这些字段。还有一种隐蔽的情况上下文里嵌套了不可序列化的对象。某个Agent把Python对象直接塞进了context.state序列化存Redis时抛异常但当时异常被吞掉了后续读取时只能拿到None。排查方式是用context dump命令打印完整上下文内容一眼就能看出异常。6.3 LLM解析器返回结构不稳定意图解析器依赖LLM输出JSON但LLM经常不按格式输出。我遇到过的怪样子包括JSON里多个逗号、中文字符串没有引号、在JSON前后输出解释性文字、把JSON包在Markdown代码块里。现在的处理策略是分三层def parse_json_llm_output(text: str) - dict: # 第一层直接解析 try: return json.loads(text) except Exception: pass # 第二层提取大括号内容 match re.search(r\{.*\}, text, re.DOTALL) if match: try: return json.loads(match.group()) except Exception: pass # 第三层修复常见JSON语法错误 fixed fix_json_syntax(text) if fixed: return json.loads(fixed) raise ValueError(f无法解析LLM输出: {text})第三层的fix_json_syntax我实现了几个常用修复去掉多余逗号、给裸关键字加引号、去掉非法转义。这些修复覆盖了90%的脏数据场景。如果三层都解析失败直接重试一次LLM调用并提示“请严格按照指定JSON格式输出”。优化之后解析成功率从大约86%提高到了98%以上剩下不到2%的失败走重试基本没有因为解析失败导致任务中断的情况了。6.4 多实例部署时的状态一致性问题调度服务单实例部署的好处是简单坏处是可用性不够。升级到多实例后最先遇到的问题就是同一个会话的请求被分发到不同实例每个实例读到的是Redis里同一份上下文数据但Agent状态信息保存在各自内存里导致A实例认为天气Agent还活着B实例认为天气Agent已经超时挂了。这个问题的解决思路是状态信息也存Redis用分布式锁控制状态变更。Agent注册、心跳上报、熔断开关、任务状态更新都走Redis的原子操作。调度实例本身变成无状态水平扩展节点任意一个实例挂了流量切到其他实例不影响正确性。从单实例到多实例的改造我的经验是先保住两条命一是状态存储从内存切到Redis二是Agent心跳上报机制建立起来。这两条做完多实例部署基本就稳了。7. 实操心得与后续扩展方向项目从设计到稳定运行我最大的体会是多Agent协作系统的复杂度不在单个Agent的实现而在Agent之间的“边界”上。协议没定好、路由决策不透明、上下文边界模糊这些问题在Agent数量少的时候完全看不出来一旦规模上来就集中爆发。对于准备做类似系统的朋友我的建议是先想清楚三层边界任务的边界什么算一个完整任务、能力的边界哪些事情应该由同一个Agent负责、上下文的边界哪些信息可以被哪些Agent看到。这三层设计想透了代码反而是最简单的一步。Agent-Reach后续可以扩展的方向我觉得最值得做的是基于反馈的动态路由优化。目前的路由依赖注册时的静态优先级和历史成功率但任务实际执行成功与否、用户对结果是否满意这些信息还没有回流到路由决策里。如果能收集这些反馈信号动态调整路由权重调度系统会越用越聪明。另一个方向是支持更丰富的工作流编排目前每个任务是单跳路由下一步可以支持多跳一个Agent的输出作为另一个Agent的输入形成Agent处理链这样能覆盖更复杂的业务场景。最后分享一个实际操作中的小技巧给所有Agent的响应统一加一个agent_id字段。无论是中间结果还是最终结果标明是哪个Agent产出的。听起来是很小的一件事但排查问题时这个字段能省你几个小时的时间尤其是链路里经过了三四个Agent的情况。系统上线初期就把这个习惯养好后面会少很多不必要的痛苦。 
长河君 · 2026/10/29 9:28:10 · 阅读 1.2k Agent-Reach：多智能体协作的轻量级调度与路由框架
1. 项目概述Agent-Reach解决的是什么问题先说结论Agent-Reach是一个面向多智能体Multi-Agent场景的轻量级调度与触达框架核心解决的是“当你有一堆各自为政的Agent时如何让它们安全、有序、高效地协作”这件事。这两年LLM应用卷到现在单Agent已经不太够用了。一个像样的业务系统里往往需要多个Agent各管一摊有的负责理解用户意图有的负责查数据有的负责调用工具有的负责生成回复。问题随之而来——这些Agent之间怎么互相发现任务怎么路由上下文怎么传递状态怎么同步如果没有一个统一调度层Agent之间就是一团乱麻每个Agent都在重复造轮子而且很容易出现任务丢失、上下文错乱、互相踢皮球的情况。Agent-Reach就是在这样的背景下出现的。你可以把它理解成一个“Agent界的消息中心路由网关”每个Agent只需要把自己注册进来声明自己能干什么、需要什么、产出什么剩下的协作细节全部交给Reach层处理。调用方不需要关心目标Agent部署在哪、用什么协议通信只要说“我要做什么”Reach会帮你找到合适的Agent并把任务送过去。这套东西适合谁如果你是做LLM应用开发的工程师或者正在从单Agent迁移到多Agent架构再或者手上有多个Agent服务但找不到一个干净的协作方式Agent-Reach这套设计和代码可以直接抄作业。我下面会从设计思路、核心模块、完整搭建过程、路由策略、问题排查五个方面把整个项目拆开讲尽量做到拿过来就能用。2. 整体设计思路为什么先定协议再谈协作2.1 多Agent协作的真实痛点在讲Agent-Reach的架构之前先聊聊多Agent系统的几个常见死法这样你才知道为什么这套方案要这么设计。第一Agent之间没有统一的“语言”。有的Agent暴露的是HTTP接口有的是gRPC有的干脆就是个Python函数。你用的时候得挨个适配每接一个新Agent就要写一堆胶水代码。第二任务没有明确的归属。一个用户请求来了先让谁处理某个Agent搞不定怎么办超时怎么办完全没有机制基本上靠硬编码写死流程。第三上下文被割裂。多个Agent协作处理一个任务时每个Agent都只看到自己那一小段历史没有全局视角结果就是答非所问或者重复劳动。第四状态管理靠人肉。谁处理到哪一步了哪个Agent挂了需要重试全靠开发者自己心里记系统规模一大必然失控。Agent-Reach的设计思路就是针对这四类问题用一套协议层路由层上下文层状态层的四层结构来兜住。2.2 四层架构的设计取舍整个Agent-Reach分为四层层级职责解决的核心问题协议层统一Agent接入的接口规范Agent之间语言不通、接口各异路由层根据任务属性和Agent能力进行匹配分发任务不知道交给谁处理上下文层维护跨Agent的会话级上下文传播上下文被割裂、信息孤岛状态层记录任务生命周期和Agent健康状态任务状态不可见、失败无法恢复这四层看着简单每层的设计都踩过不少坑我拆开说。2.3 为什么选中心化调度而不是去中心化架构选型上我没有用去中心化的点对点方案比如让Agent之间互相直接通信而是选择了中心化的调度层。理由很简单在多Agent场景里中心化调度带来的可控性远远大于它引入的单点风险。点对点方案听起来很酷但实际用起来你会面临几个很现实的问题每个Agent都要维护一份全局服务发现列表新Agent上线要广播、下线要通知任务路径不可追踪出了问题很难复现还有最要命的——没有统一的兜底机制A Agent把任务丢给B AgentB处理不了又丢给C最终任务在哪个环节丢了没人知道。中心化调度把所有这些复杂度收敛到一个点上。Agent-Reach作为唯一的“任务入口”所有请求都经过它进行能力匹配、路由分发、状态记录。带来的好处很直接任务链路完全透明每个环节耗时、参数、结果都有据可查Agent之间不需要互相感知对方的存在只管接活干完活交回来就行。单点问题怎么解决Agent-Reach在设计上做了有状态的调度节点支持多实例部署共享存储Invoker模式也支持故障转移后面实操部分我会详细说。3. 核心模块拆解注册、路由、传播、状态3.1 Agent注册与能力声明RegistryAgent-Reach里最基础的概念是Agent注册表。任何Agent要接入第一步是在注册表里登记自己的基础信息和能力描述。一个标准的Agent注册信息包含以下字段{ agent_id: weather-bot, name: 天气查询助手, version: 1.2.0, capabilities: [weather.query, weather.forecast], input_schema: { city: {type: string, required: true}, date: {type: string, required: false} }, output_schema: { temperature: {type: number}, condition: {type: string} }, endpoint: agent://weather-bot/run, timeout_ms: 3000, priority: 10 }这个设计的核心思路是把Agent的“能力”收敛为语义化能力标签。任务侧描述“我要查天气”会被解析为weather.query这个能力标签路由器直接按标签匹配而不是靠大模型去理解Agent的描述文本。这个设计踩过一个坑一开始我试图用自然语言描述Agent能力比如“这个Agent可以查询天气也可以查未来几天的天气预报”让路由器用LLM去判断该给谁。后来发现这种方案有两个问题一是LLM判断有延迟一次路由要好几百毫秒二是判断结果不稳定同一句话两次解析可能分给不同Agent。后来改成schema化能力标签LLM辅助解析的方式用户请求先由一个轻量级的意图理解器解析成能力标签再去注册表里精确匹配。解析器偶尔判断不准的情况由降级策略兜底这部分后面细说。3.2 任务路由与匹配算法Router路由层是Agent-Reach的核心也是最值得细看的地方。我把路由设计成一个分阶段的过程第一步意图解析。调用方传入的自然语言请求或者结构化任务首先被解析为意图对象。结构化任务好办直接读字段自然语言请求需要走一次LLM调用输出格式固定为{intents: [weather.query], params: {...}}。第二步能力匹配。拿意图能力标签去注册表里做匹配。这里有个细节匹配不是简单的字符串相等我实现了简单的标签相似度匹配。比如任务侧说weather.today注册表里有weather.query和weather.forecast两个Agent都没有完全相等的标签这时候走标签别名表today别名指向query匹配成功。标签命名规范和别名表的维护是个长期工程。我的建议是每个新Agent注册时先查一遍现有标签列表能复用的绝不新建命名风格统一用“领域.动作”格式。这部分可以在后续通过维护一个JSON配置文件来做。第三步排序与选择。多个Agent都能处理同一个任务时按以下优先级排序标签匹配度精确匹配优于别名匹配→ Agent优先级priority字段→ 当前负载正在处理的请求数→ 历史成功率。最后加一个随机因子避免流量永远打给同一实例。这个排序逻辑看起来简单但实际价值非常大。之前我踩过一个坑没有按负载排序结果两个天气Agent一个忙死一个闲死忙的那个由于任务是同步阻塞模式新请求直接排长队整体P95延迟从200ms飙到了3s。加上负载排序之后链路平稳很多。第四步超时与降级。路由匹配不到任何Agent时触发降级链路尝试模糊匹配标签子串包含、同域匹配→ 尝试默认兜底Agent如果注册了的话→ 返回明确错误“当前没有Agent可以处理该请求”。这里有个设计原则我在项目里反复强调过宁可明确报错也不要乱找一个不合适的Agent硬接。有一次我把降级策略配置成了“匹配不到就随机挑一个Agent”结果查天气的任务被随机丢给了一个负责订机票的Agent返回了一堆航班信息作为回复用户直接懵了。后来我把随机降级改成“返回可理解错误”用户体验反而更好。3.3 跨Agent上下文传播Context Propagation多Agent协作最棘手的问题就是上下文。举个例子用户先问“北京今天天气怎么样”再问“那明天呢”如果第二个请求被路由给了另一个Agent、且它看不到前一轮的对话历史它根本不知道“那明天呢”指的是什么地方的明天。Agent-Reach的实现方案是维护一个会话级上下文存储。每个会话有一个全局上下文IDAgent在处理任务时可以读取当前上下文的全部信息、把自己的产出追加到上下文、标记哪些上下文片段是私有的不传给下游。具体数据结构长这样class AgentContext: def __init__(self, session_id): self.session_id session_id self.messages [] self.state {} self.metadata {} def append(self, role, content, agent_idNone): self.messages.append({ role: role, content: content, agent_id: agent_id, timestamp: time.time() }) def attach_state(self, key, value, scopeglobal): if scope global: self.state[key] value else: self.state.setdefault(f{scope}:{key}, value)上下文传播的模式我实现了两种全量传播把整个会话历史都传给下游Agent。适合需要全局理解的场景缺点是Token消耗大、有隐私泄露风险。摘要传播关键状态传播用LLM把上下文压缩成摘要再附上结构化状态信息。适合链路较长的场景省Token但可能丢细节。实际使用中我的建议是混合使用前三轮对话用全量传播超过一定轮数自动切换为摘要模式。这个阈值我在项目里默认配置的是8轮具体可以根据你的模型上下文窗口和业务需要调。3.4 任务状态机与可观测性State Observability没有状态管理的多Agent系统出了问题基本靠猜。Agent-Reach为每个任务维护了完整的状态机PENDING → ROUTING → DISPATCHED → RUNNING → SUCCEEDED ↘ TIMEOUT → RETRYING → SUCCEEDED ↘ FAILED → TERMINATED每个Agent处理完任务后回传结果状态机推进一步。所有状态变更都会写入日志附带的trace_id贯穿整个调用链。这个状态机最重要的价值是可观测性。我在项目里给每个任务生成了trace_id所有涉及该任务的日志、上下文变更、路由决策都会带上这个ID。排查问题时只需要拿到用户那边的trace_id就能完整还原整个链路。可观测性这块我额外加了两个细节一个是指标埋点记录每个Agent的平均耗时、成功率、P95延迟另一个是链路可视化面板按trace_id聚合展示每个环节的耗时分布。这两个功能在项目真正上线后帮了大忙有好几次线上问题就是靠链路面板一眼定位到是某个Agent响应慢导致的级联超时。4. 从0到1搭建Agent-Reach4.1 环境准备与依赖安装Agent-Reach核心是一个Python包使用FastAPI提供调度服务通过Redis做状态存储和上下文缓存。之所以选Python是因为生态里跟LLM相关的SDK最齐全选型成本最低。依赖分两部分。核心运行依赖pip install fastapi uvicorn redis httpx pydantic可选的Agent SDK依赖——如果你接的是LangChain生态的Agent还需要装langchain库。我项目里用的是OpenAI的函数调用风格Agent所以额外装了openai。安装完依赖之后快速创建一个最小可运行的调度服务# main.py from agent_reach import ReachServer, AgentRegistry, Router app ReachServer( registryAgentRegistry(redis_urlredis://localhost:6379/0), routerRouter(redis_urlredis://localhost:6379/0) ) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8787)如果你不想装Redis项目也支持纯内存模式构造AgentRegistry和Router时不传redis_url即可。内存模式适合本地调试线上跑建议还是用Redis因为要支持多实例调度。4.2 注册你的第一个Agent搭建好调度服务后第一步是往注册表里登记Agent。我建议把Agent的注册信息做成独立配置文件这样Agent上线时可以走配置文件交付而不是手动调用注册接口。项目里我用的方式是声明式注册文件# agents/weather_agent.yaml agent_id: weather-bot name: 天气查询助手 version: 1.2.0 capabilities: - weather.query - weather.forecast endpoint: http://127.0.0.1:9001/run timeout_ms: 3000 priority: 10然后加载注册python -m agent_reach.cli register --file agents/weather_agent.yaml配置文件方式的好处是Agent的接入信息可以走版本管理回滚、审计都方便。注册完成后可以通过python -m agent_reach.cli list查看当前所有在线Agent。4.3 实现Agent的运行接口每个Agent只需要实现一个统一接口接收任务请求返回执行结果。协议采用HTTP POST请求体和响应体都是JSON。拿天气Agent举例实现起来长这样# weather_agent.py from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class TaskRequest(BaseModel): task_id: str intent: str params: dict context: dict class TaskResponse(BaseModel): task_id: str result: dict status: str success app.post(/run) def run(request: TaskRequest): city request.params.get(city, 北京) date request.params.get(date, 今天) # 这里是实际调用天气API的逻辑 weather_data query_weather(city, date) return TaskResponse( task_idrequest.task_id, resultweather_data )这里有个实践心得Agent接口的参数一定要用Schema约束好不要接受任意字段。一开始我为了省事让Agent接口的params设计成dict类型随便传结果路由层传错了参数类型下游Agent崩溃了还不知道是上游的问题。后来所有Agent接入必须声明input_schema和output_schema路由层在分发前做一次参数校验传错直接拦截。4.4 发送第一个跨Agent任务调度服务跑起来、Agent也注册好了下一步就是发送任务。import requests response requests.post( http://127.0.0.1:8787/v1/tasks, json{ session_id: session-001, request: 帮我查一下北京明天的天气, metadata: {source: playground} } ) task_id response.json()[task_id] print(task_id)发送任务后Agent-Reach会完成意图解析、能力匹配、路由分发、结果回传。任务可能是异步的你可以通过轮询或者Webhook拿结果。import time while True: result requests.get(fhttp://127.0.0.1:8787/v1/tasks/{task_id}) status result.json()[status] if status in (SUCCEEDED, FAILED): print(result.json()) break time.sleep(0.5)这个轮询模式在调试阶段完全够用。生产阶段建议走Webhook回调的方式避免大量轮询请求打到调度服务上。5. 路由策略内核调度算法与安全细节5.1 意图解析的工程实现细节前面简单提了意图解析的流程这里展开讲讲工程细节。我实现了一个轻量级的意图解析器流程分为两步def parse_intent(request: str, llm_client, registry) - Intent: # 第一步用LLM把自然语言解析为结构化意图 prompt f 你是意图解析器。请将用户请求解析为能力标签列表。 可选能力标签如下 {registry.list_capabilities()} 用户请求{request} 输出格式JSON {{intents: [标签1, 标签2], params: {{}}}} llm_output llm_client.complete(prompt) intent parse_json_llm_output(llm_output) # 第二步标签规范化 normalized_intents [registry.normalize_label(i) for i in intent[intents]] return intent这里有个很实际的工程注意点LLM输出的JSON格式不一定合法。有的模型会多输出几行解释文字有的会在JSON外面加反引号包裹。项目里我实现了一个宽容的JSON解析器先试json.loads失败则用正则提取JSON块再失败就调用LLM重试一次。这看起来是个小地方但实际运行中这是意图解析器报错的大头来源之一。另外一个设计选择是意图解析不是每次请求都要走LLM。我加了一层缓存对相同或相似的请求文本缓存解析结果。实现上用简单的编辑距离做相似度判断比如两次请求的文本编辑距离小于5直接复用上一次的解析结果。能省不少API调用。5.2 路由匹配的顺序与权重路由匹配的执行顺序是我花了不少精力调整的部分。最初版本只有一个简单的规则精确匹配标签找不到就报错。后来实际跑起来发现完全不够用用户的一句话经常解析出多个意图标签每个标签又不是都能精确匹配。现在的路由匹配完整流程是第一优先精确匹配。任务意图标签完全等于Agent能力标签这是最可靠的情况。第二优先别名匹配。通过标签别名表把任务意图转换为已有能力标签。比如任务意图是weather.today别名表里today→query于是匹配到weather.query。第三优先模糊匹配。基于标签的领域前缀匹配。比如weather.current没有Agent认领但weather.*有Agent注册可以尝试路由过去。这个策略我默认是关闭的因为它有误路由风险需要业务方确认后再打开。第四优先默认Agent。注册表里可以设置一个或一组默认兜底Agent处理所有无法匹配的请求。这个我建议是接一个专门负责“澄清”的Agent它不负责实际执行任务而是生成追问回复让用户把需求说得更具体。每条路由决策都会打日志记录匹配过程和最终选择。调试路由问题时候看路由日志比看代码靠谱得多。5.3 失败重试与超时熔断不管路由做得再好线上总会遇到下游Agent挂掉或者响应超时的情况。Agent-Reach内置了失败重试和超时熔断机制。配置项如下# reach_config.yaml retry_policy: max_retries: 2 # 最大重试次数 backoff_base_ms: 500 # 退避基数 retry_on_status: [500, 502, 503, 504] # 可重试的HTTP状态码 retry_on_timeout: true circuit_breaker: failure_threshold: 5 # 连续失败5次触发熔断 open_timeout_ms: 30000 # 熔断持续时间 half_open_max_calls: 2 # 半开状态下允许试探的请求数这里我踩过的坑值得单独说一下不要对所有失败都无脑重试。HTTP 400这种参数错误重试一万次也是失败只会放大下游压力HTTP 429说明是限流重试时应该加更大的退避时间。之前我把retry_on_status配置成了空的结果默认全部重试下游服务本来就雪崩了还被重试流量反复捶打。后来明确配置只重试5xx和超时情况立刻好转。熔断机制的触发条件是连续失败次数超过阈值。熔断打开后后续请求直接快速失败不进入路由分发等熔断时间过了进入半开状态放少量请求试探下游成功则关闭熔断失败则重新打开。这个机制没有大篇幅代码但逻辑设计对系统稳定性的提升是非常显著的。5.4 安全设计上下文的权限管控多Agent协作时上下文数据是共享的但不同Agent能读到的信息应该不一样。项目里做了一层上下文权限控制。实现方式是给上下文状态打上scope标记Agent注册时可以声明自己可访问的scopeclass AgentContext: def __init__(self, session_id, agent_access_scopeNone): self.session_id session_id self.state {} self.agent_access_scope agent_access_scope or set() def read_state(self, key, agent_id): scope, _, name key.partition(:) if scope and scope not in self.agent_access_scope: return None return self.state.get(key)每个Agent在上下文中写入数据时声明这个数据的可见范围。比如用户画像类的数据只允许特定的Agent读取执行记录类的数据全局可读。这样可以防止敏感信息在整个Agent链路上无节制传播。6. 常见问题与排查技巧实录6.1 Agent注册后路由不到这是最频繁遇到的问题。Agent明明注册成功了list命令也能看到但发任务就是报“无匹配Agent”。排查思路按顺序来第一步确认能力标签一致。注册的是weather.query任务解析出来的意图是weather.today标签不匹配自然路由不到。用python -m agent_reach.cli capabilities看一下当前注册表中所有Agent的能力标签对比意图解析日志。第二步确认Agent存活状态。注册表里Agent的状态字段是否显示为active。Agent下线时没有妥善处理注销注册表里可能残留inactive状态的记录匹配时会直接跳过。第三步确认Agent的endpoint可以访问。从调度服务所在机器直接curl Agent的endpoint确认网络通。这个坑经常发生在Agent和调度服务不在同一网段端口被防火墙挡住的情况。第四步查看路由日志。Agent-Reach会记录完整路由决策日志包括每一步匹配的结果。实测下来90%的路由问题在第一步就能定位。6.2 跨Agent上下文丢失现象是任务链路里某个Agent拿到的是空上下文或者只有部分历史信息。这个问题的常见原因有几种。一种是上下文存储的key设计不合理——我用Redis存上下文时最初key只用了session_id结果不同任务复用同一个会话时出现了上下文错位。后来改成session_id task_id做key保证每个任务看到的是该任务启动时的上下文快照。另一种是摘要模式丢数据。上下文切换成摘要模式后某些Agent需要的关键状态没有写进摘要导致后续Agent拿到不完整信息。解决方式是让Agent显式声明哪些状态字段是必须传播的摘要生成时强制包含这些字段。还有一种隐蔽的情况上下文里嵌套了不可序列化的对象。某个Agent把Python对象直接塞进了context.state序列化存Redis时抛异常但当时异常被吞掉了后续读取时只能拿到None。排查方式是用context dump命令打印完整上下文内容一眼就能看出异常。6.3 LLM解析器返回结构不稳定意图解析器依赖LLM输出JSON但LLM经常不按格式输出。我遇到过的怪样子包括JSON里多个逗号、中文字符串没有引号、在JSON前后输出解释性文字、把JSON包在Markdown代码块里。现在的处理策略是分三层def parse_json_llm_output(text: str) - dict: # 第一层直接解析 try: return json.loads(text) except Exception: pass # 第二层提取大括号内容 match re.search(r\{.*\}, text, re.DOTALL) if match: try: return json.loads(match.group()) except Exception: pass # 第三层修复常见JSON语法错误 fixed fix_json_syntax(text) if fixed: return json.loads(fixed) raise ValueError(f无法解析LLM输出: {text})第三层的fix_json_syntax我实现了几个常用修复去掉多余逗号、给裸关键字加引号、去掉非法转义。这些修复覆盖了90%的脏数据场景。如果三层都解析失败直接重试一次LLM调用并提示“请严格按照指定JSON格式输出”。优化之后解析成功率从大约86%提高到了98%以上剩下不到2%的失败走重试基本没有因为解析失败导致任务中断的情况了。6.4 多实例部署时的状态一致性问题调度服务单实例部署的好处是简单坏处是可用性不够。升级到多实例后最先遇到的问题就是同一个会话的请求被分发到不同实例每个实例读到的是Redis里同一份上下文数据但Agent状态信息保存在各自内存里导致A实例认为天气Agent还活着B实例认为天气Agent已经超时挂了。这个问题的解决思路是状态信息也存Redis用分布式锁控制状态变更。Agent注册、心跳上报、熔断开关、任务状态更新都走Redis的原子操作。调度实例本身变成无状态水平扩展节点任意一个实例挂了流量切到其他实例不影响正确性。从单实例到多实例的改造我的经验是先保住两条命一是状态存储从内存切到Redis二是Agent心跳上报机制建立起来。这两条做完多实例部署基本就稳了。7. 实操心得与后续扩展方向项目从设计到稳定运行我最大的体会是多Agent协作系统的复杂度不在单个Agent的实现而在Agent之间的“边界”上。协议没定好、路由决策不透明、上下文边界模糊这些问题在Agent数量少的时候完全看不出来一旦规模上来就集中爆发。对于准备做类似系统的朋友我的建议是先想清楚三层边界任务的边界什么算一个完整任务、能力的边界哪些事情应该由同一个Agent负责、上下文的边界哪些信息可以被哪些Agent看到。这三层设计想透了代码反而是最简单的一步。Agent-Reach后续可以扩展的方向我觉得最值得做的是基于反馈的动态路由优化。目前的路由依赖注册时的静态优先级和历史成功率但任务实际执行成功与否、用户对结果是否满意这些信息还没有回流到路由决策里。如果能收集这些反馈信号动态调整路由权重调度系统会越用越聪明。另一个方向是支持更丰富的工作流编排目前每个任务是单跳路由下一步可以支持多跳一个Agent的输出作为另一个Agent的输入形成Agent处理链这样能覆盖更复杂的业务场景。最后分享一个实际操作中的小技巧给所有Agent的响应统一加一个agent_id字段。无论是中间结果还是最终结果标明是哪个Agent产出的。听起来是很小的一件事但排查问题时这个字段能省你几个小时的时间尤其是链路里经过了三四个Agent的情况。系统上线初期就把这个习惯养好后面会少很多不必要的痛苦。