Agent路由规则设计：从if-else到四层Runtime决策流-北京尧图网络

1. 这不是选模型是设计决策流为什么“先把路由规则写进 Runtime”成了新共识最近在几个技术社区和内部分享里反复听到一句话“别急着挑 Agent 模型先把你脑子里的判断逻辑一行行写进 Runtime。”这句话乍听像反常识——毕竟大模型选型动辄涉及参数量、推理速度、工具调用能力、长上下文支持、甚至微调成本怎么就让“写路由规则”抢了风头但如果你真带过三个以上落地 Agent 项目就会发现90% 的线上故障、响应延迟、意图误判、工具链断裂根源不在 LLM 本身而在于 Runtime 层那几行没写稳的 if-else、fallback 策略、状态跳转条件甚至一个没加 timeout 的 await。所谓“Agent 模型怎么选”本质是个伪命题真实问题是——你打算让谁来当这个系统的“交通指挥员”是靠模型自己“猜”还是由你亲手定义“什么情况走哪条路”。我把这个转变称为从“LLM-centric”到“Runtime-first”的范式迁移。核心关键词——Agent、路由规则、Runtime——不是孤立概念。Agent 是行为体Runtime 是它的操作系统而路由规则就是这个操作系统里的调度内核。它决定用户问“查下我上个月的报销进度”该交给财务知识库检索还是触发审批系统 API还是直接拒答并引导人工客服它也决定当工具调用超时或返回空结果时是重试、降级为摘要生成、还是切换到备用模型兜底。这些决策不能依赖模型“临场发挥”因为 LLM 的不确定性会层层放大——一次 token 预测偏差可能引发整个任务流偏航。所以“先把路由规则写进 Runtime”不是降低对模型的要求而是把最不可控的部分模型输出框进最可控的框架确定性逻辑。这就像给一辆自动驾驶汽车装上铁轨车可以换但轨道走向必须提前铺好。适合谁看如果你正在做 RAGAgent 混合架构、需要对接多个异构系统、或者正被“模型很厉害但总不按预期做事”困扰这篇就是为你写的。它不讲大模型原理只讲你怎么用几十行代码把混沌的智能行为变成可追踪、可测试、可灰度的确定性流程。2. 路由规则不是 if-else 堆砌四层结构拆解与设计逻辑很多人一听到“写路由规则”第一反应是写一堆条件判断if intent query_order then call_order_api()。这没错但远远不够。真正健壮的 Agent Runtime 路由是一套分层结构每一层解决一类问题彼此解耦又协同。我把它拆成四层意图识别层、能力匹配层、执行策略层、容错恢复层。这四层不是线性流程而是嵌套反馈的网状结构。下面逐层解释为什么这么设计以及每层背后的真实业务约束。2.1 意图识别层为什么不用 LLM 直接分类而要加一层轻量 classifier直觉上让 LLM 自己判断“用户这句话想干什么”最省事。但实测下来问题极多一是响应延迟高每次都要过一遍大模型二是分类边界模糊比如“帮我看看账单”和“账单有问题吗”语义接近但动作不同三是无法快速迭代改个分类标签得重新训模型。我们团队在某金融客服 Agent 中试过纯 LLM 意图识别上线后发现 37% 的误分类集中在“咨询类 vs 投诉类”这种高风险区分上且 AB 测试显示增加 200ms 延迟后用户放弃率上升 11%。于是我们退了一步用一个 3MB 的 ONNX 格式轻量级文本分类器基于 distilbert-finetuned做初筛只负责 8 个一级意图查询、申请、投诉、挂失、转账、余额、帮助、闲聊准确率 92.4%P99 延迟 18ms。LLM 只在 classifier 输出置信度 0.85 时才介入二次确认。这个设计的核心逻辑是把确定性高的决策交给确定性高的工具把模糊地带留给大模型兜底。它不是取代 LLM而是给 LLM 减负、提速、划边界。你完全可以用 spaCy 规则关键词白名单做更轻量的版本重点是建立“快速过滤→模糊交由大模型”的分层信任机制。2.2 能力匹配层模型不是万能的你的 Runtime 要知道“谁擅长什么”这一层常被忽略却是选模型的关键前置动作。假设你有三个可用模型Model-A强推理长上下文 128K但工具调用慢、Model-B专精 API 调用支持 function calling但逻辑链短、Model-C轻量版用于兜底摘要。路由规则在这里要回答“当前任务哪个模型最可能一次成功” 我们不用“哪个最强”而用“哪个最适配”。例如当任务需调用 3 个以上外部 API 且需严格顺序时 → 强制路由至 Model-B因其 function calling 稳定性经压测验证当输入含大段 PDF 解析文本50K tokens且需跨页推理时 → 路由至 Model-A当前请求已连续两次失败且用户语气词增多如“啊”“真的吗”→ 降级至 Model-C 生成简洁摘要并附“可转人工”按钮。这个决策不能靠 prompt 提示必须硬编码进 Runtime 的能力矩阵表。我们维护一张 JSON 表字段包括 model_id、max_context_len、tool_call_stability_score0-1、avg_latency_ms、fallback_priority。每次路由前Runtime 查这张表结合当前任务特征token 数预估、工具调用数、历史失败次数做加权打分。这不是玄学而是把模型的非功能属性性能、稳定性、成本显式建模让选型从“感觉哪个好”变成“数据算出来哪个优”。2.3 执行策略层同一意图不同路径——这才是真正的“智能”到这里意图定了模型也选了但还没完。同一意图执行路径可以完全不同。比如“订会议室”看似简单但实际分支极多用户说“今天下午三点订个能坐 10 人的会议室”这是完整指令 → 直接调日历 API 创建事件用户说“我想订会议室”没时间没人数 → 启动多轮澄清流程Runtime 主动发问“请问是今天还是明天大概几点需要容纳多少人”此时模型只负责生成自然语言提问不执行创建用户说“昨天那个会议室取消掉”需先做指代消解Runtime 从 session state 中提取上文 ID再调取消接口。执行策略层的核心是把“任务状态机”显式化。我们用状态图 DSL类似 XState 语法定义每个意图下的合法状态转移idle → asking_time → asking_capacity → confirming → executing → done。Runtime 不仅驱动状态流转还记录每一步的耗时、用户响应、模型输出置信度。这样当某次“asking_capacity”环节用户没回复系统可自动降级为“默认按 6 人处理”而不是卡死。这层的价值在于把隐性的对话逻辑变成可配置、可审计、可回滚的显性流程。它让 Agent 的“智能”有了骨架而不是飘在 prompt 里的幽灵。2.4 容错恢复层别幻想模型永不犯错要设计“错得优雅”所有团队都经历过模型返回了根本不存在的 API 名称、工具调用返回 500 却没抛异常、用户突然中断对话导致状态不一致……容错不是加个 try-catch 就完事。我们定义了三级恢复机制L1 微恢复针对单次调用失败如 HTTP 超时Runtime 自动重试最多 2 次指数退避并记录重试日志L2 中恢复针对流程级失败如“订会议室”中地址解析失败Runtime 不报错而是切换到“基于城市名模糊匹配会议室”的备用路径并在响应末尾加一句“已为您匹配到北京中关村区域的可用会议室”L3 宏恢复针对连续失败如 3 次意图识别失败2 次工具调用失败Runtime 强制终止当前 task flow保存当前上下文快照返回结构化错误码如 ERR_FALLBACK_EXHAUSTED前端据此展示“系统正在优化请稍后重试”并提供人工入口。关键点在于所有恢复动作都必须可配置、可开关、可监控。我们在 Runtime 中预留了 recovery_config 字段支持 per-intent 配置。比如财务类意图禁用 L2 模糊匹配因数据敏感而客服类意图开启全量恢复。这层的设计哲学是承认不确定性但把不确定性关进笼子让它按你的规则出牌。3. 实操从零手写一个可运行的 Runtime 路由引擎含完整代码光讲理论不够下面带你手写一个最小可行的 Runtime 路由引擎。它不是玩具而是我们某客户生产环境简化版已跑过日均 200 万请求。我们用 Python FastAPI 实现核心就三个文件router.py路由主逻辑、models.py模型能力注册、state.py会话状态管理。全程不依赖 LangChain/LlamaIndex 等框架确保你一眼看懂每行代码在干什么。3.1 基础结构定义路由核心契约首先明确 Runtime 的输入输出契约。输入是用户消息 session_id metadata如用户角色、设备类型输出是结构化 actionfrom typing import Dict, List, Optional, Literal from pydantic import BaseModel class RouteInput(BaseModel): user_message: str session_id: str metadata: Dict[str, str] # e.g. {role: admin, device: mobile} class RouteOutput(BaseModel): model_id: str # 选定的模型ID action_type: Literal[call_tool, generate_response, ask_clarify, fallback_human] tool_name: Optional[str] None # 如需调用工具 tool_params: Optional[Dict] None # 工具参数 next_state: Optional[str] None # 下一状态名 confidence: float # 路由置信度用于监控这个契约强制把所有决策外显化。注意action_type是枚举值不是字符串拼接——这避免了后期维护时出现 call_api 和 call_tool 这种命名混乱。我们坚持契约即文档类型即规范。3.2 意图识别层实现轻量 classifier LLM 仲裁我们不用 HuggingFace pipeline而是直接加载 ONNX 模型确保低延迟# router.py import onnxruntime as ort import numpy as np from transformers import AutoTokenizer class IntentClassifier: def __init__(self, model_path: str, tokenizer_name: str): self.session ort.InferenceSession(model_path) self.tokenizer AutoTokenizer.from_pretrained(tokenizer_name) def predict(self, text: str) - tuple[str, float]: inputs self.tokenizer( text, truncationTrue, max_length128, return_tensorsnp ) ort_inputs { input_ids: inputs[input_ids].astype(np.int64), attention_mask: inputs[attention_mask].astype(np.int64) } logits self.session.run(None, ort_inputs)[0] probs np.exp(logits) / np.sum(np.exp(logits)) intent_id np.argmax(probs) confidence float(probs[0][intent_id]) return INTENT_MAP[intent_id], confidence # 全局实例单例 intent_classifier IntentClassifier(./models/intent.onnx, distilbert-base-uncased)INTENT_MAP 是预定义的映射字典如{0: query_balance, 1: apply_refund}。这里的关键细节我们限制最大长度为 128因为实测超过此长度classifier 准确率下降明显且没必要——长文本应由后续模型处理。当confidence 0.85时我们才调用 LLM 做二次确认prompt 极简请从以下选项中选择最匹配用户意图的编号只返回数字不要解释 1. 查询余额 2. 申请退款 3. 投诉服务 ... 用户输入{user_message}LLM 只返回一个数字我们用正则提取避免解析失败。这比让 LLM 自由输出“我要查余额”再做 NER 稳定得多。3.3 能力匹配层模型能力矩阵与动态打分models.py中定义所有可用模型及其能力# models.py from dataclasses import dataclass from typing import Dict, Any dataclass class ModelSpec: model_id: str max_context_tokens: int tool_call_stability: float # 0-1, 来自压测报告 avg_latency_ms: float fallback_priority: int # 数值越小越优先兜底 MODEL_SPECS: Dict[str, ModelSpec] { model-a: ModelSpec( model_idmodel-a, max_context_tokens131072, tool_call_stability0.72, avg_latency_ms1240, fallback_priority3 ), model-b: ModelSpec( model_idmodel-b, max_context_tokens32768, tool_call_stability0.98, avg_latency_ms420, fallback_priority1 ), model-c: ModelSpec( model_idmodel-c, max_context_tokens8192, tool_call_stability0.85, avg_latency_ms180, fallback_priority2 ) } def select_model(intent: str, input_tokens: int, tool_count: int, failure_count: int) - str: 根据任务特征选择最优模型 candidates [] for spec in MODEL_SPECS.values(): # 基础过滤上下文容量必须足够 if input_tokens spec.max_context_tokens * 0.8: continue # 加权打分可配置权重 score ( spec.tool_call_stability * (0.4 if tool_count 0 else 0.1) (1 - spec.avg_latency_ms / 2000) * 0.3 # 归一化延迟得分 (3 - spec.fallback_priority) * 0.3 # 优先级越高得分越高 ) candidates.append((spec.model_id, score)) if not candidates: return model-c # 终极兜底 return max(candidates, keylambda x: x[1])[0]注意input_tokens不是原始字符数而是用 tiktoken 估算的 token 数我们封装了estimate_tokens()函数。这个打分公式不是魔法而是我们 AB 测试 3 周后定稿的工具调用稳定性在“多工具流程”中权重最高延迟在“高频查询”中权重更高。你可以根据业务调整系数但关键是——把主观经验变成可调参数。3.4 执行策略层状态机驱动的多轮交互state.py管理会话状态核心是SessionState类# state.py from enum import Enum from typing import Dict, Any, Optional class SessionState: def __init__(self, session_id: str): self.session_id session_id self.current_intent: Optional[str] None self.state_stack: list[str] [idle] # 状态栈支持嵌套 self.context: Dict[str, Any] {} # 存储临时变量如 user_id, meeting_room_id def push_state(self, state: str): self.state_stack.append(state) def pop_state(self) - str: return self.state_stack.pop() if len(self.state_stack) 1 else idle def get_current_state(self) - str: return self.state_stack[-1] # 状态转移表简化版 STATE_TRANSITIONS { query_balance: { idle: asking_account, asking_account: executing, executing: done }, book_meeting: { idle: asking_time, asking_time: asking_capacity, asking_capacity: confirming, confirming: executing, executing: done } } def get_next_state(intent: str, current_state: str) - Optional[str]: transitions STATE_TRANSITIONS.get(intent, {}) return transitions.get(current_state)路由主函数整合所有层# router.py def route(input_data: RouteInput) - RouteOutput: # 1. 意图识别 intent, conf1 intent_classifier.predict(input_data.user_message) # 2. 获取会话状态 session get_session(input_data.session_id) # 从 Redis 或内存获取 current_state session.get_current_state() # 3. 获取下一状态 next_state get_next_state(intent, current_state) # 4. 估算 token 数和工具数 input_tokens estimate_tokens(input_data.user_message) tool_count 0 if intent in [book_meeting, apply_refund]: tool_count 2 # 预设 # 5. 模型选择 model_id select_model( intentintent, input_tokensinput_tokens, tool_counttool_count, failure_countsession.failure_count ) # 6. 决策 action_type action_type generate_response tool_name None tool_params None if next_state executing: action_type call_tool if intent book_meeting: tool_name calendar_api.create_event tool_params {time: session.context.get(time), capacity: session.context.get(capacity)} elif next_state in [asking_time, asking_capacity, confirming]: action_type ask_clarify elif next_state done: action_type generate_response return RouteOutput( model_idmodel_id, action_typeaction_type, tool_nametool_name, tool_paramstool_params, next_statenext_state, confidenceconf1 )这个函数就是 Runtime 的心脏。它把四层逻辑串成一条流水线每一步都有明确输入输出。你可以看到next_state直接决定了action_type而action_type又决定了后续调用哪个模型、传什么参数。没有 magic全是清晰的 if-else 和函数调用。3.5 容错恢复层三明治式错误处理最后在 FastAPI endpoint 中加入容错# main.py from fastapi import FastAPI, HTTPException from starlette.middleware.base import BaseHTTPMiddleware app FastAPI() class RouteMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): try: response await call_next(request) return response except Exception as e: # L1 微恢复记录错误返回标准错误响应 logger.error(fRoute error: {e}, extra{session_id: request.state.session_id}) return JSONResponse( status_code500, content{ error: ERR_ROUTE_FAILED, message: 系统繁忙请稍后重试, suggestion: 可尝试简化问题描述 } ) app.add_middleware(RouteMiddleware) app.post(/route) async def route_endpoint(input_data: RouteInput): try: # 主路由逻辑 output route(input_data) # L2 中恢复检查输出合理性 if output.action_type call_tool and not output.tool_name: logger.warning(Tool name missing, falling back to generate, extra{session_id: input_data.session_id}) output.action_type generate_response output.model_id model-c # 降级模型 return output except Exception as e: # L3 宏恢复触发全局降级 logger.critical(Critical route failure, exc_infoe) return RouteOutput( model_idmodel-c, action_typefallback_human, confidence0.0, next_statefallback_human )注意L2 中恢复的例子当action_type是call_tool但tool_name为空时我们不报错而是静默降级为generate_response并换模型。这是典型的“错得优雅”——用户无感知系统继续运转。所有错误都打日志且带session_id方便后续用 ELK 关联分析。4. 真实踩坑记录那些文档里不会写的 7 个致命细节写了三年 Agent踩过的坑比读过的 paper 还多。下面这 7 个细节每一个都曾让我们凌晨三点在服务器前抓狂。它们不会出现在任何官方文档里但却是 Runtime 稳定性的命脉。4.1 Token 估算必须用目标模型的 tokenizer不能混用我们曾用gpt-4的 tokenizer 估算llama3-8b的输入长度结果线上大量请求因“超长截断”导致意图识别失败。原因不同 tokenizer 对中文、标点、emoji 的切分规则天差地别。gpt-4把“✅”算作 1 tokenllama3算作 4 个。解决方案为每个模型维护独立的tokenizer实例并在estimate_tokens()函数中根据model_id动态选择。我们甚至写了个校准脚本随机抽 1000 条真实用户消息用各模型 tokenizer 分别计数画分布图找出误差 15% 的模型单独优化其估算逻辑。4.2 Session State 必须带 TTL且 TTL 要分层设置早期我们把所有 session 存 RedisTTL 统一设 24 小时。结果发现用户中午问“订会议室”晚上又来问系统还认得上午的meeting_room_id直接复用旧 ID 导致冲突。更糟的是恶意用户构造海量 session_idRedis 内存爆满。现在我们分三层 TTL短期状态如 multi-turn clarifying paramsTTL15 分钟过期自动清空中期上下文如用户身份、偏好TTL2 小时支持跨会话轻量记忆长期元数据如用户角色、权限不存 session从主数据库实时查。关键技巧在get_session()函数里每次访问都刷新短期状态的 TTL但不动中期状态——这样用户活跃时状态永不过期静默后自动清理。4.3 模型能力矩阵的 stability_score必须来自线上真实错误率而非压测我们曾用 Locust 压测得出model-b的 tool_call_stability0.99结果上线后发现当并发 500 时其 function calling 失败率飙升至 12%因底层 API 限流。压测环境太干净没模拟网络抖动、下游服务降级。现在我们的 stability_score 来自 Prometheus 抓取的tool_call_failure_rate{modelmodel-b}指标滚动 1 小时窗口计算。每天凌晨自动更新MODEL_SPECS确保路由永远基于最新真实数据。4.4 “Ask Clarify” 的提问模板必须预生成禁止 runtime 拼接最初我们让模型根据next_state自动生成提问如“请问您要订几点的会议室”。结果发现模型有时会生成“您希望会议室有咖啡机吗”完全偏离主线。现在所有 clarify 提问都写死在 YAML 配置里# prompts/clarify.yaml book_meeting: asking_time: 请问您希望会议安排在几点例如下午3点 asking_capacity: 请问需要容纳多少人例如8人 confirming: 请确认{time}{capacity}人地点{location}是否正确Runtime 根据next_state查表填充变量。这样保证提问精准、友好、可控。变量填充用jinja2安全无注入。4.5 Fallback Human 的触发必须带上下文快照且快照要脱敏当触发人工客服时不能只传“用户说了一句话”而要传结构化快照{ session_id: abc123, route_trace: [idle-asking_time-asking_capacity, model-b selected, tool_call_failed], last_3_messages: [ {role: user, content: 订个会议室}, {role: assistant, content: 请问您希望会议安排在几点}, {role: user, content: 下午三点} ], redacted_context: {time: 15:00, capacity: 6} // 敏感字段已脱敏 }这个快照让客服 3 秒内理解来龙去脉无需再问“您之前说什么了”。脱敏用正则白名单如手机号、身份证号、银行卡号字段一律替换为[REDACTED_PHONE]。4.6 路由日志必须包含 decision_reason 字段否则无法 debug我们曾花两天排查一个“总是路由到 model-c”的问题最后发现是input_tokens估算函数里有个 bug把所有中文都算成 0 token导致input_tokens spec.max_context_tokens * 0.8永远为 True。如果日志里有decision_reason: input_tokens(0) model-a_threshold(104857)10 分钟就能定位。现在所有RouteOutput都带decision_reason字段内容是决策过程中的关键判断依据如intent_confidence_low(0.780.85), tool_count_high(32)。这些字符串在日志里用|分隔方便 Splunk 提取分析。4.7 灰度发布路由规则必须用 A/B 测试框架而非简单开关上线新路由逻辑时我们不用if new_routing_enabled:而是用 Statsig 的 Feature Gateif statsig.check_gate(new_routing_logic, {session_id: input_data.session_id}): output new_route_logic(input_data) else: output legacy_route_logic(input_data)Feature Gate 支持按 session_id 哈希分流5% 流量且可实时关闭。更重要的是它自动上报指标新逻辑的success_rate、avg_latency_ms、fallback_human_rate与旧逻辑对比。只有新逻辑的success_rate提升 0.5% 且fallback_human_rate不升才全量。这避免了“我觉得改得更好结果线上崩了”的悲剧。5. 模型选型的真相没有最好的模型只有最合适的路由策略回到标题那个问题“Agent 模型怎么选” 现在我可以给你一个直白的答案别选让路由规则替你选。你不需要背熟 Llama3 的 MoE 结构、Qwen2 的 RoPE 配置、或者 Claude 的 constitutional AI 原理。你需要的是搞清楚三件事第一你的任务流里哪些环节必须高稳定如支付扣款哪些可以高灵活如闲聊第二你的用户容忍度在哪——是宁可多问两句也要准确还是宁愿给个近似答案也要快第三你的工程资源够不够——能不能为每个模型配专属 tokenizer、能不能建实时 metrics pipeline、能不能写状态机 DSL。我们团队目前维护着 7 个模型从 1B 到 72B有的跑在 GPU有的跑在 CPU。但对外暴露的只有一个/route接口。运维同学从不关心“model-b 今天 latency 高不高”只看 Dashboard 上route_success_rate是否低于 99.5%。因为路由规则已经把模型的波动转化成了可监控、可告警、可自动降级的确定性指标。这就像电力系统你不用懂发电机原理只要插上插座灯就亮。Runtime 的价值就是把大模型的复杂性封装成一个稳定、可靠、可预测的“智能插座”。最后分享一个小技巧每次上线新路由规则前我们必做“三问测试”——问失败如果这个规则失效最坏结果是什么如路由错导致扣错钱还是只是回复慢一点问监控这个规则的健康度用哪 1 个指标能最准反映如clarify_ask_rate突增说明意图识别出问题问回滚如果出问题30 秒内如何切回旧逻辑我们所有路由函数都带version参数/route?v2即可秒切这三问逼你把模糊的“应该没问题”变成具体的“哪里可能出问题”。Agent 的未来不属于最炫的模型而属于最稳的 Runtime。当你能把路由规则写进代码、测进 CI、监控进大盘、回滚进一键操作时模型选型真的就只是一个配置项了。

Agent路由规则设计：从if-else到四层Runtime决策流

相关推荐

图书系统软件测试课程设计：从测试需求分析到用例设计全流程

视频配乐新范式：语义、时间、节奏三重对齐的AI生成系统

Windows 安装 Redis 全攻略：原生 exe 与容器化部署详解

更多资讯

litellm：统一大模型接口，实现多模型路由与成本追踪

论文复现为何失败？从“标杆成果”被神化到拆穿的技术真相

AI辅助发明专利申请：发明人资格与权利要求构建

knowledge-work-plugins：知识工作插件化设计与实践

文本聚类技术全解析：从向量化到K-Means的实践指南

逸珑8089A运行AOSC-RETRO时apt升级卡顿的根因与四步优化方案

今日更新

微信云开发旅游小程序实战：LBS打卡+UGC社交全栈方案

金融AI代理平台核心拆解：从多代理协作到财报分析实战

基于大数据的智慧旅游推荐与可视化平台架构实践

本周精选

金融AI落地指南：从调研报告到技术选型与工程实践

大模型上下文成本与注意力优化技术深度解析——为何O(n²)平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测

基于GMM的语音识别系统原理与MATLAB实现详解

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？