多彩编程 多彩编程MZPH · CODE BLOG
ARTICLE DETAIL

文章详情

深耕前端与后端开发技术的一线实战笔记与踩坑复盘。

Agent开发实战:动态路由与自适应编排的落地指南

Agent开发实战:动态路由与自适应编排的落地指南 做了这么久Agent开发我越来越觉得只要场景稍微复杂一点最头疼的根本不是模型能力而是“下一步该走哪条分支”。早期我习惯把流程全写死在代码里if-else一层套一层或者靠一个巨大的prompt让模型自己“悟”结果一上线就是各种离谱走位该调工具的时候不调该查记忆的时候乱猜多Agent协作时任务在子Agent之间踢皮球。后来我逐步把路由和编排从业务代码里拆出来整条链路的稳定性和可维护性完全不是一个量级。这篇是Agent系列的第9.3篇核心就聊两件事动态路由dynamic routing和自适应编排adaptive orchestration。前者解决的是“当前位置该往哪走”后者解决的是“整体流程怎么根据实际执行情况自我调整”。这两个词听起来玄落地其实是一套可以量化的决策机制和循环控制逻辑。我会用完整的代码示例、路由表设计方案、以及实际踩坑记录来讲清楚适合已经能跑通单个Agent、正准备往多工具、多Agent协作方向深挖的开发者。1. 内容整体设计与思路拆解1.1 为什么静态编排走不远先说说我为什么开始正儿八经研究动态路由。早期做Agent流程编排最直接的做法就是把流程画成一张固定图先做意图识别然后走固定的步骤A再走步骤B最后收尾。听起来没问题但真实业务场景里用户的输入是有大量随机性的。同样一句“帮我查一下上个季度的销售数据”有时候需要直接查数据库有时候需要先读一份报表文件有时候需要先反问用户“你指的是哪个区域的销售数据”。这些分支如果全用if-else去枚举你会发现代码里全是“屎山分支”而且需求一变整个流程图都要推倒重来。静态编排还有一个隐藏问题模型的实际行为和预设流程经常对不上。你设定好先调用工具A再调用工具B但模型在真实执行时可能觉得工具B的信息足够回答了根本不需要走A。这时候如果你的代码强制要求“必须按顺序执行”就会浪费一次推理甚至因为多了一次冗余调用把结果搞乱。动态路由的核心思路恰恰相反不预设完整路径而是只定义好“每个决策点有哪些可选分支以及如何选择”让模型在执行过程中实时决定下一步。1.2 动态路由到底在决策什么把“路由”这个词拆开看它在Agent系统里通常负责三类决策一类是工具路由。系统里有多个工具时当前这一步应该调用哪个。比如同时有sql_query和web_search两个工具用户问“库里有没有这个用户”就该走sql_query问“这个新闻怎么回事”就该走web_search。二类是记忆路由。当前任务是否需要先读取长期记忆或者需要写入新的记忆。很多Agent框架里记忆模块不是无脑加载的否则上下文塞满一堆无关历史反而干扰生成。三类是任务路由。在多Agent场景下更为明显当前这个子任务该由哪个专职Agent处理是代码Agent、数据分析Agent还是通用对话Agent。这三类决策有一个共同点它们都不需要模型重新“从头想一遍”而是让模型在有限的候选集合里做选择。这就把开放式的推理问题变成了一个轻量级的分类或匹配问题。落实到工程上动态路由的本质就是“把决策显式化”——你不再依赖模型的自由发挥而是给模型一个清晰的决策框架让它输出结构化结果再由代码去执行真正的分支跳转。1.3 自适应编排解决的另一个维度动态路由管的是“单步走向”自适应编排管的是“整条路的形状”。我曾做过一个需求Agent要完成一个跨数据源分析任务需要先查A库再根据结果决定是否查B库最后生成报告。如果写死“A→B→报告”那当A库查出来结果为空时B库压根没必要查。自适应编排的做法是让循环体不断检查“当前是否满足继续条件”不满足就换路径、补工具、甚至把任务拆小。编排器和路由器的关系我用一个类比来理解路由器是立交桥上的每个出口指示牌编排器是整体交通调度系统。指示牌决定“这个路口往哪拐”调度系统决定“什么时候让车流走、哪条路堵了就分流、某条路彻底断了怎么绕行”。两者结合起来整个Agent才具备应对真实环境不确定性的能力。2. 动态路由的核心细节与实现要点2.1 路由的前置意图识别到底怎么做动态路由的第一步永远是理解用户想干什么也就是意图识别。很多项目在这里栽跟头是因为他们一上来就上大模型分类把意图识别做得又重又慢。我自己的经验是按场景分三级方案不要一套逻辑打天下。最低成本的做法是关键词规则路由。比如用户消息里出现“查一下”“数据库”“有多少”这类词可以优先路由到数据查询类分支。这个方案适合候选意图少、关键词区分度高的内部工具场景。实测下来响应速度在毫秒级而且完全可控缺点是遇到说法换汤不换药的情况就抓瞎了。更通用的是向量相似度路由。把所有候选意图预先写成一段话比如“用户想查询数据库中的某个数据记录”存入向量库。用户输入也向量化算余弦相似度取Top1决定走哪条分支。这个方案不依赖大模型速度快且支持模糊表达但它要求候选意图之间的语义差异足够大否则相似度区分不明显。最灵活的是LLM结构化分类路由。让模型从候选路线列表中选一个并输出JSON。我给个实际用的prompt骨架你是一个路由决策器。根据用户的输入从以下候选中选择最合适的一条路线 1. database_query用户需要查询结构化数据 2. document_search用户需要查找文档内容 3. web_search用户需要查找实时信息 4. clarification用户意图不明确需要追问 只输出JSON{route: 路线标识, reason: 一句话理由}这个方案的优点是天生的开放域用户怎么说都能映射到有限集合里而且可以要求模型输出reason用于后续日志分析。代价是每一次路由都消耗一次模型调用。我实际项目中的折中做法是先跑规则路由规则命中不了再跑LLM路由双保险。2.2 路由表设计把所有可选路径变成数据动态路由能落地的关键是把“路由规则”从代码里搬出来变成一张可配置的路由表。我在项目里通常用Python字典或数据库表来定义结构大致是ROUTES [ { route_id: database_query, description: 查询数据库表适合用户询问具体记录、统计数据, handler: handle_database_query, keywords: [查询, 数据库, 记录, 统计], priority: 10, fallback: web_search }, { route_id: web_search, description: 搜索互联网获取实时信息适合新闻、时效内容, handler: handle_web_search, keywords: [新闻, 最新, 热点, 搜索], priority: 5, fallback: clarification }, { route_id: clarification, description: 用户意图不够明确需要追问补充信息, handler: handle_clarification, keywords: [], priority: 0, fallback: None } ]这个路由表有几个设计要点。第一每一条路由都有明确的description这是给LLM分类用的候选语义描述写得好不好直接决定分类准确率。第二priority标记优先级规则命中时可以按权重叠加比如“查询”和“数据库”两个词都命中时database_query的得分就更高。第三fallback定义了走不通时的降级路线保证系统永远不会无路可走。路由决策函数长这样def route_request(user_input: str) - RouteResult: # 阶段一规则匹配 scores {route[route_id]: 0 for route in ROUTES} for route in ROUTES: for kw in route[keywords]: if kw in user_input: scores[route[route_id]] route[priority] best_route max(scores, keyscores.get) if scores[best_route] 0: return build_result(best_route) # 阶段二规则没命中走LLM分类 chosen llm_route_decision(user_input, ROUTES) return build_result(chosen)这套设计的核心收益是大部分高频、明确的请求都在毫秒级完成路由只有模糊请求才会消耗额外的模型调用。在实际项目里规则命中率大约能覆盖60%~70%的流量整体路由成本被控制得很低。更重要的是路由表是纯数据新增一条分支只需要加一行配置不需要改动路由引擎代码。2.3 工具路由让模型在行动空间内做选择很多人问我为什么我调工具经常“调错”我排查下来十有八九不是模型笨而是工具列表的语义描述写得太糊。工具路由本质上同样是决策问题但候选集是“工具”决策依据是“工具描述”。一个被我反复验证的经验是工具描述必须包含三个信息——这个工具是干什么的、什么情况下用、什么情况下不用。再看一个反面例子某工具最初的描述是查询用户信息参数为user_id这种描述喂给模型它根本不知道什么时候该用经常把普通对话问题路由到这个工具上。改成了下面的描述后准确率明显提升根据用户ID查询用户的注册信息。仅当用户明确要求查询具体某个用户的资料时使用。如果用户是在闲聊或提问一般性问题不要使用此工具。这里还有一个小细节如果工具多了把全部工具描述塞进上下文token消耗和决策干扰都会上升。我建议工具数量超过8个时先做一层粗粒度的“工具分类路由”。比如有一批数据分析工具先根据用户输入决策“是否需要数据分析”需要时再往子路由里展开而不是一次性把20个工具全部暴露给模型。这有点类似人脑先判断学科再判断具体知识点分层决策的准确度和效率都远高于一次性决策。2.4 路由层的安全边界动态路由带来灵活性的同时也引入了一个安全风险模型可能因为被误导或幻觉把请求路由到不该走的路上。比如用户说“把这张表删了”如果工作流里恰好有drop_table工具模型直接路由过去后果不堪设想。我在路由层强制加了两个机制。第一高危操作白名单审批对删除、修改、发送消息、执行外部请求这类有副作用的路由不在路由决策层直接放行而是强制进入人工确认分支打印操作预览让用户确认。第二路由结果审计每次路由决策的route_id、reason、触发条件、命中关键词全部落日志。事后一旦发现某条路由频繁误判可以回看reason定位是描述问题还是规则冲突。不要小看这两步Agent系统上线之后安全不是靠运气而是靠边界控制。3. 自适应编排的实操与关键环节实现3.1 编排器的工作循环说完路由再来看编排。自适应编排最经典的落地形态是一个循环执行器用一段代码就能说清楚核心逻辑def adaptive_execute(task: str, max_steps: int 10): state { task: task, context: [], step_count: 0, finished: False, result: None } while not state[finished] and state[step_count] max_steps: state[step_count] 1 # 1. 思考当前状态决定下一步动作 decision agent_think(state[task], state[context]) if decision[action] finish: state[finished] True state[result] decision[answer] break # 2. 路由决定调用哪个工具或委派哪个子Agent route route_request(decision[query]) # 3. 执行并观察结果 observation execute_route(route, decision[arguments]) # 4. 把观察结果写回上下文影响下一轮决策 state[context].append({ action: route.route_id, args: decision[arguments], observation: observation }) # 5. 自适应调整如果连续多轮没有进展切换策略 if detect_no_progress(state[context]): state[context].append({system: switch_strategy}) return state[result]这个循环体之所以能实现“自适应”核心在两点。一是context是动态累积的每一轮的执行结果都会影响下一轮的路由决策相当于系统在根据实际反馈调整自己的路径。二是detect_no_progress这类探针函数它可以在循环里检测“是不是在原地打转”从而触发策略切换。3.2 用ReAct范式理解自适应调整如果对一个基础Agent框架有了解会发现这个循环和ReActReasoning Acting范式一脉相承。模型先输出Thought该做什么再输出Action怎么路由拿到Observation环境反馈后继续循环。自适应编排本质上是给ReAct循环加了三个增强件。第一个增强是路由约束。经典的ReAct里Action是完全开放的一串字符串模型爱写什么写什么。但开放Action很容易导致模型自创工具名明明系统里没有这个工具它也敢输出。我见过只写了Action: query_sales_data但系统里根本没有这个工具的案例结果解析直接崩。正确的做法是把Action改成受控枚举候选值全部来自路由表从根本上杜绝幻觉工具名。第二个增强是步长预算。真实场景中模型可能陷入过度调用。比如一个“写周报”任务它读了十个文档还在继续读始终不敢开始写。max_steps就是硬性预算到点必须产出或降级。我在项目中把默认max_steps设为8超过之后强制让模型基于已有信息回答并明确告知“你现在的可用步骤已经用完”。第三个增强是策略切换探针。我常用的两个探针是“重复检测”和“信息增益检测”。重复检测是看最近两轮是否调用了同一个工具且参数一致如果是说明这一支路没走通应切换路线。信息增益检测则是对比最近一轮observation里是否出现新信息如果连续三轮没有新信息继续走下去大概率也是空转这时候应该触发澄清分支直接问用户。3.3 多Agent协作场景中的路由与编排单Agent玩明白之后很多人会往多Agent方向走。多Agent里的动态路由典型的形态是“管理员Agent分发任务”。系统里注册了多个子Agent每个子Agent有一份能力清单管理员Agent根据用户请求选择合适的执行者。一个我实际在用的注册表结构如下AGENT_REGISTRY [ { agent_id: data_agent, capabilities: [数据分析, SQL查询, 可视化], metadata: {max_input_tokens: 4000} }, { agent_id: code_agent, capabilities: [代码生成, 调试, 代码审查], metadata: {max_input_tokens: 8000} }, { agent_id: writer_agent, capabilities: [文案撰写, 润色, 摘要], metadata: {max_input_tokens: 6000} } ]多Agent路由决策就不能只看关键词了因为用户的请求描述通常是复合的比如“分析数据并生成一份图文报告”这既涉及数据分析又涉及文案撰写。这里我建议把决策从“选一个”改成“做一个编排计划”。让编排器输出一个计划数组依次执行计划: 1. data_agent 分析数据输出结果表 2. writer_agent 基于结果表撰写报告这种“子任务编排计划”比单纯的单跳路由更接近真实业务需求。但要注意这里的编排计划必须由编排器做合理性校验。比如计划中某个Agent的输入依赖另一个Agent的输出如果顺序写反了整个任务链就会断掉。我在编码时会把每个子Agent的输出schema固定好前一个Agent的output_key必须是后一个Agent的input_key校验不通过就不放行。3.4 失败分支的自动降级自适应编排最见功力的地方不是顺利的时候而是失败的时候怎么兜底。线上最常见的三类失败我整理成了对应的降级策略。工具执行异常是最常见的比如SQL语句报错、外部API超时。不要直接把报错信息甩给用户也不要无脑重试而是把错误信息回填给模型让它分析原因并修正参数。我做过一个SQL查询Agent它在查询失败时会把报错内容拼接进上下文然后再次生成修正后的SQL成功率能提高五六个百分点。但必须设置重试上限同一工具的连续重试超过2次就放弃该路线并切换分支。第二类是模型输出格式异常。比如要求输出JSON结果模型给了一段带注释的文本。解析失败时不要简单判定“失败了”可以做一次格式修复。用一段宽松的正则把JSON部分抽取出来或者给模型一次“修正机会”把解析错误信息反馈给它让它重新输出。实际效果显著尤其是引入了带引号转义等复杂JSON结构后一次修复的成功率大约在70%以上。第三类是整条链路走不通。比如任务需要查A库但A库连不上。自适应编排要能“换路”从备用数据源读取或者询问用户是否接受降级数据。如果备用也没有就把“能力边界”明确告诉用户而不是硬着头皮编造结果。做过Agent的人应该都有共鸣模型在走投无路时特别容易开始一本正经地胡说八道可控的自适应编排就是指明确告诉它“到这里真的没路了请停止”。3.5 记忆资源对编排的影响写到这里还要提一下记忆。一开始很多人以为记忆模块只是“存取聊天记录”但在自适应编排里记忆直接影响路由质量。为什么因为意图识别在上下文不足时很容易把用户的追问当成独立的新问题。比如用户先问“帮我查A公司的联系方式”Agent已经走完了一遍数据库查询流程用户接着问“那B公司呢”如果编排器没有把上一轮的任务目标挂到当前上下文路由决策时就会把这句话当成无头无尾的闲聊。我的做法是维护一个轻量级的“会话任务栈”在每个路由决策之前先检查当前请求是否与最近一次任务目标存在指代关系。如果有就把原始任务描述和当前请求合并成一个完整意图再去路由。这个机制不需要额外调模型用规则判断代词或上下文主题一致性就够用。执行效果是追问类场景的路由准确率从惨不忍睹提升到可接受水平。4. 常见问题与排查技巧实录4.1 高频问题速查表我把实际项目里遇到的高频问题整理成了一个速查表基本覆盖了动态路由和自适应编排上线初期的绝大多数疑难杂症现象可能原因排查方法解决方案路由经常选错分支意图描述写得太笼统打印每轮路由的reason日志观察决策依据再写路由表description加入正面/反面约束模型输出不存在的工具名Action字段完全开放检查工具列表是否有遗漏名称将Action改为枚举类型只允许路由表内标识同一轮循环反复调用同一个工具探针函数没有检测到重复检查detect_no_progress逻辑对比最近两轮action_id和参数一致则强制切换Agent执行几轮后突然报错终止执行过程中某个观察值不是预期格式查看终止前一步的原始observation给工具输出增加统一schema封装任务内容不完整就强行回答max_steps设置得太小统计真实执行的步数分布调大max_steps或优化prompt让模型减少冗余步骤多Agent协作时子任务顺序错乱编排计划缺少依赖校验打印计划数组人工核对引入input_key/output_key依赖检查追问场景路由到错误意图会话任务栈没有传递上下文查看路由日志中是否存在孤立请求实现指代消解合并前后轮次的目标描述无可用分支时模型开始编造答案没有配置fallback路线检查路由表是否每条都有兜底增加clarification兜底允许明确说做不到4.2 路由决策过程的可视化复盘做动态路由还有一个绕不开的经验可观测性一定要前置。我见过太多项目上了动态路由之后只能看到“用户问了一个问题Agent答了一个结果”中间经历了哪几条分支、为什么选这条、哪一步触发了切换全是一团黑盒。出了问题想复盘一脸懵。我的做法是给每个请求生成一个独立的trace_id并在路由决策、工具执行、策略切换这三类关键节点埋点。每个埋点记录timestamp、route_id、匹配到的关键词、LLM决策的reason、工具的输入输出摘要。后续可以用简单的脚本把某次请求的所有埋点串起来生成一条可读的执行链路。这样做还有一个额外收益——当老板或者产品经理问“为什么Agent这次答错了”的时候你甩出一条链路日志就能定位问题而不是拍脑袋说“模型抽风”。埋点本身不复杂寥寥数行代码就能搞定import json import time class RouterTracer: def __init__(self, trace_id: str): self.trace_id trace_id self.events [] def log(self, event_type: str, payload: dict): self.events.append({ trace_id: self.trace_id, ts: time.time(), type: event_type, **payload }) def dump(self): return json.dumps(self.events, ensure_asciiFalse, indent2)4.3 一个真实案例的排障过程讲一个我印象很深的线上事故。某次新功能上线后用户反馈Agent频繁答非所问尤其在“查一下本周的数据”这种请求上时而查库时而去搜网络。排查时第一轮我怀疑是LLM路由分类不准但日志拉出来一看发现一个意外现象这类请求其实每一次都准确命中了database_query路由问题出在数据库查询结果为空时编排器把空结果当成了“无信息”触发了一条我事先没注意的兜底路线——web_search降级。更棘手的是web_search搜索回来的都是些泛泛而谈的行业新闻跟用户想要的本周内部数据完全不搭边。系统就这么一本正经地把搜来的内容包装成回答给了用户。问题根源不在路由选错了而在于空结果的语义理解不对。database_query返回空正确动作应该是提示“没有找到匹配记录请缩小查询范围”而不是转去搜索外部信息。修复方案也很直接给查询结果的返回增加一个状态字段区分“查询成功但无数据”和“查询失败”编排器只在查询失败时才允许降级到web_search空数据则直接进入澄清分支。这个案例给我的教训是路由表和降级策略一定要把“业务语义”考虑进去技术上的空值和业务上的没有数据是两回事。4.4 测试与压测别只在Golden Set上自嗨最后聊一下测试。动态路由加自适应编排之后系统的状态空间变得非常大靠手工Case验证根本覆盖不过来。我建议维护一套分层的评测集第一层是稳定的黄金数据集大概几百条典型的输入每条标注期望路由和期望最终行为每次改代码后全量回归。第二层是模糊输入集专门放那些没有标准答案的输入看路由能不能给出合理路线不要求唯一正确答案但要求不崩、不越权、合理降级。压测同样重要尤其是规则路由阶段一定要确认海量并发时匹配效率不会退化。我记得有一次压测规则路由逻辑里用了双重循环去遍历关键词路由表一涨耗时就线性飙升压测直接超时。后来把所有关键词预编译成ac自动机多模式匹配耗时从几百毫秒降到几毫秒。动态路由虽然不复杂但真要支撑高并发细节里的坑一个都不会少。写在最后的一点个人体会做动态路由和自适应编排这段时间我最大的感受是不要在提示词里让模型“自由发挥流程”而是把流程拆成可观察、可控制、可回退的决策点。路由表是数据编排循环是骨架埋点日志是眼睛三者缺一不可。很多人一说动态路由就想到复杂的强化学习或图搜索但在绝大多数业务场景下一张精心设计的路由表加一个带探针的循环执行器已经能解决90%的灵活性问题。如果让我给刚入局的同行一个建议我会说先把一次请求的完整执行链路完整打印出来盯着一百条链路看你会比看一百篇论文收获更多。路由不准就改描述循环空转就加探针分支走死就配降级一步一步迭代系统会越来越稳。
返回列表