看到不少团队在推广“AI 模型路由器”这类方案时都会给出一个很夸张的成本下降数字比如“省了 94%”。这个数字确实让人心动但也容易让人误以为只要接一个中间层模型账单就能自动缩水。实际上模型路由器并不是什么黑魔法它的核心思路非常简单别再让所有请求都走同一个最贵的大模型了。真正有落地价值的地方在于如何把“路由”这件事设计得足够细让简单任务用便宜模型复杂任务才动用旗舰模型。本文将完整拆解一个模型路由器的设计与实现思路覆盖成本构成、路由策略、代码实现、缓存优化、质量兜底和工程落地注意事项。无论你是正在做 LLM 应用开发的工程师还是在为团队 API 账单发愁的负责人都能用这篇文章里的思路和代码搭出一个可运行的降本框架。1. 背景LLM 成本失控的根源在哪里1.1 先看懂大模型账单由什么构成大模型 API 计费通常不按调用次数而是按Token 数量和模型单价计费。一次请求往往分为输入 Token 和输出 Token 两部分不同服务商对两部分的定价可能不同而且输出 Token 往往比输入 Token 更贵。举个例子假设一个团队所有线上请求都调用某旗舰模型每次请求平均消耗 2000 个输入 Token、500 个输出 Token。按旗舰模型每百万 Token 输入 3 美元、输出 15 美元估算一次请求的成本大约是输入成本 2000 / 1000000 * 3 0.006 美元 输出成本 500 / 1000000 * 15 0.0075 美元 单次成本 ≈ 0.0135 美元看起来不多但如果每天有 100 万次请求一天就是 13500 美元一个月就是 40 万美元级别。而如果换成更便宜的模型同样请求的成本可能只有原来的几十分之一。问题就出在这里很多请求根本不需要这么强的模型。1.2 所有请求都用同一个模型是最大的浪费在实际业务中LLM 请求的类型差异极大。有的请求是“把这句话翻译成英文”有的请求是“帮我调试这段递归代码”还有的请求是“根据这份财务报告生成 10 条风险建议”。这些任务的难度完全不在一个量级但很多团队在接入初期为了省事会把所有请求都统一指向一个高能力模型。这样做的结果是简单任务支付了远高于必要水平的费用。模型上下文越长费用越高简单任务经常携带大量历史消息。高负载场景下旗舰模型限流更容易触发导致业务抖动。与其让模型自己消化所有任务不如在模型前面加一层“调度系统”让请求在到达模型之前先被判断出该用哪一档模型。1.3 模型路由器是什么模型路由器Model Router是一个位于业务代码与大模型 API 之间的中间层。它接收请求之后根据任务复杂度、输入长度、质量要求、当前预算等信号决定把请求转发给哪个模型。它解决的核心问题是在质量和成本之间找到动态平衡点。常见的路由信号包括信号说明示例任务类型区分翻译、摘要、分类、代码生成等翻译任务走小模型输入长度短文本与小模型更匹配长文本可能需要强模型超过 4000 Token 路由到大模型关键词规则命中“写代码”“推理”“计算”等模式命中复杂关键词时升级模型质量反馈小模型输出不合格时回退到大模型置信度低或校验失败时重试预算信号本月成本剩余越多越可以放宽模型选择月初适当多用强模型月底收紧模型路由器并不神秘它本质上是一个带策略的代理层。关键在于策略设计得是否合理以及是否具备兜底与监控机制。1.4 与微调、蒸馏、缓存方案的边界很多人在优化 LLM 成本时会同时听到微调Fine-tuning、蒸馏Distillation、缓存Caching这几个方案。它们和模型路由器并不冲突但解决的问题不同微调让模型更适应特定任务但模型单次推理成本不变。蒸馏训练一个小模型来模拟大模型输出训练成本高适合非常固定的场景。缓存对重复请求直接复用历史结果适合相似度高的流量。模型路由在请求入口处动态选模型不改变模型本身实施成本最低。模型路由通常作为“第一个落地的低成本优化手段”因为它不需要训练只需要写策略代码。等路由稳定后再叠加缓存、微调等手段可以进一步压缩成本。2. 模型路由器的核心设计思路2.1 分层设计模型池、路由策略、调用层一个可维护的模型路由器建议至少分成三个模块模型池Model Pool维护所有可用模型的元信息包括供应商、模型名、价格、最大 Token、超时时间。路由策略Routing Strategy负责给请求打分或分类输出“该走哪个模型”的决策结果。调用与兜底层Execution Layer负责真正调用模型、记录成本、处理失败回退。这种分层的好处是以后新增模型或调整价格时不需要动路由逻辑。比如某天你接入了更便宜的国产模型只需在模型池里加一条配置再在路由策略里把它挂到某个档位即可。2.2 输出“模型档位”而不是直接输出模型名很多初学者会把路由策略写死成“直接返回一个模型名”。但当你接入了 10 个模型之后策略会变得很难维护。更推荐的方式是路由策略先输出一个档位Tier例如easy、standard、advanced再由配置层把档位映射到具体模型。这样做有几个好处模型替换时不需要改路由代码。不同环境测试、生产可以配置不同的模型池。成本调整时只需要改价格表不需要动业务代码。2.3 路由结果不止是“选模型”模型路由器的决策维度还可以扩展。除了选择模型它还可以决定是否启用缓存。是否启用流式输出。是否压缩上下文。是否启用思考链Chain of Thought。是否进入 RAG 检索链路。例如一个知识库问答请求如果问题之前被问过路由器可以直接跳过模型调用返回缓存结果成本为 0。如果问题很复杂路由器还可以在调用模型之前先触发检索流程把检索结果拼进 prompt 再调用高配模型。这些能力让模型路由器不只是“省钱工具”更是整个 LLM 应用编排框架中的关键调度节点。3. 环境准备与基础架构3.1 技术栈说明为了实现一个可运行的模型路由器本文选择以下技术栈Python 3.9FastAPI用于暴露统一 HTTP 接口openai调用 OpenAI 系模型anthropic调用 Anthropic 系模型Redis可选用于缓存层JSON Lines用于本地成本日志存储版本不需要完全一致只要保证主要依赖为最新稳定版即可。示例代码的核心思路比版本更重要。3.2 项目目录结构建议按下面的目录结构组织代码llm-router/ ├── app/ │ ├── __init__.py │ ├── config.py # 模型配置与价格表 │ ├── router.py # 路由策略 │ ├── clients.py # LLM 调用封装 │ ├── cost_tracker.py # 成本统计 │ └── api.py # FastAPI 入口 ├── requirements.txt └── .env这样拆分后每个文件职责单一后续扩展也比较方便。3.3 环境变量配置在.env文件中写入 API KeyOPENAI_API_KEYsk-你的openai密钥 ANTHROPIC_API_KEYsk-ant-你的anthropic密钥实际项目中不要把这些密钥硬编码在代码里更不要把生产密钥提交到 Git 仓库。建议用环境变量或密钥管理服务管理。依赖文件requirements.txt内容如下fastapi uvicorn[standard] openai anthropic redis pydantic python-dotenv安装命令pip install -r requirements.txt4. 实现一个基础版 AI 模型路由器下面我们从一个最小可运行版本开始逐步实现完整的模型路由能力。4.1 定义模型配置与价格表价格表是模型路由器的核心配置。本文使用示例价格实际价格以各厂商官方报价为准。# app/config.py from dataclasses import dataclass, field from typing import Dict dataclass class ModelSpec: key: str model_name: str provider: str # openai / anthropic input_price: float # 每 100 万 tokens 输入价格美元 output_price: float # 每 100 万 tokens 输出价格美元 max_tokens: int 4096 temperature: float 0.7 dataclass class RouterConfig: models: Dict[str, ModelSpec] field(default_factorydict) cache_ttl: int 3600 enable_cache: bool True enable_fallback: bool True这里将模型池分成三档easy处理翻译、摘要、分类等简单任务。standard处理常规的生成、问答任务。advanced处理代码生成、推理、多步任务。对应模型配置如下def build_config() - RouterConfig: models { easy: ModelSpec( keyeasy, model_namegpt-4o-mini, provideropenai, input_price0.15, output_price0.6, ), standard: ModelSpec( keystandard, model_namegpt-4o, provideropenai, input_price2.5, output_price10.0, ), advanced: ModelSpec( keyadvanced, model_nameclaude-3-5-sonnet-latest, provideranthropic, input_price3.0, output_price15.0, ), } return RouterConfig(modelsmodels)这里需要注意claude-3-5-sonnet-latest是一个动态别名具体可用模型 ID 需要以 Anthropic 官方文档为准。价格也是示例价格实际接入时请填入你的真实采购价。4.2 实现路由策略路由策略是整个模型路由器最核心的模块。下面实现一个基于“评分制”的路由规则。# app/router.py from .config import RouterConfig class ModelRouter: def __init__(self, config: RouterConfig): self.config config def route(self, prompt: str, task_type: str ) - str: 根据 prompt 内容返回模型档位easy / standard / advanced score self._score(prompt, task_type) if score 0: return easy elif score 1: return standard else: return advanced def _score(self, prompt: str, task_type: str) - float: score 0.0 # 1. 根据任务类型判断 simple_tasks {translation, summarization, classification, keyword, rewrite} hard_tasks {code, debug, reasoning, multi_step, sql} if task_type in simple_tasks: score - 2 if task_type in hard_tasks: score 2 # 2. 根据 prompt 长度判断 length len(prompt) if length 4000: score 1 elif length 200: score - 1 # 3. 根据中文关键词判断 simple_patterns [翻译, 摘要, 分类, 提取关键词, 改写, 润色] hard_patterns [写代码, debug, 推理, 计算, 多步, SQL, 正则] for p in simple_patterns: if p in prompt: score - 1 break for p in hard_patterns: if p in prompt: score 1 break return score这个策略虽然简单但已经体现了路由决策的基本逻辑多信号融合。你可以根据业务特点把关键词列表替换成更准确的业务词表也可以把长度阈值调成适合你场景的数值。4.3 实现 LLM 调用封装有了路由决策下一步就是真正调用模型。我们需要一个统一的调用层屏蔽不同厂商 API 的差异。# app/clients.py import os import openai import anthropic from .config import ModelSpec class LLMClient: def __init__(self): self.openai_client openai.OpenAI(api_keyos.getenv(OPENAI_API_KEY)) self.anthropic_client anthropic.Anthropic(api_keyos.getenv(ANTHROPIC_API_KEY)) def call(self, spec: ModelSpec, prompt: str): if spec.provider openai: return self._call_openai(spec, prompt) elif spec.provider anthropic: return self._call_anthropic(spec, prompt) else: raise ValueError(fUnsupported provider: {spec.provider}) def _call_openai(self, spec: ModelSpec, prompt: str): resp self.openai_client.chat.completions.create( modelspec.model_name, messages[{role: user, content: prompt}], max_tokensspec.max_tokens, temperaturespec.temperature, ) content resp.choices[0].message.content return content, resp.usage.prompt_tokens, resp.usage.completion_tokens def _call_anthropic(self, spec: ModelSpec, prompt: str): resp self.anthropic_client.messages.create( modelspec.model_name, max_tokensspec.max_tokens, temperaturespec.temperature, messages[{role: user, content: prompt}], ) content .join([block.text for block in resp.content if block.type text]) return content, resp.usage.input_tokens, resp.usage.output_tokens注意Anthropic 的messages.create参数里max_tokens是必填参数这一点和 OpenAI 不同。不同厂商 API 的返回结构也不一致统一封装后业务层就不需要关心这些细节了。4.4 实现成本统计与日志降本方案如果无法量化就无法证明它有效。所以成本统计模块是必不可少的。# app/cost_tracker.py import json import time from dataclasses import dataclass, asdict dataclass class CostRecord: ts: float trace_id: str model_key: str model_name: str input_tokens: int output_tokens: int cost_usd: float latency_ms: float fallback: bool class CostTracker: def __init__(self, pathcost_records.jsonl): self.path path def append(self, record: CostRecord): with open(self.path, a, encodingutf-8) as f: f.write(json.dumps(asdict(record), ensure_asciiFalse) \n) def stats(self): total_cost 0.0 total_requests 0 model_counter {} try: with open(self.path, encodingutf-8) as f: for line in f: data json.loads(line) total_requests 1 total_cost data[cost_usd] key data[model_key] model_counter[key] model_counter.get(key, 0) 1 except FileNotFoundError: pass return { total_requests: total_requests, total_cost_usd: round(total_cost, 6), model_counter: model_counter, }这里使用 JSON Lines 格式逐行追加日志。虽然是本地文件方案但在中小流量的场景下已经够用。流量更大时可以把这段逻辑改成写入 ClickHouse、Elasticsearch 或云日志服务。4.5 用 FastAPI 暴露统一接口有了路由、调用、统计三个模块就可以用 FastAPI 把它们串起来。# app/api.py import time import uuid from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from .config import RouterConfig, ModelSpec from .router import ModelRouter from .clients import LLMClient from .cost_tracker import CostTracker, CostRecord def build_config() - RouterConfig: models { easy: ModelSpec( keyeasy, model_namegpt-4o-mini, provideropenai, input_price0.15, output_price0.6, ), standard: ModelSpec( keystandard, model_namegpt-4o, provideropenai, input_price2.5, output_price10.0, ), advanced: ModelSpec( keyadvanced, model_nameclaude-3-5-sonnet-latest, provideranthropic, input_price3.0, output_price15.0, ), } return RouterConfig(modelsmodels) config build_config() router ModelRouter(config) client LLMClient() tracker CostTracker() app FastAPI(titleAI Model Router, version0.1.0) class ChatRequest(BaseModel): prompt: str Field(..., min_length1) task_type: str class ChatResponse(BaseModel): model_key: str model_name: str content: str cost_usd: float latency_ms: float trace_id: str fallback: bool False app.post(/v1/chat, response_modelChatResponse) def chat(req: ChatRequest): trace_id uuid.uuid4().hex[:12] model_key router.route(req.prompt, req.task_type) spec config.models[model_key] start_time time.time() try: content, input_tokens, output_tokens client.call(spec, req.prompt) fallback False except Exception as e: if not config.enable_fallback: raise HTTPException(status_code502, detailstr(e)) fallback True fallback_key advanced spec config.models[fallback_key] try: content, input_tokens, output_tokens client.call(spec, req.prompt) except Exception as e2: raise HTTPException(status_code502, detailfall models failed: {e2}) cost (input_tokens / 1_000_000) * spec.input_price (output_tokens / 1_000_000) * spec.output_price latency_ms (time.time() - start_time) * 1000 record CostRecord( tstime.time(), trace_idtrace_id, model_keymodel_key if not fallback else advanced, model_namespec.model_name, input_tokensinput_tokens, output_tokensoutput_tokens, cost_usdcost, latency_mslatency_ms, fallbackfallback, ) tracker.append(record) return ChatResponse( model_keymodel_key if not fallback else advanced, model_namespec.model_name, contentcontent, cost_usdcost, latency_mslatency_ms, trace_idtrace_id, fallbackfallback, ) app.get(/v1/stats) def stats(): return tracker.stats()4.6 运行与验证启动服务uvicorn app.api:app --host 0.0.0.0 --port 8000 --reload发送一个简单翻译请求curl -X POST http://127.0.0.1:8000/v1/chat \ -H Content-Type: application/json \ -d {prompt: 把这句话翻译成英文今天天气很好, task_type: translation}预期会路由到easy档位。再发送一个复杂请求curl -X POST http://127.0.0.1:8000/v1/chat \ -H Content-Type: application/json \ -d {prompt: 请写一个快速排序算法并解释时间复杂度, task_type: code}预期会路由到advanced档位。查看成本统计curl http://127.0.0.1:8000/v1/stats返回示例{ total_requests: 2, total_cost_usd: 0.002812, model_counter: { easy: 1, advanced: 1 } }这个最小版本已经能完整跑通“路由 - 调用 - 统计”的全链路。5. 进阶优化把成本压得更低基础版模型路由器解决了“选模型”的问题但距离“成本下降 90% 以上”还有很大距离。要让成本进一步下降还需要在以下几个方向做优化。5.1 缓存优先相同请求不再重复调用在很多业务场景中用户的请求重复率非常高。比如“帮我总结这篇文章”内容相同或相似时完全没有必要重复调用大模型。引入 Redis 缓存后请求流程变成计算 prompt 的哈希值。如果缓存中存在直接返回缓存结果成本为 0。如果没有命中再执行路由与模型调用。调用成功后把结果写入缓存并设置 TTL。示例缓存代码思路如下import hashlib import redis import json redis_client redis.Redis(hostlocalhost, port6379, db0) def make_cache_key(prompt: str, model_key: str) - str: raw f{model_key}:{prompt}.encode(utf-8) return llm: hashlib.sha256(raw).hexdigest() def get_cache(prompt: str, model_key: str): key make_cache_key(prompt, model_key) value redis_client.get(key) if value: return json.loads(value) return None def set_cache(prompt: str, model_key: str, content: str, ttl: int 3600): key make_cache_key(prompt, model_key) redis_client.set(key, json.dumps({content: content}, ensure_asciiFalse), exttl)需要注意的是缓存不要盲目对所有请求开启。涉及用户隐私、实时数据、个性化内容时需要谨慎设置 TTL 或直接关闭缓存。5.2 语义路由让模型自己判断任务复杂度关键词规则虽然简单但覆盖不全面。比如“帮我看看这段代码为什么报错”并不包含“代码”这个词但显然是一个复杂任务。更高级的路由方式是语义路由。做法是提前准备一批典型的简单任务样本和复杂任务样本。对样本和当前请求做 embedding 向量化。计算当前请求与两批样本的相似度。相似度更接近哪一边就路由到对应模型档位。这种方式比关键词规则更鲁棒但需要引入 embedding 模型和向量数据库实施成本会高一些。适合流量大、任务类型丰富的场景。你还可以让一个便宜模型先做“任务复杂度分类”比如你是一个任务分类器。判断下面的用户请求属于简单任务还是复杂任务。 简单任务包括翻译、摘要、分类、闲聊、简单问答。 复杂任务包括代码调试、数学推理、长文档分析、多步规划。 只输出 simple 或 complex。 用户请求{{prompt}}用gpt-4o-mini做这个分类即使分类本身有少量误判也比直接把所有请求都发给旗舰模型要省钱得多。5.3 质量兜底路由决策允许“反悔”路由决策不可避免存在误判。有时候一个小模型被派去处理复杂任务输出质量明显不行。这时候需要引入质量兜底机制。常见的兜底方式有两种自评兜底让小模型生成答案之后再用一个便宜模型评估答案是否完整、是否解决了用户问题。如果评估不通过重新调用高配模型。规则兜底对输出做规则校验比如代码能否编译、JSON 是否符合格式、关键字段是否缺失。校验失败时触发升级调用。兜底会增加延迟和调用次数所以不能对所有请求启用。建议只在任务类型本身属于“高价值场景”时才启用例如金融分析、代码生成、法律咨询等。5.4 预算约束与负载控制模型路由器的另一个优势是可以在入口处控制整体成本。例如你可以给系统设置“每日成本预算”。当今日成本接近上限时自动把策略调成“更省钱模式”让更多请求走小模型或直接返回缓存。这种控制力度是单个模型 API 无法做到的。实现思路很简单在路由策略中增加一个预算上下文class BudgetAwareRouter: def __init__(self, config: RouterConfig, daily_budget: float): self.config config self.daily_budget daily_budget self.today_cost 0.0 def route(self, prompt: str, task_type: str ) - str: base_tier super().route(prompt, task_type) if self.today_cost self.daily_budget * 0.8: # 预算紧张时所有请求降一档 if base_tier advanced: return standard if base_tier standard: return easy return base_tier这里的预算值要根据你的业务流量动态调整核心思路是让路由策略具备“成本感知”能力。6. 常见问题与排查思路模型路由器上线后通常会遇到以下几类问题。下面把它们整理成一张排查表方便你对照处理。问题现象常见原因解决思路简单请求被路由到高级模型关键词表覆盖不全prompt 命中复杂关键词检查路由日志补充简单任务词表或增加“任务类型”入参高级模型频繁触发回退小模型能力不足输出质量差优化兜底条件或把该任务类型默认路由到标准档缓存命中率极低请求变化大或多轮对话上下文携带大量 unique 信息只缓存单轮关键字段用归一化后的 prompt 做缓存 key成本统计与账单不一致记账价格与实际采购价不一致定期从云厂商账单拉取真实单价校准配置表接口返回超时小模型排队或模型 API 限流增加超时配置失败时快速回退不要长时间阻塞不同环境路由结果不一致测试环境模型池与生产环境不一致统一模型池配置使用同一套配置管理机制其中最容易忽略的是第一点。关键词规则虽然简单但会误伤。建议上线初期先把所有路由日志保留下来每天分析一次误判样本持续迭代词表。7. 工程落地最佳实践如果你准备把模型路由器应用到生产环境以下几个实践建议可以直接参考。7.1 先审计再改造不要急于把全部流量切换到模型路由器。先做两件事统计当前所有请求的任务类型分布。随机抽取一定比例的请求人工或半自动地标记“理想模型档位”。基于这份样本才能评估路由策略的准确率。建议先用 1% 到 5% 的流量跑影子模式也就是只记录“路由决策结果”但不按结果去调用模型。等策略稳定后再逐渐放开流量。7.2 监控指标要选对模型路由器的核心指标不只是成本还包括质量。建议至少监控各档位请求占比。单日总成本与单请求平均成本。缓存命中率。模型调用失败率与回退率。平均响应延迟。用户侧质量反馈或点赞点踩比例。其中质量指标是底线。如果一个路由策略把成本降了 90%但用户满意度也降了 50%那这个方案是不可持续的。建议在关键业务场景同时采集质量信号。7.3 灰度发布策略模型路由器本身是一个策略系统策略改动不需要重启服务只要让配置支持热更新即可。灰度发布可以按这样的节奏推进先用影子模式评估策略。再放 5% 流量观察成本和质量指标 24 小时。没有问题扩大到 30%。稳定后再扩大到 100%。如果发现质量下滑第一时间把策略回退到上一版本。7.4 安全与合规注意调用第三方大模型 API 时要特别注意数据安全涉及用户隐私、企业机密的 prompt不要路由到外部模型。如果要使用外部模型需要保证链路做了脱敏处理。涉及数据库操作、代码执行、支付等敏感动作时都要先经过合法授权和严格校验不能让模型直接控制生产系统。另外不要因为模型便宜就放松对输出的检查。无论使用哪个档位的模型输出都应该经过内容安全过滤和业务规则校验。7.5 成本报表要按业务线拆分成本优化能不能持续很大程度上取决于数据能否被看见。建议按以下维度拆分成本报表业务线。模型档位。请求来源。任务类型。时间粒度小时 / 天 / 周。这样每次优化动作都能直接对应到数字变化。比如“上周把客服场景的摘要任务迁移到 easy 档成本下降了 12%”这类结论只有在你把成本数据拆得足够细时才能得出。8. 总结与下一步模型路由器的核心价值不是用一个“高级算法”来代替模型而是在业务请求与大模型之间建立一道有策略的闸门。它让简单任务不再浪费旗舰模型的计算资源让复杂任务仍然可以获得足够的模型能力并在质量与成本之间动态调整。本文实现了从配置、路由策略、调用封装、成本统计到 HTTP 接口的完整最小系统。你可以直接运行这份代码也可以通过扩展语义路由、缓存、预算控制等能力把它接入真实业务链路。下一步建议按这个方向继续深入如果业务场景比较固定可以尝试用少量已标注数据训练一个轻量分类模型替代关键词规则。如果请求重复率高优先优化缓存策略这是成本下降最快、风险最低的一步。如果团队有本地 GPU 资源可以在模型池中混入本地部署的开源模型进一步拉低中低档任务成本。如果业务涉及 Agent、RAG 等多跳链路可以把模型路由器嵌入现有的 LLM 编排框架中让每一跳调用都走路由决策。一个成功的降本项目不是一上来就追求“省 94%”而是先把每笔调用成本记录清楚再逐步让路由策略变得更聪明。当你把“成本”和“质量”两套指标都掌握在手里时模型路由器的优势才会真正显现出来。如果你正在设计自己的模型路由器建议从本文这个最小版本开始跑通链路再结合业务数据不断优化策略。先跑数据再谈节约这是最稳妥的路径。