直接说结论这次我们要拆解的项目不是某个新模型也不是某个推理框架而是一种能在生产环境里显著降低 LLM 调用成本的工程方案AI 模型路由器Model Router。原始项目帖子给出的收益是“cut LLM costs by 94%”也就是 LLM 成本降低 94%。这个数字来自特定业务负载和模型价差下的实测结果不同场景复现时会有浮动但核心思路是通用的不是所有请求都需要让最强、最贵的模型来处理。把简单的查询路由到便宜的小模型把复杂任务交给旗舰模型总成本就会大幅下降。这篇文章会从工程角度完整拆解模型路由器的核心能力、路由策略、成本控制原理、部署方式、接口设计与批量任务接入。你会看到为什么“按难度分流 自动升级 答案缓存”能带来数量级的成本变化也会拿到一套可以直接改造成内部工具的代码骨架。适合已经在用 OpenAI、Claude、DeepSeek、Qwen 等模型 API 的开发者也适合正在做 LLM 应用架构选型、想要控制推理成本的技术负责人。1. 核心能力速览先看这个方案的整体规格。模型路由器本质上是一个位于业务和 LLM 提供商之间的智能转发层它接收上游请求判断任务复杂度和质量要求然后选择最合适的模型去执行。能力项说明项目类型LLM 请求路由 / 模型网关 / 成本控制中间层核心功能按任务难度路由、低配模型兜底、失败自动升级、响应缓存、调用日志成本改善原帖在特定场景下实现 94% 成本下降实际收益取决于负载分布和模型价差延迟影响增加少量路由判断开销毫秒级但可降低排队和超时成本支持模型理论支持所有 OpenAI 兼容 API、Claude API、本地 vLLM / Ollama 服务接口能力可作为独立 HTTP 服务对外暴露统一的/v1/chat/completions风格接口批量任务支持离线批处理队列可在低峰期执行低成本模型任务依赖环境Python 3.10、FastAPI、Redis可选、OpenAI SDK 或其他 HTTP 客户端部署方式容器或裸机运行与业务服务同区域部署可进一步降低转发延迟适合场景客服、内容分类、摘要、RAG 问答、意图识别、批量审核等混合任务负载从表里能看出来这个项目的重点不在模型本身而在“如何把请求分配得更聪明”。它解决的是规模化调用 LLM 时最现实的问题——成本和收益之间的平衡。2. 成本降低 94% 的原理拆解很多团队一上来就盯着最强模型跑所有业务结果一个月下来账单非常可观。实际上大多数 LLM 应用里请求的难度服从长尾分布大量请求是简单的分类、抽取、改写、短问答少数请求才需要深度推理、长上下文理解、复杂代码生成。如果所有请求都走旗舰模型等于用跑车的成本去送外卖。模型路由器降本的核心原理可以拆成四条链路。2.1 难度分层便宜的模型先接路由器维护一个模型优先级表每个模型都有各自的擅长领域和价格档位。接收到请求后它先做一次快速预判。比如判断任务是否有明确指令模板、是否包含多步推理、是否需要最新的知识库信息。如果预判结果是“简单任务”就直接发给小模型比如 DeepSeek-V3、Qwen-Turbo 这类价格便宜的模型。只有预判为复杂任务时才升级到 GPT-4o、Claude Opus 这类旗舰模型。2.2 自动升级小模型失败就换大模型单纯按规则分发并不够。小模型偶尔会读不懂用户意图或者输出的 JSON 格式不合法、答案不符合约束条件。路由器会加一层质量校验校验失败后自动把同一个请求转发给更强的模型。这种“先试低价模型失败再升级”的策略在多数场景下可以显著降低平均成本。2.3 相似答案缓存相同问题不重复计费很多业务里用户会反复问同一个问题。路由器可以按问题的语义哈希或嵌入向量做缓存命中后直接返回历史答案不再调用任何模型 API。缓存本身是降本最快的一步尤其适合客服 FAQ、政策问答、固定知识库检索场景。2.4 批量任务调度把任务放到更便宜的时段和模型对非实时任务路由器可以放入队列在业务低峰期批量调用低成本模型甚至可以切换离线推理。批量处理能提高模型调用吞吐也能减少对在线服务的依赖进一步摊薄单次成本。这里的核心观点是成本下降 94% 不是把模型换成免费的而是让模型调用次数变少、平均单价变低、失败重试变得更加可控。如果业务负载结构不同收益也会不同。比如全部请求都是高难度代码生成路由器的降本空间会小很多。3. 适用场景与使用边界模型路由器不是银弹它适合的任务有几类明显特征。判断你的场景是否符合可以从下面三点入手。第一任务难度有明显差异。比如客服系统里既有“查订单状态”这种简单意图又有“帮忙分析售后纠纷责任”这种复杂问题。路由器可以给前者分配小模型给后者分配大模型。第二大多数请求允许 1 到 3 秒的额外排队和路由时间。第三业务可以接受“先便宜后升级”的失败模式而不是每次都要一次到位。不适合路由器的场景包括所有请求都是硬性高难度任务业务对回复质量要求极高且不能容忍任何小模型输出强依赖单一模型的私有人格和专属能力离线环境下没有可选的多模型池。合规边界方面必须注意几点。使用模型路由时请求会转发给多个提供商要在隐私协议里明确告知用户数据处理范围。涉及敏感身份信息、医疗数据、未成年人数据时不应发送到第三方模型服务。涉及人脸、声音、版权素材的生成类任务必须确认授权链条完整。公司内部使用时要做好数据脱敏和日志脱敏避免 prompt 中的密钥、手机号、身份证号直接落盘。4. 整体架构与关键设计模型路由器的架构通常分成四层接入层、路由决策层、执行层、可观测层。业务服务 → API 网关 / 模型路由器 → 路由决策模块 → 模型提供商 / 本地推理 ↓ 缓存服务 日志 指标接入层负责对外提供统一接口兼容 OpenAI 的/v1/chat/completions格式业务方无需改动代码即可切换。路由决策层是核心它根据规则、上下文长度、历史反馈、成本预算来判断把请求分给谁。执行层封装了多个模型提供商的 HTTP 调用统一重试、超时和鉴权。可观测层负责记录每次调用的模型、token 数、成本、延迟和失败原因。关键设计点是路由决策模块不能太重。如果为了判断复杂度先调用一个大模型那路由本身就可能比直接调小模型还贵。所以路由判断要尽量使用轻量规则指令模板匹配通过正则匹配请求里是否包含明确的复杂任务指令比如“写代码”“解释原理”“对比方案”。文本长度判断超过某个 token 阈值就默认走长文本强模型。工具调用检测如果请求里带了多条工具调用记录或较长的 function calling 上下文需要更强模型。历史行为反馈同一个用户或同一个任务 ID 的历史失败率如果偏高就直接升级模型。关键词置信度内部关键词模型对任务类型打标签置信度高走小模型置信度低走大模型。这套规则都是白盒、可控、可测试的不会引入额外的模型成本。5. 环境准备与前置条件要在本地把模型路由器跑起来先准备一套干净的环境。推荐配置如下。依赖项版本 / 说明Python3.10FastAPI用于搭建 HTTP 服务uvicornASGI 服务器openai官方 SDK兼容多种提供方接口httpx异步 HTTP 客户端redis-py可选用于共享缓存pydantic配置校验loguru 或 structlog结构化日志操作系统不限Windows、Linux、macOS 都可以跑。如果只是接入云厂商 API不需要 GPU如果要接入本地 vLLM、Ollama则需要按对应的模型显存要求准备显卡。建议第一次测试时只接两家 API跑通路由逻辑后再扩展模型池。创建项目目录和虚拟环境mkdir llm-model-router cd llm-model-router python -m venv venv source venv/bin/activate pip install fastapi uvicorn openai httpx redis pydantic loguru这里没有把具体版本写死因为不同环境下 Python 版本和依赖冲突情况不一样。建议使用 requirements.txt 固定版本。6. 核心实现搭建模型路由器服务下面给出一套可运行的最小模型路由器实现骨架。它包含四个模块模型池定义、路由判断、模型调用转发、HTTP 服务。6.1 模型池配置先定义一个数据类来描述不同的模型服务。每个模型都有名称、接口地址、API Key、价格档位和支持能力。from dataclasses import dataclass from typing import List dataclass class ModelSpec: name: str api_type: str # openai / anthropic / vllm base_url: str api_key: str price_per_1k_input: float 0.0 price_per_1k_output: float 0.0 max_context: int 8192 tags: List[str] None# models.py MODEL_POOL { gpt-4o: ModelSpec( namegpt-4o, api_typeopenai, base_urlhttps://api.openai.com/v1, api_key$OPENAI_API_KEY, price_per_1k_input0.005, price_per_1k_output0.015, max_context128000, tags[complex, code, reasoning], ), deepseek-chat: ModelSpec( namedeepseek-chat, api_typeopenai, base_urlhttps://api.deepseek.com/v1, api_key$DEEPSEEK_API_KEY, price_per_1k_input0.001, price_per_1k_output0.002, max_context64000, tags[fast, cheap, general], ), qwen-turbo: ModelSpec( nameqwen-turbo, api_typeopenai, base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, api_key$QWEN_API_KEY, price_per_1k_input0.0005, price_per_1k_output0.002, max_context128000, tags[cheap, classify, extract], ), }注意这里只给了配置示例具体 API Key 和生产版本价格需要用实际环境替换。生产环境中不要直接写在代码里建议用环境变量或密钥管理服务。6.2 路由决策模块路由决策模块是核心。它接收一个 Chat 请求输出应该使用哪个模型。先做规则判断再做语义标签判断最后给默认值。import re from typing import Dict from dataclasses import dataclass dataclass class RouteDecision: model_name: str reason: str upgraded: bool False COMPLEX_PATTERNS [ r写代码|代码, r解释|原理|为什么, r对比|比较|区别, r设计|架构|方案, rdebug|调试|报错, ] def judge_complexity(prompt: str) - tuple[bool, str]: 基于规则判断任务复杂度不调用任何模型。 prompt_lower prompt.lower() for pattern in COMPLEX_PATTERNS: if re.search(pattern, prompt_lower): return True, pattern # 长度超过阈值默认当作复杂任务 if len(prompt) 3000: return True, long_context return False, simple def route_decision(prompt: str, user_params: Dict None) - RouteDecision: complex_task, pattern judge_complexity(prompt) if user_params and user_params.get(force_model): return RouteDecision( model_nameuser_params[force_model], reasonuser_override ) if complex_task: return RouteDecision( model_namegpt-4o, reasonfcomplex_pattern:{pattern} ) # 简单任务走便宜模型 return RouteDecision( model_nameqwen-turbo, reasonsimple_rule )这个模块最大的特点是零成本。它只用正则和长度判断不引入额外的模型调用。6.3 模型调用与自动升级接下来实现执行层。调用模型时统一使用异步 HTTP 客户端。如果小模型失败或返回内容不满足校验条件自动升级到强模型。import httpx import asyncio import json from typing import List class ModelClient: def __init__(self): self.client httpx.AsyncClient(timeout60) async def chat_completion( self, model_spec, messages: List[Dict], temperature: float 0.3, max_tokens: int 1024 ): if model_spec.api_type ! openai: raise ValueError(unsupported api_type) headers { Authorization: fBearer {model_spec.api_key}, Content-Type: application/json, } payload { model: model_spec.name, messages: messages, temperature: temperature, max_tokens: max_tokens, } resp await self.client.post( f{model_spec.base_url}/chat/completions, headersheaders, jsonpayload, ) resp.raise_for_status() data resp.json() return data[choices][0][message][content] async def close(self): await self.client.aclose() async def execute_with_upgrade(messages: List[Dict]): 先走便宜模型失败或校验失败时升级到旗舰模型。 client ModelClient() try: prompt messages[-1][content] if messages else decision route_decision(prompt) # 第一次调用使用路由决策的模型 primary_model MODEL_POOL[decision.model_name] try: result await client.chat_completion(primary_model, messages) if validate_output(result): return { model: primary_model.name, content: result, upgraded: False, reason: decision.reason, } except Exception as exc: logger.warning(fprimary model failed: {primary_model.name}, error{exc}) # 升级调用 fallback_model MODEL_POOL[gpt-4o] result await client.chat_completion(fallback_model, messages) return { model: fallback_model.name, content: result, upgraded: True, reason: auto_upgrade, } finally: await client.close()validate_output 函数可以根据业务定义比如检查返回内容是否包含指定字段、JSON 是否可解析、长度是否达标。这里是典型的“廉价模型优先 失败升级”模式。6.4 缓存模块缓存可以减少重复请求。最简单的实现是缓存的哈希对普通文本可以解决问题。稍微复杂一点的版本可以接 Redis。import hashlib import json class SimpleCache: def __init__(self): self._store {} def _key(self, messages: List[Dict], model: str) - str: raw json.dumps(messages, ensure_asciiFalse) |v1 return hashlib.sha256(raw.encode()).hexdigest() def get(self, messages: List[Dict], model: str): key self._key(messages, model) return self._store.get(key) def set(self, messages: List[Dict], model: str, content: str, ttl_seconds: int 3600): key self._key(messages, model) self._store[key] { content: content, expire_at: time.time() ttl_seconds, } def get_or_set(self, messages, model, fallback): cached self.get(messages, model) if cached and cached[expire_at] time.time(): return cached[content] content fallback() self.set(messages, model, content) return content更合理的做法是同时把缓存键拆成“用户输入 系统提示词 温度”。温度会影响输出的稳定性如果业务允许建议把温度固定为 0 或 0.1提高缓存命中率。6.5 HTTP 服务最后把服务暴露成 OpenAI 兼容接口。业务方只需要改一个 base_url 就能接入。from fastapi import FastAPI, Request from pydantic import BaseModel, Field from typing import List, Dict, Optional app FastAPI(titleModel Router Service, version0.1.0) class ChatMessage(BaseModel): role: str content: str class ChatCompletionRequest(BaseModel): model: Optional[str] auto messages: List[ChatMessage] temperature: float 0.3 max_tokens: int 1024 stream: bool False metadata: Optional[Dict] None class ChatCompletionResponse(BaseModel): id: str object: str chat.completion model: str choices: list usage: dict app.post(/v1/chat/completions) async def chat_completions(req: ChatCompletionRequest): messages [m.dict() for m in req.messages] cache SimpleCache() cached cache.get(messages, routed) if cached: return build_response(cache, cached, usage{hit: True}) result await execute_with_upgrade(messages) cache.set(messages, result[model], result[content]) return build_response(result[model], result[content], usageresult)这里简化了响应格式。生产环境建议把响应结构做成和 OpenAI 官方一致这样上游迁移成本最低。7. 接口 API 与批量任务接入服务跑起来后业务方调用方式非常直接。先启动服务uvicorn main:app --host 127.0.0.1 --port 8000启动后可以用 curl 验证curl http://127.0.0.1:8000/v1/chat/completions \ -H Content-Type: application/json \ -d { model: auto, messages: [{role: user, content: 把下面这句话翻译成英文今天天气很好}], temperature: 0.3 }返回结果会标注实际使用的模型名方便核对路由行为{ id: chatcmpl_xxx, object: chat.completion, model: qwen-turbo, choices: [ { index: 0, message: { role: assistant, content: The weather is very nice today. }, finish_reason: stop } ], usage: {} }Python 调用端也兼容 OpenAI 官方 SDK只需要覆盖 base_urlfrom openai import OpenAI client OpenAI( api_keytest-key, base_urlhttp://127.0.0.1:8000/v1 ) response client.chat.completions.create( modelauto, messages[{role: user, content: 这篇文档讲了什么}], temperature0.2 ) print(response.choices[0].message.content)7.1 批量任务设计批量任务场景下路由器要提供两个能力一个是可以自定义每个任务的模型优先级另一个是支持失败重试。推荐的批量任务格式{ tasks: [ { task_id: 001, mode: auto, fallback_order: [deepseek-chat, gpt-4o, qwen-turbo], messages: [ {role: user, content: 判断下面这段评论是好评还是差评发货很快质量很好} ], temperature: 0.1, max_tokens: 200 } ] }批量任务调度器可以基于 asyncio 队列实现。核心调度逻辑读取任务列表。每个任务执行路由决策。如果任务失败按 fallback_order 依次重试。记录每个任务的实际模型、token 消耗、耗时和错误信息。全部完成后输出结果 CSV 或 JSON 文件。一个简化版批量处理脚本import asyncio import json async def process_batch(input_file: str, output_file: str): with open(input_file, r, encodingutf-8) as f: batch json.load(f) results [] for task in batch[tasks]: try: result await execute_with_upgrade(task[messages]) results.append({ task_id: task[task_id], status: ok, model: result[model], content: result[content], reason: result[reason], }) except Exception as exc: results.append({ task_id: task[task_id], status: error, error: str(exc), }) with open(output_file, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) if __name__ __main__: asyncio.run(process_batch(tasks.json, results.json))批量处理时要特别关注 Token 配额和限流。很多模型 API 有每分钟请求数限制建议在批量处理器里加延迟和退避重试。8. 成本测算与效果验证要验证路由器到底省了多少钱需要在路由前后各做一轮统计。最关键的有效指标是“每 1000 次请求的实际计费金额”而不是只看总 token 数。8.1 成本记录每次调用模型后路由器都应该把以下信息写入日志request_id任务类型标签实际使用的模型输入 token 数输出 token 数预估成本是否从缓存命中是否经过升级路由决策原因请求耗时日志结构示例{ request_id: req_001, task_type: classification, model: qwen-turbo, prompt_tokens: 120, completion_tokens: 32, estimated_cost_usd: 0.000064, cache_hit: false, upgraded: false, route_reason: simple_rule, latency_ms: 420 }有了日志以后按天聚合成本SELECT model, COUNT(*) AS requests, SUM(estimated_cost_usd) AS total_cost, SUM(CASE WHEN upgraded THEN 1 ELSE 0 END) AS upgrade_count, SUM(CASE WHEN cache_hit THEN 1 ELSE 0 END) AS cache_hit_count FROM router_usage_logs GROUP BY model ORDER BY total_cost DESC;8.2 对比测试方法比较合理的验证流程分成三步。第一步用原来的单一模型方案处理一万条真实业务请求记录总费用和质量指标。第二步开启路由器用相同的请求集重放记录新的总费用和质量指标。第三步对比两组数据。需要对比的指标至少包括五个指标单一模型方案模型路由器方案变化总成本美元基线路由后下降幅度每千请求成本基线路由后核心指标平均响应延迟基线路由后需观察答案通过率 / 质检通过率基线路由后不能明显下降高优先级任务失败率基线路由后建议低于基线对日志口径特别提醒实际账单可能还会包含平台服务费、网络传输费、额外 token 溢出、缓存计算开销等。如果要做预算汇报最好把“估算成本”和“实际账单”分开列出来。8.3 为什么可以达到 94%从原帖标题看这个路由器把成本砍掉了 94%。在真实数据中要做到这种量级通常需要三个条件同时成立一是任务池里有大量简单任务例如超过 90% 的请求只是问答分类或文本抽取二是简单任务和复杂任务模型单价相差 5 到 20 倍三是缓存命中率能稳定在 30% 以上。三个条件叠加总成本下降确实可能接近 90% 以上。如果业务负载全是复杂代码推理那成本下降幅度会小很多。所以部署前先抽样统计你自己的请求分布这是最重要的一步。9. 性能观察与资源占用模型路由器本身是一个非常轻的代理层。它的 CPU 和内存占用主要来自请求转发、JSON 序列化和缓存读写。在没有启用本地大模型的情况下不需要 GPU。单独运行时内存占用通常在 200MB 到 500MB 之间具体取决于并发连接数和缓存规模。性能观察建议关注以下指标路由决策耗时正则匹配一般小于 1ms如果引入关键词模型分类单请求会增加 10 到 100ms。代理层转发延迟每增加一层转发会多 1 到 5ms 网络开销如果服务部署在模型 API 同一区域影响可忽略。缓存命中率命中率越高平均响应延迟越低。理论上有缓存时延迟可能低于直接调用成本模型。超时设置模型 API 超时建议设置在 30 到 120 秒之间。路由器本身的上游超时要比业务超时短避免连接堆积。避免端口冲突和进程残留是本地部署最容易踩的坑# 查看端口占用 lsof -i :8000 # 如果端口被占用杀掉旧进程 kill -9 $(lsof -t -i :8000) # 或用不同端口启动 uvicorn main:app --host 127.0.0.1 --port 8001生产环境建议用 systemd 或 Docker Compose 管理服务进程增加健康检查和自动重启。10. 常见问题与排查方法模型路由器在部署和运行中会遇到一些固定问题排查思路基本可以复用。问题现象可能原因排查方式解决方案启动后接口 404路由路径未正确注册检查 FastAPI 路由装饰器确认/v1/chat/completions路径与调用一致请求一直超时上游模型 API 不可用或网络被限制单独 curl 上游接口检查 API Key、网络策略、代理设置请求全部走了大模型路由规则误判复杂任务查看 route_reason 日志调整正则或阈值检查是否误传 force_model缓存命中率低消息里带时间戳或随机参数对比文本缓存 key清理无关参数再生成缓存 key小模型返回格式不合格模型没有遵循 JSON 约束查看全量返回日志在 prompt 中增加示例或失败后升级模型批量任务中途卡死没有设置超时或重试查看任务日志为每个任务配置超时和退避重试成本没有下降复杂任务占比过高按 task_type 统计成本优化简单任务 prompt提升缓存命中使用本地 vLLM 部署时报错base_url 不是 OpenAI 兼容格式检查 vLLM 的 service 地址确认/v1路径正确模型名一致API Key 泄露密钥写在代码仓库扫描日志和代码改用环境变量轮换密钥日志脱敏11. 最佳实践与使用建议模型路由器能做到成本优化但也要注意质量和成本之间的动态平衡。下面这几条是工程上验证过的建议。11.1 先跑通最小链路第一次接入时不要直接上全量业务。先拿 1000 条真实请求用“qwen-turbo gpt-4o 升级”两模型池跑通。11.2 用成本日志驱动调参路由规则的调整不能拍脑袋。连续记录一周的路由决策日志列出按 route_reason 分组的成本分布。如果某个规则导致大量请求走大模型且理由不成立就调整正则。11.3 质量抽检和反馈闭环部署后要定期抽检小模型的输出。可以在路由器里加一层采样逻辑比如每 20 个请求抽出 1 个把结果推送到人工审核队列。这样能在不影响整体成本的前提下发现质量劣化。11.4 隐私与安全边界路由器会作为集中式模型网关长期保存所有业务请求与响应。这既带来便利也带来安全风险。建议做三件事对请求和响应中的敏感字段脱敏后再写入日志。如果业务方混用不同模型提供商要在用户协议里体现。对访问接口的 token 做权限控制避免内部服务随意调用高成本模型。11.5 控制模型池规模模型池不是越大越好。每个额外接入的模型都会增加路由、测试和故障处理成本。初期建议只保留两到三个档位入门模型、标准模型、旗舰模型。深度使用后再逐步扩展。12. 总结与下一步这个项目最值得尝试的点是把成本优化放到请求路由层来处理而不是纠结于选择某一个模型。它的实现难度不高收益却非常直接。如果你手头已经有正在用的 LLM 应用最应该先验证的三件事是抽样统计自己的任务难度分布、接入一个便宜模型作为默认模型、开启核心问题的缓存。这三步做完再对比成本日志就能看到明显的费用改善。最容易踩的坑有两个一是路由规则设计得太复杂反而引入额外的维护成本二是只关注成本指标忽视了回答质量和失败率。成本优化必须以质量不下降为前提否则省下来的钱会在返工和用户投诉里消耗掉。后续可以继续扩展的方向包括把路由决策升级成基于强化学习的动态路由在每次请求后根据用户反馈调整模型选择策略增加成本预算控制模块设置每日配额上限超阈值后自动切换到保守路由模式接入更多本地推理引擎把低风险任务在离线环境完成。相比一个静态的 Switch 语句这才是把 LLM 成本真正握在自己手里的工程化做法。 模型路由器工程拆解：如何将大模型调用成本降低94%
直接说结论这次我们要拆解的项目不是某个新模型也不是某个推理框架而是一种能在生产环境里显著降低 LLM 调用成本的工程方案AI 模型路由器Model Router。原始项目帖子给出的收益是“cut LLM costs by 94%”也就是 LLM 成本降低 94%。这个数字来自特定业务负载和模型价差下的实测结果不同场景复现时会有浮动但核心思路是通用的不是所有请求都需要让最强、最贵的模型来处理。把简单的查询路由到便宜的小模型把复杂任务交给旗舰模型总成本就会大幅下降。这篇文章会从工程角度完整拆解模型路由器的核心能力、路由策略、成本控制原理、部署方式、接口设计与批量任务接入。你会看到为什么“按难度分流 自动升级 答案缓存”能带来数量级的成本变化也会拿到一套可以直接改造成内部工具的代码骨架。适合已经在用 OpenAI、Claude、DeepSeek、Qwen 等模型 API 的开发者也适合正在做 LLM 应用架构选型、想要控制推理成本的技术负责人。1. 核心能力速览先看这个方案的整体规格。模型路由器本质上是一个位于业务和 LLM 提供商之间的智能转发层它接收上游请求判断任务复杂度和质量要求然后选择最合适的模型去执行。能力项说明项目类型LLM 请求路由 / 模型网关 / 成本控制中间层核心功能按任务难度路由、低配模型兜底、失败自动升级、响应缓存、调用日志成本改善原帖在特定场景下实现 94% 成本下降实际收益取决于负载分布和模型价差延迟影响增加少量路由判断开销毫秒级但可降低排队和超时成本支持模型理论支持所有 OpenAI 兼容 API、Claude API、本地 vLLM / Ollama 服务接口能力可作为独立 HTTP 服务对外暴露统一的/v1/chat/completions风格接口批量任务支持离线批处理队列可在低峰期执行低成本模型任务依赖环境Python 3.10、FastAPI、Redis可选、OpenAI SDK 或其他 HTTP 客户端部署方式容器或裸机运行与业务服务同区域部署可进一步降低转发延迟适合场景客服、内容分类、摘要、RAG 问答、意图识别、批量审核等混合任务负载从表里能看出来这个项目的重点不在模型本身而在“如何把请求分配得更聪明”。它解决的是规模化调用 LLM 时最现实的问题——成本和收益之间的平衡。2. 成本降低 94% 的原理拆解很多团队一上来就盯着最强模型跑所有业务结果一个月下来账单非常可观。实际上大多数 LLM 应用里请求的难度服从长尾分布大量请求是简单的分类、抽取、改写、短问答少数请求才需要深度推理、长上下文理解、复杂代码生成。如果所有请求都走旗舰模型等于用跑车的成本去送外卖。模型路由器降本的核心原理可以拆成四条链路。2.1 难度分层便宜的模型先接路由器维护一个模型优先级表每个模型都有各自的擅长领域和价格档位。接收到请求后它先做一次快速预判。比如判断任务是否有明确指令模板、是否包含多步推理、是否需要最新的知识库信息。如果预判结果是“简单任务”就直接发给小模型比如 DeepSeek-V3、Qwen-Turbo 这类价格便宜的模型。只有预判为复杂任务时才升级到 GPT-4o、Claude Opus 这类旗舰模型。2.2 自动升级小模型失败就换大模型单纯按规则分发并不够。小模型偶尔会读不懂用户意图或者输出的 JSON 格式不合法、答案不符合约束条件。路由器会加一层质量校验校验失败后自动把同一个请求转发给更强的模型。这种“先试低价模型失败再升级”的策略在多数场景下可以显著降低平均成本。2.3 相似答案缓存相同问题不重复计费很多业务里用户会反复问同一个问题。路由器可以按问题的语义哈希或嵌入向量做缓存命中后直接返回历史答案不再调用任何模型 API。缓存本身是降本最快的一步尤其适合客服 FAQ、政策问答、固定知识库检索场景。2.4 批量任务调度把任务放到更便宜的时段和模型对非实时任务路由器可以放入队列在业务低峰期批量调用低成本模型甚至可以切换离线推理。批量处理能提高模型调用吞吐也能减少对在线服务的依赖进一步摊薄单次成本。这里的核心观点是成本下降 94% 不是把模型换成免费的而是让模型调用次数变少、平均单价变低、失败重试变得更加可控。如果业务负载结构不同收益也会不同。比如全部请求都是高难度代码生成路由器的降本空间会小很多。3. 适用场景与使用边界模型路由器不是银弹它适合的任务有几类明显特征。判断你的场景是否符合可以从下面三点入手。第一任务难度有明显差异。比如客服系统里既有“查订单状态”这种简单意图又有“帮忙分析售后纠纷责任”这种复杂问题。路由器可以给前者分配小模型给后者分配大模型。第二大多数请求允许 1 到 3 秒的额外排队和路由时间。第三业务可以接受“先便宜后升级”的失败模式而不是每次都要一次到位。不适合路由器的场景包括所有请求都是硬性高难度任务业务对回复质量要求极高且不能容忍任何小模型输出强依赖单一模型的私有人格和专属能力离线环境下没有可选的多模型池。合规边界方面必须注意几点。使用模型路由时请求会转发给多个提供商要在隐私协议里明确告知用户数据处理范围。涉及敏感身份信息、医疗数据、未成年人数据时不应发送到第三方模型服务。涉及人脸、声音、版权素材的生成类任务必须确认授权链条完整。公司内部使用时要做好数据脱敏和日志脱敏避免 prompt 中的密钥、手机号、身份证号直接落盘。4. 整体架构与关键设计模型路由器的架构通常分成四层接入层、路由决策层、执行层、可观测层。业务服务 → API 网关 / 模型路由器 → 路由决策模块 → 模型提供商 / 本地推理 ↓ 缓存服务 日志 指标接入层负责对外提供统一接口兼容 OpenAI 的/v1/chat/completions格式业务方无需改动代码即可切换。路由决策层是核心它根据规则、上下文长度、历史反馈、成本预算来判断把请求分给谁。执行层封装了多个模型提供商的 HTTP 调用统一重试、超时和鉴权。可观测层负责记录每次调用的模型、token 数、成本、延迟和失败原因。关键设计点是路由决策模块不能太重。如果为了判断复杂度先调用一个大模型那路由本身就可能比直接调小模型还贵。所以路由判断要尽量使用轻量规则指令模板匹配通过正则匹配请求里是否包含明确的复杂任务指令比如“写代码”“解释原理”“对比方案”。文本长度判断超过某个 token 阈值就默认走长文本强模型。工具调用检测如果请求里带了多条工具调用记录或较长的 function calling 上下文需要更强模型。历史行为反馈同一个用户或同一个任务 ID 的历史失败率如果偏高就直接升级模型。关键词置信度内部关键词模型对任务类型打标签置信度高走小模型置信度低走大模型。这套规则都是白盒、可控、可测试的不会引入额外的模型成本。5. 环境准备与前置条件要在本地把模型路由器跑起来先准备一套干净的环境。推荐配置如下。依赖项版本 / 说明Python3.10FastAPI用于搭建 HTTP 服务uvicornASGI 服务器openai官方 SDK兼容多种提供方接口httpx异步 HTTP 客户端redis-py可选用于共享缓存pydantic配置校验loguru 或 structlog结构化日志操作系统不限Windows、Linux、macOS 都可以跑。如果只是接入云厂商 API不需要 GPU如果要接入本地 vLLM、Ollama则需要按对应的模型显存要求准备显卡。建议第一次测试时只接两家 API跑通路由逻辑后再扩展模型池。创建项目目录和虚拟环境mkdir llm-model-router cd llm-model-router python -m venv venv source venv/bin/activate pip install fastapi uvicorn openai httpx redis pydantic loguru这里没有把具体版本写死因为不同环境下 Python 版本和依赖冲突情况不一样。建议使用 requirements.txt 固定版本。6. 核心实现搭建模型路由器服务下面给出一套可运行的最小模型路由器实现骨架。它包含四个模块模型池定义、路由判断、模型调用转发、HTTP 服务。6.1 模型池配置先定义一个数据类来描述不同的模型服务。每个模型都有名称、接口地址、API Key、价格档位和支持能力。from dataclasses import dataclass from typing import List dataclass class ModelSpec: name: str api_type: str # openai / anthropic / vllm base_url: str api_key: str price_per_1k_input: float 0.0 price_per_1k_output: float 0.0 max_context: int 8192 tags: List[str] None# models.py MODEL_POOL { gpt-4o: ModelSpec( namegpt-4o, api_typeopenai, base_urlhttps://api.openai.com/v1, api_key$OPENAI_API_KEY, price_per_1k_input0.005, price_per_1k_output0.015, max_context128000, tags[complex, code, reasoning], ), deepseek-chat: ModelSpec( namedeepseek-chat, api_typeopenai, base_urlhttps://api.deepseek.com/v1, api_key$DEEPSEEK_API_KEY, price_per_1k_input0.001, price_per_1k_output0.002, max_context64000, tags[fast, cheap, general], ), qwen-turbo: ModelSpec( nameqwen-turbo, api_typeopenai, base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, api_key$QWEN_API_KEY, price_per_1k_input0.0005, price_per_1k_output0.002, max_context128000, tags[cheap, classify, extract], ), }注意这里只给了配置示例具体 API Key 和生产版本价格需要用实际环境替换。生产环境中不要直接写在代码里建议用环境变量或密钥管理服务。6.2 路由决策模块路由决策模块是核心。它接收一个 Chat 请求输出应该使用哪个模型。先做规则判断再做语义标签判断最后给默认值。import re from typing import Dict from dataclasses import dataclass dataclass class RouteDecision: model_name: str reason: str upgraded: bool False COMPLEX_PATTERNS [ r写代码|代码, r解释|原理|为什么, r对比|比较|区别, r设计|架构|方案, rdebug|调试|报错, ] def judge_complexity(prompt: str) - tuple[bool, str]: 基于规则判断任务复杂度不调用任何模型。 prompt_lower prompt.lower() for pattern in COMPLEX_PATTERNS: if re.search(pattern, prompt_lower): return True, pattern # 长度超过阈值默认当作复杂任务 if len(prompt) 3000: return True, long_context return False, simple def route_decision(prompt: str, user_params: Dict None) - RouteDecision: complex_task, pattern judge_complexity(prompt) if user_params and user_params.get(force_model): return RouteDecision( model_nameuser_params[force_model], reasonuser_override ) if complex_task: return RouteDecision( model_namegpt-4o, reasonfcomplex_pattern:{pattern} ) # 简单任务走便宜模型 return RouteDecision( model_nameqwen-turbo, reasonsimple_rule )这个模块最大的特点是零成本。它只用正则和长度判断不引入额外的模型调用。6.3 模型调用与自动升级接下来实现执行层。调用模型时统一使用异步 HTTP 客户端。如果小模型失败或返回内容不满足校验条件自动升级到强模型。import httpx import asyncio import json from typing import List class ModelClient: def __init__(self): self.client httpx.AsyncClient(timeout60) async def chat_completion( self, model_spec, messages: List[Dict], temperature: float 0.3, max_tokens: int 1024 ): if model_spec.api_type ! openai: raise ValueError(unsupported api_type) headers { Authorization: fBearer {model_spec.api_key}, Content-Type: application/json, } payload { model: model_spec.name, messages: messages, temperature: temperature, max_tokens: max_tokens, } resp await self.client.post( f{model_spec.base_url}/chat/completions, headersheaders, jsonpayload, ) resp.raise_for_status() data resp.json() return data[choices][0][message][content] async def close(self): await self.client.aclose() async def execute_with_upgrade(messages: List[Dict]): 先走便宜模型失败或校验失败时升级到旗舰模型。 client ModelClient() try: prompt messages[-1][content] if messages else decision route_decision(prompt) # 第一次调用使用路由决策的模型 primary_model MODEL_POOL[decision.model_name] try: result await client.chat_completion(primary_model, messages) if validate_output(result): return { model: primary_model.name, content: result, upgraded: False, reason: decision.reason, } except Exception as exc: logger.warning(fprimary model failed: {primary_model.name}, error{exc}) # 升级调用 fallback_model MODEL_POOL[gpt-4o] result await client.chat_completion(fallback_model, messages) return { model: fallback_model.name, content: result, upgraded: True, reason: auto_upgrade, } finally: await client.close()validate_output 函数可以根据业务定义比如检查返回内容是否包含指定字段、JSON 是否可解析、长度是否达标。这里是典型的“廉价模型优先 失败升级”模式。6.4 缓存模块缓存可以减少重复请求。最简单的实现是缓存的哈希对普通文本可以解决问题。稍微复杂一点的版本可以接 Redis。import hashlib import json class SimpleCache: def __init__(self): self._store {} def _key(self, messages: List[Dict], model: str) - str: raw json.dumps(messages, ensure_asciiFalse) |v1 return hashlib.sha256(raw.encode()).hexdigest() def get(self, messages: List[Dict], model: str): key self._key(messages, model) return self._store.get(key) def set(self, messages: List[Dict], model: str, content: str, ttl_seconds: int 3600): key self._key(messages, model) self._store[key] { content: content, expire_at: time.time() ttl_seconds, } def get_or_set(self, messages, model, fallback): cached self.get(messages, model) if cached and cached[expire_at] time.time(): return cached[content] content fallback() self.set(messages, model, content) return content更合理的做法是同时把缓存键拆成“用户输入 系统提示词 温度”。温度会影响输出的稳定性如果业务允许建议把温度固定为 0 或 0.1提高缓存命中率。6.5 HTTP 服务最后把服务暴露成 OpenAI 兼容接口。业务方只需要改一个 base_url 就能接入。from fastapi import FastAPI, Request from pydantic import BaseModel, Field from typing import List, Dict, Optional app FastAPI(titleModel Router Service, version0.1.0) class ChatMessage(BaseModel): role: str content: str class ChatCompletionRequest(BaseModel): model: Optional[str] auto messages: List[ChatMessage] temperature: float 0.3 max_tokens: int 1024 stream: bool False metadata: Optional[Dict] None class ChatCompletionResponse(BaseModel): id: str object: str chat.completion model: str choices: list usage: dict app.post(/v1/chat/completions) async def chat_completions(req: ChatCompletionRequest): messages [m.dict() for m in req.messages] cache SimpleCache() cached cache.get(messages, routed) if cached: return build_response(cache, cached, usage{hit: True}) result await execute_with_upgrade(messages) cache.set(messages, result[model], result[content]) return build_response(result[model], result[content], usageresult)这里简化了响应格式。生产环境建议把响应结构做成和 OpenAI 官方一致这样上游迁移成本最低。7. 接口 API 与批量任务接入服务跑起来后业务方调用方式非常直接。先启动服务uvicorn main:app --host 127.0.0.1 --port 8000启动后可以用 curl 验证curl http://127.0.0.1:8000/v1/chat/completions \ -H Content-Type: application/json \ -d { model: auto, messages: [{role: user, content: 把下面这句话翻译成英文今天天气很好}], temperature: 0.3 }返回结果会标注实际使用的模型名方便核对路由行为{ id: chatcmpl_xxx, object: chat.completion, model: qwen-turbo, choices: [ { index: 0, message: { role: assistant, content: The weather is very nice today. }, finish_reason: stop } ], usage: {} }Python 调用端也兼容 OpenAI 官方 SDK只需要覆盖 base_urlfrom openai import OpenAI client OpenAI( api_keytest-key, base_urlhttp://127.0.0.1:8000/v1 ) response client.chat.completions.create( modelauto, messages[{role: user, content: 这篇文档讲了什么}], temperature0.2 ) print(response.choices[0].message.content)7.1 批量任务设计批量任务场景下路由器要提供两个能力一个是可以自定义每个任务的模型优先级另一个是支持失败重试。推荐的批量任务格式{ tasks: [ { task_id: 001, mode: auto, fallback_order: [deepseek-chat, gpt-4o, qwen-turbo], messages: [ {role: user, content: 判断下面这段评论是好评还是差评发货很快质量很好} ], temperature: 0.1, max_tokens: 200 } ] }批量任务调度器可以基于 asyncio 队列实现。核心调度逻辑读取任务列表。每个任务执行路由决策。如果任务失败按 fallback_order 依次重试。记录每个任务的实际模型、token 消耗、耗时和错误信息。全部完成后输出结果 CSV 或 JSON 文件。一个简化版批量处理脚本import asyncio import json async def process_batch(input_file: str, output_file: str): with open(input_file, r, encodingutf-8) as f: batch json.load(f) results [] for task in batch[tasks]: try: result await execute_with_upgrade(task[messages]) results.append({ task_id: task[task_id], status: ok, model: result[model], content: result[content], reason: result[reason], }) except Exception as exc: results.append({ task_id: task[task_id], status: error, error: str(exc), }) with open(output_file, w, encodingutf-8) as f: json.dump(results, f, ensure_asciiFalse, indent2) if __name__ __main__: asyncio.run(process_batch(tasks.json, results.json))批量处理时要特别关注 Token 配额和限流。很多模型 API 有每分钟请求数限制建议在批量处理器里加延迟和退避重试。8. 成本测算与效果验证要验证路由器到底省了多少钱需要在路由前后各做一轮统计。最关键的有效指标是“每 1000 次请求的实际计费金额”而不是只看总 token 数。8.1 成本记录每次调用模型后路由器都应该把以下信息写入日志request_id任务类型标签实际使用的模型输入 token 数输出 token 数预估成本是否从缓存命中是否经过升级路由决策原因请求耗时日志结构示例{ request_id: req_001, task_type: classification, model: qwen-turbo, prompt_tokens: 120, completion_tokens: 32, estimated_cost_usd: 0.000064, cache_hit: false, upgraded: false, route_reason: simple_rule, latency_ms: 420 }有了日志以后按天聚合成本SELECT model, COUNT(*) AS requests, SUM(estimated_cost_usd) AS total_cost, SUM(CASE WHEN upgraded THEN 1 ELSE 0 END) AS upgrade_count, SUM(CASE WHEN cache_hit THEN 1 ELSE 0 END) AS cache_hit_count FROM router_usage_logs GROUP BY model ORDER BY total_cost DESC;8.2 对比测试方法比较合理的验证流程分成三步。第一步用原来的单一模型方案处理一万条真实业务请求记录总费用和质量指标。第二步开启路由器用相同的请求集重放记录新的总费用和质量指标。第三步对比两组数据。需要对比的指标至少包括五个指标单一模型方案模型路由器方案变化总成本美元基线路由后下降幅度每千请求成本基线路由后核心指标平均响应延迟基线路由后需观察答案通过率 / 质检通过率基线路由后不能明显下降高优先级任务失败率基线路由后建议低于基线对日志口径特别提醒实际账单可能还会包含平台服务费、网络传输费、额外 token 溢出、缓存计算开销等。如果要做预算汇报最好把“估算成本”和“实际账单”分开列出来。8.3 为什么可以达到 94%从原帖标题看这个路由器把成本砍掉了 94%。在真实数据中要做到这种量级通常需要三个条件同时成立一是任务池里有大量简单任务例如超过 90% 的请求只是问答分类或文本抽取二是简单任务和复杂任务模型单价相差 5 到 20 倍三是缓存命中率能稳定在 30% 以上。三个条件叠加总成本下降确实可能接近 90% 以上。如果业务负载全是复杂代码推理那成本下降幅度会小很多。所以部署前先抽样统计你自己的请求分布这是最重要的一步。9. 性能观察与资源占用模型路由器本身是一个非常轻的代理层。它的 CPU 和内存占用主要来自请求转发、JSON 序列化和缓存读写。在没有启用本地大模型的情况下不需要 GPU。单独运行时内存占用通常在 200MB 到 500MB 之间具体取决于并发连接数和缓存规模。性能观察建议关注以下指标路由决策耗时正则匹配一般小于 1ms如果引入关键词模型分类单请求会增加 10 到 100ms。代理层转发延迟每增加一层转发会多 1 到 5ms 网络开销如果服务部署在模型 API 同一区域影响可忽略。缓存命中率命中率越高平均响应延迟越低。理论上有缓存时延迟可能低于直接调用成本模型。超时设置模型 API 超时建议设置在 30 到 120 秒之间。路由器本身的上游超时要比业务超时短避免连接堆积。避免端口冲突和进程残留是本地部署最容易踩的坑# 查看端口占用 lsof -i :8000 # 如果端口被占用杀掉旧进程 kill -9 $(lsof -t -i :8000) # 或用不同端口启动 uvicorn main:app --host 127.0.0.1 --port 8001生产环境建议用 systemd 或 Docker Compose 管理服务进程增加健康检查和自动重启。10. 常见问题与排查方法模型路由器在部署和运行中会遇到一些固定问题排查思路基本可以复用。问题现象可能原因排查方式解决方案启动后接口 404路由路径未正确注册检查 FastAPI 路由装饰器确认/v1/chat/completions路径与调用一致请求一直超时上游模型 API 不可用或网络被限制单独 curl 上游接口检查 API Key、网络策略、代理设置请求全部走了大模型路由规则误判复杂任务查看 route_reason 日志调整正则或阈值检查是否误传 force_model缓存命中率低消息里带时间戳或随机参数对比文本缓存 key清理无关参数再生成缓存 key小模型返回格式不合格模型没有遵循 JSON 约束查看全量返回日志在 prompt 中增加示例或失败后升级模型批量任务中途卡死没有设置超时或重试查看任务日志为每个任务配置超时和退避重试成本没有下降复杂任务占比过高按 task_type 统计成本优化简单任务 prompt提升缓存命中使用本地 vLLM 部署时报错base_url 不是 OpenAI 兼容格式检查 vLLM 的 service 地址确认/v1路径正确模型名一致API Key 泄露密钥写在代码仓库扫描日志和代码改用环境变量轮换密钥日志脱敏11. 最佳实践与使用建议模型路由器能做到成本优化但也要注意质量和成本之间的动态平衡。下面这几条是工程上验证过的建议。11.1 先跑通最小链路第一次接入时不要直接上全量业务。先拿 1000 条真实请求用“qwen-turbo gpt-4o 升级”两模型池跑通。11.2 用成本日志驱动调参路由规则的调整不能拍脑袋。连续记录一周的路由决策日志列出按 route_reason 分组的成本分布。如果某个规则导致大量请求走大模型且理由不成立就调整正则。11.3 质量抽检和反馈闭环部署后要定期抽检小模型的输出。可以在路由器里加一层采样逻辑比如每 20 个请求抽出 1 个把结果推送到人工审核队列。这样能在不影响整体成本的前提下发现质量劣化。11.4 隐私与安全边界路由器会作为集中式模型网关长期保存所有业务请求与响应。这既带来便利也带来安全风险。建议做三件事对请求和响应中的敏感字段脱敏后再写入日志。如果业务方混用不同模型提供商要在用户协议里体现。对访问接口的 token 做权限控制避免内部服务随意调用高成本模型。11.5 控制模型池规模模型池不是越大越好。每个额外接入的模型都会增加路由、测试和故障处理成本。初期建议只保留两到三个档位入门模型、标准模型、旗舰模型。深度使用后再逐步扩展。12. 总结与下一步这个项目最值得尝试的点是把成本优化放到请求路由层来处理而不是纠结于选择某一个模型。它的实现难度不高收益却非常直接。如果你手头已经有正在用的 LLM 应用最应该先验证的三件事是抽样统计自己的任务难度分布、接入一个便宜模型作为默认模型、开启核心问题的缓存。这三步做完再对比成本日志就能看到明显的费用改善。最容易踩的坑有两个一是路由规则设计得太复杂反而引入额外的维护成本二是只关注成本指标忽视了回答质量和失败率。成本优化必须以质量不下降为前提否则省下来的钱会在返工和用户投诉里消耗掉。后续可以继续扩展的方向包括把路由决策升级成基于强化学习的动态路由在每次请求后根据用户反馈调整模型选择策略增加成本预算控制模块设置每日配额上限超阈值后自动切换到保守路由模式接入更多本地推理引擎把低风险任务在离线环境完成。相比一个静态的 Switch 语句这才是把 LLM 成本真正握在自己手里的工程化做法。