Agent动态路由模型与工具：五件自制件的设计实践

实话说，这两年我经手过好几个 Agent 类项目，最让我头大的环节往往不是怎么写提示词，也不是怎么搭 RAG，而是另一件听起来特别基础、做起来却容易翻车的事：Agent 遇到一个任务时，它自己心里有没有数，该找哪个模型？该调哪个工具？我们最近在内部的一个模拟助手项目里，干脆把“选择权”直接交给了 Agent，让它自己挑模型、自己挑工具。为此写了五个自制件：模型描述符注册表、工具能力网关、任务意图路由器、反馈回路、配置热加载器。整个过程踩了不少坑，也摸出了一些取舍逻辑，这篇就专门聊聊这些设计背后的“为什么”和实操里的教训。

如果你正在做类似的 Agent 编排，或者你手头的智能体总是“什么任务都打同一个模型、乱调工具”，那这五件自制件的思路大概率用得上。哪怕你只是想搞明白 Agent 怎么动态路由模型，也能从这套设计中找到可落地的切入点。我不保证这方案是银弹，但它确实让我们项目的灵活性上了一个台阶。

1. 先把五件自制件拆开：各自解决什么问题，为什么这么切

1.1 模型描述符注册表：把模型变成可枚举、可打分的数据

先把“模型”这个词从抽象概念变成数据问题。以前项目里的模型选择基本都是写死的，代码里硬编码一个模型名，所有请求全往那边发，模型升级、换供应商、加新模型都得动代码。五件自制件里的第一件，就是做了一张“模型身份证”登记表。

每个模型都会登记一组结构化的描述，包括：模型唯一标识、供应商、上下文长度、最大输出 token、擅长任务类型（代码、数学、摘要、分类、生成等）、成本权重（按每千 token 折算出一个相对分数）、平均延迟水位、稳定性级别，以及这个模型是否允许被 Agent 直接调用。这些信息不是存字典里拍脑袋硬写，而是做成一张持续维护的注册表，Agent 在运行时读取它，才能有“挑选”的依据。

为什么必须这么做？很简单：如果 Agent 手里连模型清单都没有，那“自己挑模型”就是一句空话。我们见过不少团队想实现模型动态切换，结果只是把模型名放到环境变量里，那只能叫“手动切换”，不算“自己挑”。真正的动态路由，得有数据支撑，模型描述符就是这个数据底座。实际落地时我们建议用强类型结构来定义描述符，后面第三章会给出完整代码，现在先搭概念框架。

1.2 工具能力网关：让工具按统一格式“自我介绍”

第二个自制件是工具能力网关。它的作用是把散落在业务系统里的各种工具统一包装起来，让 Agent 能“看见”每个工具是干嘛的、需要什么参数、会产生什么副作用。

举例来说，内部有一个查询工单状态的接口、一个文档解析服务、一个SQL执行器。如果没有统一封装，让 Agent 自己去理解这几个接口的差异，它很容易搞混参数格式，也搞不清哪些操作是只读、哪些会改数据。工具能力网关做的事情，就是给每个工具写一份“能力白皮书”：工具名称、功能描述、入参 JSON Schema、出参结构、权限级别（只读/写入/高风险）、是否幂等、超时阈值、别名索引。

这里面有个很关键的设计取舍：工具描述到底是给模型看还是给路由代码看？答案是两者都要。给模型看的部分会放进函数调用机制里，让 Agent 自己判断调用哪个工具；给路由代码看的部分，则用于工具网关在运行时做参数校验、权限检查、链路追踪。如果你只做了给模型看的那份，后面会遇到无数“模型认为自己调用了工具，但代码里根本没有对应实现”的尴尬错误。

1.3 任务意图路由器：先给任务分类，再决定找谁

第三件是任务意图路由器，这个组件是所有自制件里最核心的。它解决的不再是“能选谁”，而是“该选谁”的决策逻辑。任务进来以后，路由器会先对任务做一次画像：任务类型是什么？复杂度高不高？是否是高风险操作？对实时性要求高不高？

画像完成之后，路由器会动态拼出候选集：代码类任务只从模型注册表里筛选“擅长代码”的那几个模型参与竞争；摘要类任务则优先拉取长上下文模型。这里有一个非常重要的工程取舍：不要让 Agent 在“全部模型”里做选择，而是先由规则把范围缩小到可竞争的候选集，再由 Agent 在候选集里做最终决策。为什么？因为模型列表一旦超过五个，把全量列表塞给模型，它就容易“挑花眼”，决策质量直线下降，同时 token 消耗也会上升。

这种“先分类后挑选”的结构，本质上就是给 Agent 装了一个决策漏斗。顶层是轻量级规则分类器，跑得飞快；底层是模型语义级的细粒度评估，交给大模型自己判断。我们曾经试过直接让大模型在几十个模型里选，效果惨不忍睹，后面章节会细说。

1.4 反馈回路：让错误选择成为下一次的线索

第四个自制件是反馈回路。这名字听起来有点学术，其实做出来就是一本“选择账本”。每次 Agent 做完一次模型选型或工具调用，最终任务的执行结果会回写到账本中，包括耗时、成功/失败、失败原因、用户是否纠正、模型返回质量评分等。

反馈回路最大的价值，是让 Agent 在不断试错中生成经验。比如有一次路由到模型 A 结果生成了错误的 JSON 格式，模型 B 之前遇到同类任务时表现很好，那么下一次遇到相似任务时，反馈回路就会提升模型 B 的优先级。这比完全靠规则拍脑袋科学得多。

但反馈回路也是最容易做砸的一个组件。如果只是无条件把历史经验写成硬规则，Agent 很快就会“路径依赖”——上一次用 A 成功了，下一百次也只会选 A，哪怕 B 新版本已经表现更好。如何避免这种震荡和僵化，我放到第二章专门讲，这里先记住核心原则：反馈信息只能作为“倾向权重”，不能作为“硬性约束”。

1.5 配置热加载器：改模型和工具不再需要重启服务

第五件是配置热加载器。刚开始我们的模型注册表和工具网关都是写在本地代码里的静态配置，每次调整模型优先级、新增工具接口，都得改代码、重新构建、重启服务。一次两次没感觉，等到频繁调优的时候，就会发现背后全是无效加班。

配置热加载器做的事情，是把模型注册表、工具清单、路由规则、成本阈值这些配置全部外部化存储，程序启动时加载一份快照，同时监听配置文件变化，校验通过后热更新到内存。注意，这里必须带“校验”两个字，不能盲改。

热加载核心挑战是并发一致性：一个请求在读取新配置，另一个请求还在用旧配置，很容易出现同一个任务，同一个模型，因为配置刷新时间不同而表现不一致。我们的妥协方案是“版本快照式加载”：每次请求开始前，统一获取当前的配置版本号，整条链路都用该版本的配置，不中途切换，这样从源头避免了吃半个新配置、半个旧配置的诡异状态。

2. 设计取舍与核心细节：选择权给到 Agent 之后，边界在哪

2.1 让 Agent“自己挑模型”：选型打分与成本护栏

模型选择的本质，是给候选模型按任务适配度打分。很多人刚开始会想用一个高大上的算法，比如训练一个排序模型或者嵌入向量做语义匹配。我们实操下来发现：先用加权规则打分，效果已经足够好，而且可解释性强得多。

打分规则可以按这个思路设计：基础分来自任务意图路由器给的画像匹配度。比如任务是代码生成，画像里“task_type=code”，注册表里“擅长任务类型包含 code”的模型加 20 分；上下文长度小于任务预估 token 量的模型直接淘汰；延迟敏感型任务，平均延迟高于阈值的模型扣 15 分；成本敏感型任务，成本权重高的模型扣 10 分。最后再加上反馈回路的倾向权重，就是该模型的最终得分。

这里有一个决策点：最终的 Top1 选择是不是直接执行？我们曾经图省事直接选最高分，结果偶尔会选到一个小众模型，虽然历史反馈分高，但对新任务的泛化能力其实不强。后来加了“最低采纳阈值”：如果最高分也低于一个合理线，就回落到默认模型处理，宁可保守也不要冒险。

成本护栏是另一个容易被忽略的点。模型自由选择的必然代价是成本不可控。我们加了三层闸门：第一层是全局成本预算，按日或按月设定总额；第二层是单次任务成本上限，当模型估算成本超过上限时直接淘汰；第三层是供应商配比，给每个供应商设定最大调用占比，避免业务被单一供应商“绑架”。这三层闸门筛选下来，模型选择的自由度还在，但失控的风险小很多。

2.2 让 Agent“自己挑工具”：函数调用的白名单策略

工具选择这块，主流的实现方向是依托模型厂商提供的内置函数调用机制，也就是让模型在生成回复时附带一个结构化的工具调用请求。这种做法本身没问题，但有个大坑：如果把所有工具全部放进函数调用列表里，模型会拿不准用哪个，还容易出现工具描述互相“串味”。

我们最踩的一个坑就是把所有工具一股脑塞给大模型。当时系统里大概有 30 多个工具，结果模型经常选错工具，或者在两个相似工具之间反复横跳。后来读了相关案例，意识到应该用“候选集闸门”：任务意图路由器先把工具全集缩小到可能相关的 3~5 个，再把这些候选工具交给模型做最终决定，整体准确率显著提升。这个逻辑和模型选择是同一个思路。

另外，工具能力网关里每个工具的“功能描述”写法，对选择准确性影响巨大。我们总结出一个规律：描述里要包含“触发场景”和“不适用场景”，而不是只写“这个工具可以做什么”。举例，一个 SQL 执行器的描述如果只写“执行 SQL 查询”，模型很可能在需要“只读报表”的时候也去调用它，实际上这个工具写操作隐患很大。正确写法是“仅可用于只读 SELECT 查询，所有写操作（INSERT/UPDATE/DELETE）一律拒绝”。把边界写清楚，模型不仅不会更糊涂，反而更果断。这大概是最低成本、最大收益的优化点。

2.3 反馈机制的防震荡：避免越修越歪

反馈回路如果不加约束，很快会演变成一场灾难。我们在内部把这种失控局面叫“反馈震荡”：某次调用模型 A 返回了错误结果，反馈回路立刻给 A 记了一笔，于是后面所有相似任务都绕开模型 A；但可能那只是一次偶发的服务不稳定，A 本身质量依然最高，导致系统“越修越歪”。

怎么防？第一，反馈样本必须带时间窗。只有最近一段时间内的反馈才有高权重，很早之前的成功经验会自然衰减，避免旧经验永久压制新模型。第二，反馈要区分“用户明确纠正”和“系统自动判断”。系统自动判断失败有时是误判，权重必须低于用户主动纠正。第三，必须保留随机探索比例。即使某个模型历史得分很高，也要有少量请求随机分给其他候选模型，只有不断探索，才能发现“老模型被偷偷升级了”或“新模型已经足够好用”这些变化。

我们实际采用的比例是 90% 按加权得分选，10% 按均匀随机选。就这 10% 的随机探索，帮我们捕获了好几次模型服务商悄悄升级或者降级的波动，这是纯规则系统永远发现不了的。

3. 实操：跑通一个可复现的最小闭环

理论说了一堆，还是得上代码。下面我会给出一个简化但完整的最小闭环，覆盖模型描述符、工具网关、意图路由、运行时选择、反馈回写和配置热加载这几个环节。为了便于阅读，这里用 Python 做示例，实际项目换成别的语言，思路通用。

3.1 定义模型描述符数据结构

模型描述符用强类型定义，这是后续一切逻辑的基础。我用的是带类型校验的数据类，让每个模型字段都有明确约束。

from enum import Enum from typing import List, Optional from pydantic import BaseModel, Field class TaskType(str, Enum): code = "code" summary = "summary" extract = "extract" math = "math" general = "general" class ModelProfile(BaseModel): model_id: str = Field(description="模型唯一标识") vendor: str = Field(description="供应商代号") context_window: int = Field(description="上下文窗口长度") max_output_tokens: int = Field(description="最大输出 token 数") skills: List[TaskType] = Field(description="擅长任务类型") cost_weight: float = Field(description="成本权重，数值越高越贵") latency_level: float = Field(description="平均延迟水位，数值越高越慢") enabled: bool = Field(description="是否允许被 Agent 直接调用") extra: dict = Field(default_factory=dict, description="扩展信息") def is_qualified(self, estimated_need_tokens: int) -> bool: # 上下文不够直接淘汰 if self.context_window < estimated_need_tokens: return False return True

在这个结构里，每个字段都服务于后续的决策逻辑。特别注意estimated_need_tokens这个潜在参数，它会在路由意图阶段被算出来，用来做粗筛。注册表的实际样子可以是一份 JSON 文件，也可以存到数据库里，但代码里一定要有 Profile 对象，而不是用零散的字典。

3.2 工具注册与能力描述

工具能力网关的代码，核心是提供一个统一的 register 装饰器，把原来散落的函数包成带能力描述的 Tool 对象。

from typing import Callable, Dict, Any, Optional import json class ToolProfile(BaseModel): name: str = Field(description="工具名，Agent 会基于这个名字做匹配") description: str = Field(description="包含触发场景和不适用场景的完整描述") parameters_schema: dict = Field(description="参数 JSON Schema，用于运行时校验") permission: str = Field(description="readonly / write / risky") idempotent: bool = Field(description="是否幂等") timeout_sec: int = Field(default=10) class Tool: def __init__(self, profile: ToolProfile, fn: Callable): self.profile = profile self.fn = fn REGISTRY: Dict[str, Tool] = {} def register_tool(name: str, description: str, permission: str, idempotent: bool, schema: dict, timeout_sec: int = 10): def decorator(fn): profile = ToolProfile( name=name, description=description, parameters_schema=schema, permission=permission, idempotent=idempotent, timeout_sec=timeout_sec, ) REGISTRY[name] = Tool(profile=profile, fn=fn) return fn return decorator @register_tool( name="query_order_status", description="查询工单当前状态。当用户询问工单进展、处理节点、停留时间时使用。不适用场景：创建工单、修改工单、删除工单。", permission="readonly", idempotent=True, schema={"type": "object", "properties": {"order_id": {"type": "string"}}, "required": ["order_id"]}, ) def query_order_status(order_id: str): return {"order_id": order_id, "status": "processing", "updated_at": "2026-01-01 12:00"}

注意描述里的“不适用场景”。之前提到过，这个报价相当关键。我们曾经测试过几组不同的描述写法，发现包含边界说明的版本在选择准确率上比只说功能高 20% 以上。别嫌描述字多，模型看得懂，省下的纠错成本才是大头。

3.3 任务意图路由：生成任务画像与候选集

任务意图路由器这一步，既不用每次都调大模型，也不用做太重的模型推理。先用规则 + 小分类器，产生一个 TaskProfile，然后基于 TaskProfile 过滤候选模型和候选工具。

class TaskProfile(BaseModel): task_type: TaskType estimated_need_tokens: int = Field(description="预估任务所需上下文 token 量") latency_sensitive: bool = Field(description="任务是否对延迟敏感") cost_sensitive: bool = Field(description="是否优先省钱") risky_operation: bool = Field(description="是否涉及高权限操作") keywords: List[str] = Field(default_factory=list) def route_task(user_query: str) -> TaskProfile: # 简化示例：基于关键词打标签，实际项目可以接一个轻量分类模型 if any(k in user_query for k in ["代码", "python", "function", "bug", "debug"]): task_type = TaskType.code elif any(k in user_query for k in ["摘要", "总结", "长文概括"]): task_type = TaskType.summary else: task_type = TaskType.general estimated_tokens = int(max(1000, len(user_query) * 12)) return TaskProfile( task_type=task_type, estimated_need_tokens=estimated_tokens, latency_sensitive=False, cost_sensitive=True, risky_operation=False, )

然后基于 TaskProfile 做候选集过滤。这一步替代了“全量列表塞给 Agent”的错误做法。

def filter_candidate_models(profile: TaskProfile, profiles: list[ModelProfile]) -> list[ModelProfile]: result = [] for mp in profiles: if not mp.enabled: continue if not mp.is_qualified(profile.estimated_need_tokens): continue if profile.task_type not in mp.skills: # 通用型模型可以兜底参与候选 if TaskType.general not in mp.skills: continue # 有风险操作时过滤掉不安全的模型 if profile.risky_operation and "risky_unsafe" in mp.extra: continue result.append(mp) return sorted(result, key=lambda x: -x.cost_weight)

过滤之后，把候选模型列表和候选工具列表交给 Agent 做最终决策。注意：候选集控制在 3~5 个以内，宁缺毋滥。

3.4 运行时选择：让 Agent 在候选集里最终拍板

真正交给 Agent 的时候，我们不再单独让模型“选模型”，而是把模型选择转变成一个标准的自主决策任务：给出一个“决策提示”，包含任务描述、候选模型清单、候选工具清单，让模型输出一个结构化的选择结果。

import asyncio def build_model_decision_prompt(user_query: str, profile: TaskProfile, cand_models: list[ModelProfile], cand_tools: list[ToolProfile]) -> str: model_lines = [] for m in cand_models: model_lines.append( f"- {m.model_id} (vendor={m.vendor}, cost_weight={m.cost_weight}, skills={[s.value for s in m.skills]})" ) tool_lines = [t.description for t in cand_tools] return f""" 你是一个资源调度决策器。根据任务描述，从候选模型和候选工具中选出最合适的组合。 只允许选择候选清单中出现的模型和工具，不要擅自引入清单之外的资源。 任务描述： {user_query} 任务画像： - 类型: {profile.task_type.value} - 预估token: {profile.estimated_need_tokens} - 延迟敏感: {profile.latency_sensitive} - 成本敏感: {profile.cost_sensitive} 候选模型： {chr(10).join(model_lines)} 候选工具： {chr(10).join(tool_lines)} 请输出 JSON，格式如下： {{"model_id": "选中的模型ID", "tool_names": ["选中的工具名"], "reason": "简短理由"}} 只输出 JSON，不要输出其他内容。 """

这一步有一个容易被忽略的细节：决策提示里要明确“只允许出现候选清单里的资源”。如果不加这句，大模型往往会自信地输出一个候选集之外的虚构工具名或模型名，相当于我们兜了半天把它关进笼子，它又自己撬锁跑出去了。加上这句以后，此类幻觉明显减少。

调用模型时，注意把输出的 JSON 做一轮严格解析，如果模型输出了非法 JSON，就自动重试一次，而不是直接抛错。我们把这个逻辑封装成一个 robust_llm_call 函数。

def robust_llm_call(prompt: str, retry: int = 2) -> dict: for attempt in range(retry): try: raw = llm_text_completion(prompt) # 你使用的模型 SDK data = json.loads(extract_json(raw)) return data except Exception as e: if attempt == retry - 1: # 最终失败回落到默认模型 return {"model_id": "fallback_default", "tool_names": [], "reason": f"llm parse error: {e}"} continue return {"model_id": "fallback_default", "tool_names": [], "reason": "unexpected"}

“回落默认模型”这个动作很重要。Agent 自主决策不是绝对自由，任何一步异常，都应该有安全网兜底。我们在生产环境里甚至预设了“默认调用链”，一旦 Agent 决策失败，直接走固定优先级的资源组合，确保业务流程不中断。

3.5 反馈回写与配置热加载

反馈回写是比较简单的，任务执行完毕后，把结果写进一个结构化日志文件，或者丢进消息队列由异步任务做聚合分析。这里给出一个极简的回写函数。

def write_feedback(profile: TaskProfile, selected_model: str, success: bool, latency_ms: int, error: Optional[str] = None): record = { "task_type": profile.task_type.value, "model_id": selected_model, "success": success, "latency_ms": latency_ms, "error": error, "ts": time.time(), } with open("feedback_log.jsonl", "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n")

后面定期跑一个离线任务，把这个 jsonl 文件聚合成“每个模型每个任务类型的平均成功率”，再更新到模型描述符的extra字段中，作为加权得分的一部分。这套方案不需要引入复杂的向量库，数据量小时简单文件 + 定时聚合就够；数据量大了再迁移到数据库即可。

配置热加载方面，不需要真的引入一堆外部依赖，一个简单的轮询文件变化的机制就够：

import os import hashlib from typing import Optional class ConfigHotLoader: def __init__(self, path: str): self.path = path self._version: Optional[str] = None self._data: dict = {} def reload_if_changed(self) -> bool: with open(self.path, "rb") as f: data = f.read() digest = hashlib.md5(data).hexdigest() if digest != self._version: self._version = digest self._data = json.loads(data.decode("utf-8")) return True return False def get_config_snapshot(self) -> dict: return self._data

启动时先调用一次reload_if_changed加载初始配置，然后放到一个后台定时任务里每 30 秒检查一次。与之前提到的一致，每个请求在执行开始时取一次快照，整条链路使用同一个快照，避免“半新半旧”的配置。

实际部署时，更有经验的方案是引入一个分布式配置中心，但内部项目完全没必要一上来就上重型组件，文件监听 + 快照版本已经足够清爽。

4. 踩坑实录与排查速查表

这一章我想把踩过的坑集中整理一下，做成一个可以快速对照的清单。没有这一章，前面的代码就只是理论。

4.1 高发问题实录

第一个坑，把全部工具和全部模型都塞给 Agent 自己选。这几乎是最常见的问题。我们当时天真地以为模型越强越能handle所有选项，结果模型选了完全无关的工具，还编了一个不存在的工具名。问题排查了很久才发现是“选项过载”，而不是模型的智商问题。解决方式就是前面反复提的候选集闸门。

第二个坑，反馈回路引起“模型垄断”。系统运行几天后，某个模型因为最开始表现不错，反馈权重越滚越高，导致几乎所有请求都涌向它，形成一个正反馈的“赢者通吃”局面。直到有一天该模型供应商限流，我们才发现整个系统已经毫无分散性了。后来加上 10% 随机探索比例才缓解。

第三个坑，工具描述写得不设边界。之前提过，工具描述只写功能不写边界，模型会把只读工具用在写操作场景。这里再补充一个细节：当时有个工具可以查用户信息和修改用户备注，我们在一个描述里含着两个操作，模型经常就只能看到“查用户”而看不到“改备注”，导致业务方收到用户的投诉。拆成两个工具并分别写清边界之后，问题立刻消失。

第四个坑，模型名与供应商版本号漂移。我们曾用代码里的硬编码模型名，比如“某通用模型V1”，但供应商悄悄升级了模型版本，同一个 model_id 在服务端的行为完全变了，Agent 还以为是同一个模型。后来在模型描述符里加入 vendor_version 字段，每次版本变化都当一个新模型注册，老模型的反馈权重不迁移到新版本，避免旧经验污染新模型。

第五个坑，热加载配置导致并发请求拿到不同配置。我们没有加快照版本时，线上出现了一个诡异的现象：同一个用户短时间内发出两个相同的任务请求，一次选模型 A，一次选模型 B，路径完全不同。排查了一天，才发现是配置热加载把工具白名单扩容了，前一个请求走了旧逻辑，后一个请求走了新逻辑。强行整请求绑定配置版本号后，行为就稳定了。

第六个坑，决策提示词里的角色设定影响选择逻辑。我们一开始用“你是全能助手”来让 Agent 做资源决策，结果发现它对任务的想象空间太大，经常无视候选集，输出一些看起来合理但清单里根本不存在的东西。改成“你是资源调度决策器，只做路由选型”这种约束型角色提示之后，情况好了很多。角色越具体，越不会越权。

4.2 五件自制件速查表

下面是我们在项目后期总结的一张速查表，基本就是一次踩坑后的标准操作手册。

有一点提醒一下：这五件自制件不是所有项目都需要第一周就全部上齐。如果你只是做个 Demo，任务意图路由器 + 候选集闸门就足够；一旦上生产，模型描述符注册表和配置热加载器就成了刚需，否则频繁改代码的上线节奏会让你痛不欲生。

5. 演进建议：这五件还能往哪个方向走

如果你按上面的思路把这套东西跑起来了，后续可以考虑做两件更有意思的事：一是把模型描述符里的打分权重视图化，跑几天数据之后，你能直观看到一个任务类型下各个模型的表现曲线；二是把反馈回路做成“暂停/冻结”机制，当 Agent 探测到历史反馈数据剧烈波动时，自动降低历史权重，进入一段观察期，而不是立刻信任最新反馈。

以我个人在实际操作中的体会来说，这套机制的复杂度控制是关键。五件自制件听起来很多，但每件的边界都很清晰，单个拆开都不复杂。真正需要花时间的不是写代码，而是定义清楚“什么任务该走什么候选集”“每个工具的边界描述怎么写规范”。这些文字层面的打磨，效果比调参还明显。

最后再分享一个小技巧：给反馈记录打上任务原始输入的 hash 值。排查线上路由问题时，能直接通过原始请求定位到那条反馈记录，省去大量大海捞针的时间。无论你最后怎么设计五件自制件，这个 hash 字段都建议保留。它不参与任何逻辑，但在你抓 bug 时真的能救命。