Agent-Reach 这名字听起来像是一个网络工具但它其实是我最近两个月一直在折腾的一个 AI 智能体调度项目的代号。如果你也在做多智能体系统或者正在为“Agent 总是接不到该接的活、做不对该做的事”而头疼那这篇复盘应该能给你一些可落地的参考思路。简单说Agent-Reach 要做的事情是在一个多智能体协作系统里让每一个任务都能被最合适的 Agent 稳定触达、正确响应、高效完成。它既是一个调度框架也是一套触达策略的工程化实现。我之所以把它写出来是因为这个过程中踩了不少坑也总结了一些并非教科书上会写的经验。项目本身不大但麻雀虽小五脏俱全从意图解析到任务路由从重试策略到上下文管理基本都过了一遍。这篇文章就按我实际的开发顺序把设计思路、核心代码逻辑、上线后的排查记录和迭代方向完整拆开希望能帮到正在做 Agent 调度或想入门的同学。1. 先从业务痛点讲起为什么要做 Agent-Reach1.1 多智能体协作里的触达困境最近大半年我很长时间都在和各种 Agent 框架打交道。刚接触的时候最大的感受是单个 Agent 做点垂直任务挺容易但一旦把三五个 Agent 放一起协作立刻会陷入“消息满天飞活没人干”的混乱局面。举一个真实场景。我在做一个智能客服系统的时候原本只需要一个 Agent 来回答商品咨询问题。后来业务方加了需求要能自动识别用户情绪、判断是否需要人工介入、还要在特定场景下发起订单催付。于是我把系统拆成了三个 Agent一个负责意图识别一个负责情绪识别一个负责任务执行。问题马上就来了——用户说了一句“这个衣服洗了会不会缩水”意图识别 Agent 觉得是售前咨询情绪识别 Agent 判断有退货倾向任务执行 Agent 又不知道到底该调售前的知识库还是售后的退换货策略。三个 Agent 都在响应却没有一个真正“触达”了问题的本质。这就是我所说的“触达困境”消息确实发给了所有 Agent但语义上没有到达那个“该管这件事”的 Agent。有时候是意图分类有歧义有时候是多个 Agent 的职责边界重叠还有些时候是执行 Agent 因为上下文信息不完整根本不敢拍板。结果就是整个系统表现得像个巨慢且混乱的会议而不是一个高效协作的团队。1.2 Agent-Reach 的定位名字里藏着两层意思“触达困境”的本质其实是两个“Reach”的问题第一层是Task Reach也就是任务能不能被准确路由到正确的 Agent第二层是Data Reach也就是执行任务的 Agent 有没有获得足够的信息来完成它。所以我把项目命名为 Agent-Reach就是想把这两个问题一起解决。Task Reach 靠调度器Data Reach 靠上下文管理两者结合才是完整的触达闭环。一开始我也天真地以为只要在大模型提示词里把“你是负责售后的 Agent”说清楚就行后来实践发现“指派”只是最短的那块木板真正决定系统能在多智能体协作中走多远的是后面那一整套策略组合。这个项目最终实现的效果是在模拟客服场景的 200 个测试用例里任务被错误路由的比例从一开始的 37% 降到了 6% 以内同一个任务被重复执行的比例从 19% 降到了接近 0整体任务完成耗时缩短了约三分之一。虽然样本不算大但改进趋势非常明显给我的感觉是这套思路是站得住脚的。2. 整体架构与核心设计思路2.1 五层架构从任务进入到底层反馈Agent-Reach 不是我一开始就设计成型的而是先搭了个粗糙版本跑了一周之后我按照实际出问题的地方重新梳理才调整成了下面这个五层结构。第一层任务接入层。这一层负责接收外部请求。不管是 Webhook、消息队列还是手工调用进来之后统一封装成标准格式的任务对象。任务对象里至少包含三个字段task_id、raw_input、priority。这一步很重要因为后续所有调度逻辑都依赖一个统一的“语言”。第二层意图解析层。这一层不只是简单做一个分类而是要把用户输入分解成结构化信息。我用了两段式处理先用一个轻量的分类模型或者大模型把输入归纳为几个粗粒度意图类别再结合预设的实体提取器把关键对象比如订单号、商品ID、用户ID提取出来。这样做的好处是调度器不需要在原始文本上做语义判断只需要看结构化的意图和实体。第三层调度决策层。这是 Agent-Reach 的核心也就是前面说的 Task Reach。它的输入是意图解析层产生的结构化任务包输出是一个“Agent 选择加策略选择”的决策结果。调度器内部有一个评分函数能对每个可用的 Agent 计算一个“适合度”然后决定谁最有资格触达这个任务。第四层触达执行层。决定由哪个 Agent 干活之后这一层负责把任务真正推到执行者面前。对应的是 Data Reach 的落地。它要做的事情包括组装该 Agent 需要的上下文、拼接动态提示词、调用模型或服务接口、监控执行状态。如果执行失败它会根据预设策略触发重试或者降级。第五层反馈回填层。这是很容易被忽视的一层但恰恰决定了系统能不能越用越准。每次任务执行完不管成功失败都会把结果、调度决策、执行耗时、上下文使用量等信息记录到反馈存储中。用这些数据我可以事后分析调度评分函数的参数是否需要调整也能生成训练数据去做更精细的微调。这五层各有各的职责我实际开发时的经验是前三层要严格解耦第四层要尽量“傻”第五层要尽量“全”。换句话说决策逻辑越集中越好执行层越简单越好日志记录越详细越好。这个原则帮我少踩了很多坑。2.2 调度器不是规则引擎权重评分的逻辑最早我也试图用一堆 if-else 来做路由比如“意图是A就调用Agent A意图是B就调用Agent B”。但很快发现真实场景里的意图往往不是非黑即白的。用户说一句“我想退货但是还想再看看别的款式”这里面既有退货意图又有继续浏览意图规则引擎根本处理不了这种叠加情况。所以我换成了权重评分方式。每个 Agent 注册进入系统的时候不再只是简单声明“我处理退货”而是声明一组“能力特征”比如capability_intents该 Agent 能处理的意图列表以及每个意图的擅长度0-1capability_entities该 Agent 能处理哪些实体类型比如订单实体、商品实体required_context该 Agent 执行任务必须依赖哪些上下文比如必须要有订单号才能查物流constraints该 Agent 的禁入条件比如不处理售后问题调度器在给每个 Agent 评分的时候会拿任务包里的结构化信息和这些能力特征做匹配。评分函数我简化成三个维度的加权和def compute_score(task, agent): # 1. 意图匹配度 intent_score 0.0 for intent, confidence in task.intents: if intent in agent.capability_intents: intent_score agent.capability_intents[intent] * confidence # 2. 实体匹配度 entity_score 0.0 matched_entities 0 for entity in task.entities: if entity.type in agent.capability_entities: matched_entities 1 entity_score matched_entities / max(len(task.entities), 1) # 3. 上下文就绪度 context_score sum([1 for ctx in agent.required_context if ctx in task.provided_context]) context_score context_score / max(len(agent.required_context), 1) # 加权求和权重可配置 final_score 0.5 * intent_score 0.3 * entity_score 0.2 * context_score return final_score权重是 0.5、0.3、0.2一开始是我拍脑袋定的后来通过反馈回填层的数据反复调了几轮最后发现这个比例在两难场景中表现比较稳。为什么意图匹配占大头因为任务触达的第一原则是先找对“懂行的”再看“有没有基础条件”。如果你把一个需要售后知识的问题分配给一个只会售前的 Agent就算上下文给得再足它也只能一本正经地胡说。这里有几个实际细节值得聊一下。处理实体匹配的时候我一开始对“正确”的 Agent 定义太严格导致一个任务同时没有 Agent 得分高于阈值就会落入兜底系统。后来我加了“模糊匹配下的容忍度”实体类型完全一致时得 1 分属于同一大类时得 0.5 分。比如“订单实体”和“售后订单实体”在大多数场景下其实是能由同一个 Agent 处理的0.5 的过渡分比 0 分要合理得多。还有一个细节是每个 Agent 的能力特征不能写死需要通过配置接口动态更新。我在上线后遇到过两次业务调整一次是把“退款”从售后 Agent 划给了财务 Agent一次是新增了一个“评价管理” Agent如果能力特征是硬编码的这种调整就会很难受。2.3 触达策略引擎重试、退避与语义降级Task Reach 解决了“任务该给谁”的问题但还没解决“任务给出去之后怎么保证被执行成功”的问题。触达执行层就是干这个的。我最开始觉得调用 Agent 就跟调用普通 API 一样发出去等返回就行结果发现完全不是这么回事。大模型接口的失败模式比普通 API 要复杂得多有时候是超时有时候是返回内容不符合格式要求有时候是模型“抠字眼”——比如它明明分配到了任务却因为上下文里某句话的引导开始质疑自己的角色拒绝执行。这种失败无法通过简单重试解决因为重试同样的内容大概率还会失败。我的做法是在触达执行层引入一个策略引擎它包含三个机制。第一个机制是指数退避重试。这个逻辑比较简单第一次失败等 2 秒再试第二次失败等 4 秒最多重试 3 次。每次重试时把上次失败的反馈比如“上次返回内容 JSON 解析失败”拼进提示词里这样模型有机会自己修正。第二个机制是降级路径。如果同一个 Agent 连续重试两次都失败说明这个 Agent 在当前状态下可能根本搞不定这个任务策略引擎就会启动降级把任务转移到备用 Agent或者用预设的模版化逻辑兜底。比如智能客服场景里如果 Agent 三次都无法生成合规回复就切换到人工工单流程而不是让用户无限等待。第三个机制是部分成功。这个想法来自于我自己一次很惨痛的经历有一次任务列表里有 5 个子任务执行 Agent 完成了前两个然后接口超时。如果不做任何处理重试时会从头开始前两个子任务就会被重复执行造成数据重复。所以我在执行层里给每个子任务都加了单独的状态标记重试的时候只补做未完成的部分。# 子任务状态标记示意 subtask.status # pending / running / success / failed / skipped这套策略引擎看着不难但确实是把任务触达成功率稳定在了 95% 以上的关键所在。模型本身的能力再强没有一个健壮的策略包裹层也无法在生产环境里可靠运行。3. 关键模块实操从 0 到 1 搭建 Agent-Reach3.1 技术选型与基础环境Agent-Reach 的技术栈不算新潮我用的是 Python 3.10 FastAPI 做服务层Redis 做任务队列缓存SQLite 的 JSON 字段做反馈存储Agent 的执行部分调用的是 OpenAI 兼容的 API 接口。之所以这样选主要是三个原因第一OpenAI 兼容接口是当下最通用的协议不管底层用的是哪家大模型只要提供 base_url 和 api_key就能无缝切换。这让我在开发阶段避免被某个特定平台锁死。第二Redis 做任务队列非常直观。任务进来之后LPUSH 进列表消费端 BRPOP 出来处理。配合 Redis 的原子操作天然支持多实例部署。我在本地开发时甚至可以不开 Redis直接用一个简单的 Python 队列类替换数据结构和接口保持一致就行。第三SQLite 的 JSON 字段对于我这种偏好轻量开发的个人项目来说足够了。我只需要把完整的任务快照和调度决策快照存进一个字段后面做分析时用 pandas 读取 JSON 再展开就行比搞一个完整的 PostgreSQL 环境省了太多事。一下是项目的基础目录结构我尽量把“干什么的”写在文件名里减少沟通成本agent_reach/ ├── api/ # FastAPI 路由层 │ └── main.py ├── core/ # 核心逻辑 │ ├── scheduler.py # 调度评分 │ ├── strategy.py # 重试/降级策略 │ ├── context_mgr.py # 上下文组装与管理 │ └── feedback.py # 反馈回填 ├── agents/ # 具体 Agent 实现 │ ├── base.py │ ├── pre_sales.py │ ├── after_sales.py │ └── finance_agent.py ├── store/ # 存储层 │ └── db.py └── config/ └── agents.yaml # Agent 能力特征配置3.2 任务拆解与意图解析的落地写法任务拆解是整个管线的敲门砖如果这一层做得粗糙后面调度器的输入就会很稀碎。我用的方式是一次调用完成“分类 实体提取”而不是分两步。这条经验来自我最初的失败教训当时我把意图分类和实体提取分成了两次独立的大模型调用结果在 100 个任务上测试实体提取的准确率下降了 12%——因为第二次调用时上下文延续性变差模型已经忘了第一次分类的语义重点了。后来我改成了一次提示词请求同时输出 JSON 结构def parse_task(raw_input: str) - TaskPackage: prompt f 请解析用户输入输出 JSON。 需要包含: - intents: 列表每个元素包含 intent 和 confidence - entities: 列表每个元素包含 type 和 value 用户输入: {raw_input} 只输出 JSON不要额外说明。 resp call_llm(prompt, response_formatjson) return TaskPackage(**resp)这里有个重要的小细节意图可能有多个不要只取置信度最高的那一个。像“我想退货但还是想看看别的款式”在 intents 里应该包含两个意图只是置信度不同。调度器评分的时候会乘以置信度这样既能识别出主要意图也能保留辅助信息供后续 Agent 判断使用。实体提取也是同样的道理尽量遍历输入中的所有候选对象。比如用户说“订单 20240531 的商品还没到”实体不仅要提取“订单号 20240531”也应该提取“商品这个泛指对象”因为如果后续调度针对的是售后订单号就够了但是如果用户还想催物流物流 Agent 需要的其实是“商品”和“订单号”两个实体。我在实际使用中还发现解析结果需要加一层“合法性校验”。大模型输出的 JSON 偶尔会少一个字段或者把 bool 写成 string。这一步不校验的话错误会在调度器里被放大。我写了一个非常简单的validate_task_package函数检查必填字段是否都存在、类型是否正确不对就触发一次单点重试解析。3.3 调度评分函数的实现细节调度器的核心我已经在前面展示了简化版。但有几个实现细节值得展开因为它们直接影响生产环境的行为。第一个细节是阈值处理。每个 Agent 的评分都需要和一个可配置的阈值对比只有高于阈值才可能被选中。这个阈值不能全局统一而是每个 Agent 单独配置。原因是不同 Agent 的擅长度设定本来就不同有的 Agent 擅长度普遍偏高有的偏低统一阈值会导致某些 Agent 永远抢占任务而另一些永远接不到任务。第二个细节是冲突决策。当两个 Agent 评分非常接近的时候比如分数差小于 0.05硬选一个不是好策略。这种情况通常意味着任务本身有歧义。我的处理方式是记录一次ambiguous_route并让两个候选 Agent 里评分较高的那个执行但在提示词里附加一句“这是一个跨领域任务你有权限调用另一个 Agent 的输出结果作为辅助信息”。这样就给 Agent 留了一个获得外部信息的接口而不是在语义边界模糊时硬砍一刀。第三个细节是反馈机制校准。调度器不是一次调好就万事大吉的。我在 feedback 表里会记录每次调度决策和最终执行结果。每周跑一次分析脚本统计每个 Agent 在哪些意图上的“选中率”和“成功率”出现了偏离。比如某个 Agent 在订单查询意图上选中率有 80%但成功率只有 40%就说明它的能力特征里擅长度设置得太高了需要下调。这个校准过程比重新训练模型成本低得多而且在数据积累足够的情况下效果很显著。3.4 触达执行与反馈回填的闭环触达执行层实际调用 Agent 时最核心的是上下文组装。这个环节如果没做好Agent 就像是拿到了任务书但没带地图一样会迷路。我在context_mgr.py里实现了一个函数专门负责给被选中的 Agent 构建上下文。它要做的事情有三件第一从任务包里提取关键片段第二从历史会话存储里取出与该用户或该实体相关的最近 N 条记录第三把这些信息按照“当前任务描述在前历史参考在后系统规则最后”的顺序拼装成提示词。def build_context(task, agent, history): sections [] # 当前任务 sections.append(f## 当前任务\n{task.raw_input}) # 结构化补充 sections.append(f## 结构化信息\n{task.to_json()}) # 历史上下文 if history: sections.append(f## 历史参考\n{history}) # Agent 的职责描述 sections.append(f## 你的职责\n{agent.role_desc}) return \n\n.join(sections)这里总被问到一个问题为什么任务描述放最前面而不把职责描述放最前面我测试过两种顺序效果差异比较大。把任务放最前面会让模型先把注意力集中在“要解决什么问题”上职责描述放在后面作为约束条件不容易破坏它本来的推理节奏。如果反过来模型容易把职责背诵得太大声却忘记了对任务的理解。尤其是多 Agent 场景里一个 Agent 直视自己的角色描述时很容易陷入角色固化的表演状态在一个特定框架里打转不愿意接受任务里稍微越界的数据。反馈回填则是在 Agent 返回结果之后做的最后一道工序。我创建了一个标准化的日志格式每次执行记录{ task_id: xxx, task_raw_input: 用户原文, parsed_intents: [...], parsed_entities: [...], selected_agent: after_sales, score_detail: {...}, execution_status: success, execution_output: ..., cost: 0.023, latency_ms: 1840 }这些数据才是 Agent-Reach 真正长期的财富。没有这些反馈调度器永远只是“看着合理的猜测”有了反馈调度器才能变成“越用越准的尺子”。4. 上线后的真实问题与排查记录4.1 幂等性问题同一个任务被触达了三次上线第一周遇到最离谱的问题是一个订单催付任务被重复触达了三次给用户发了三条一模一样的提醒。排查下来的原因有三层第一层是我的重试机制没有区分“请求失败”和“执行失败”第二层是子任务状态标记没有持久化进程重启后丢失了第三层是前端用户在等待响应时手动刷新触发了二次请求。解决方式我给执行层加了一个任务幂等令牌。任务在接入层生成task_id的时候同时生成一个idempotency_key。在执行层真正调用 Agent 之前先往 Redis 里写入一个对应task_id的记录设置一个合理的过期时间。执行完之后更新状态。这样即使同一个任务被并发传入多次也只有第一个请求能拿到执行锁。这里的一个选择是幂等检查的粒度不是整个任务而是子任务。因为一个任务可能包含多个子任务如果第一个子任务已经完成了第二个子任务因为超时重试不应该从头再来。所以幂等键需要放在子任务级别比如{task_id}:{subtask_index}。4.2 上下文窗口溢出与记忆压缩上下文窗口溢出基本是所有 Agent 类项目的必经之坑。Agent-Reach 里有一个 Agent 会持续对话历史记录累积得很快50 轮之后拼接的提示词直接超过了模型上下文限制报 400 错误。一开始我简单地把超出的部分裁掉结果裁掉之后 Agent 完全忘了用户最开始的需求回复质量直线下降。后来我采用了一个记忆压缩策略每次对话结束后用一次轻量的模型调用把当前会话压缩成一段 200 字以内的摘要存到记忆区。在后续构建上下文时历史参考部分用压缩摘要而不是原始对话。def compress_history(history: list[dict]) - str: prompt f请把以下对话压缩为不超过200字的摘要保留关键事实、用户意图和未完成事项。\n{history} return call_llm(prompt, max_tokens300)压缩策略上线后我发现有一个附带的好处Agent 的执行速度变快了因为没有那么多历史信息混杂在提示词里模型处理输入的开销变小了。这也从侧面说明上下文并不是越长越好真正有效的是结构和摘要而不是无脑堆量。4.3 提示词漂移Agent 越来越“不听指挥”另外一个比较隐蔽的问题是提示词漂移。具体表现是同一个 Agent 上线了一段时间后在同一个任务上的表现慢慢发生偏差最初的回答规规矩矩按照设定的格式来后来开始自由发挥格式开始放飞到 Json 外面去甚至自己加很多与职责无关的补充说明。排查后发现部分原因出在反馈回填的信息被拼进了提示词。我在上下文构建的时候把用户历史消息偶尔也带到了 Agent 的提示词里而历史消息里的提问方式和语气会反向影响模型后续的输出风格。这个机制模型原封不动地学下来了。解决方式很简单也很反直觉在提示词里增加一句明确的输出格式指令而且放在“你的职责”之后。这比放在开头管用得多因为我猜测模型对紧邻输出位置的指令记忆更牢固。我后来把这句话固定为“所有回复必须是 JSON 对象不允许输出 Markdown、解释或额外文本。”这句话虽然粗鲁但确实治好了格式漂移。4.4 性能与成本平衡并发限制与预算控制最后是性能和成本。Agent-Reach 上一开始是每个任务进来就同步去调用大模型接口结果单机时间全部被 LLM 延迟吃掉了。一个任务全链路耗时经常在 3-5 秒之间用户体验非常差。我改成了异步流水线任务接入后立刻返回task_id后台由 worker 池异步处理。worker 数量根据大模型接口的限流阈值来定。如果你用的是付费 API建议在代码里维护一个简单的并发信号量把并发数控制在限流阈值的 80% 左右留出缓冲。成本控制上我加了一个简单的预算检查每次调用前计算预计 token 数乘以单价累加到当日消费计数器。如果超过设定值就不再触发新的 LLM 调用而是走降级路径。这个逻辑很简单但在项目早期真的避免了我好几次费用失控。class BudgetGuard: def __init__(self, daily_limit): self.daily_limit daily_limit self.used 0 def check(self, estimated_cost): if self.used estimated_cost self.daily_limit: return False self.used estimated_cost return True5. 迭代方向与我的个人体会Agent-Reach 目前还在持续迭代中。我自己下一步想做的有三件事第一是让调度器从“基于配置的评分系统”逐步过渡到基于反馈数据的强化学习版本也就是把compute_score的权重参数交给数据来决定第二是把上下文压缩从固定 200 字改成自适应长度针对不同 Agent 的复杂度和任务类型动态调整第三是把整个系统容器化做成一个可以一条命令部署的版本方便在更多场景里复用。说回整体体会。这几个月实际做下来我最大的感受是Agent 系统的瓶颈往往不在模型能力而在工程结构。模型再强如果任务路由混乱、触达没有策略、上下文管理粗放输出依然是不可用的垃圾。Agent-Reach 本质上就是把“让合适的 Agent 做合适的事并且稳定地做完”这件事工程化了。它不算什么惊天动地的设计都是些老老实实的架构和策略但在真实业务里的改善是肉眼可见的。如果你也在搭建多智能体系统我的建议是先别急着上什么新框架把调度和触达这两个基本功做好整个系统会立刻不一样。