ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

大模型路由实战:三层动态分流中枢,让API成本与性能兼得

大模型路由实战:三层动态分流中枢,让API成本与性能兼得

这个月我朋友圈里最热闹的一件事,就是 AA 智能指数 10 月榜单更新后的那场“大洗牌”。闭源旗舰 C1 以综合分 89.7 登顶,看起来顺理成章;真正让讨论炸锅的是两款开源模型——轻量黑马 M 和主攻代码的新秀 D——分别在性价比和代码专项上把不少老牌付费模型踩了下去。群里有人感慨“大模型神坛一夜崩塌”,我倒是觉得,榜单本身没那么重要,重要的是它戳破了一个幻觉:以为只要抱住一个最强模型,所有需求都能搞定。这篇内容我想和大家聊聊,我为什么在榜单洗牌后不再迷信单一模型,以及我自己搭建的一套“三层动态分流中枢”是怎么做到的。

这套东西技术上不算难,但它确实在解决一个很现实的问题:怎么把不同模型的优势组合起来,同时控制成本、延迟和稳定性。我会把架构拆解、策略代码、跨平台部署脚本都放出来,最后还会聊几个真正落地时才会踩到的坑。不管你是个人开发者还是小团队后端,只要每天在调大模型 API,这篇文章应该能给你一套直接抄作业的方案。

1. “神坛”怎么塌的:榜单洗牌背后的真实信号

1.1 这份榜单到底在说什么

AA 智能指数每个月会做一轮综合评测,覆盖文本理解、代码生成、数学推理、长文档、Agent 工具调用等维度。10 月榜单最有意思的地方,不是旗舰 C1 拿了第一,而是第二名之后的座次发生了明显松动。

模型类型综合分代码专项数学推理输入价格(元/M token)上榜亮点
旗舰 C1闭源旗舰89.79.49.160综合分第一,Agent 调用最稳
开源 M开源轻量86.28.58.72性价比突出,长文本窗口实用
开源 D开源代码向84.89.67.93.5代码专项反超旗舰 C1
旧旗舰 R闭源旧款80.17.88.145相比上季度下滑明显

这里的数据只是示意,我不是要做评测,而是想说明一个容易被忽略的事实:榜单比的是“总分”,但真实业务用到的是“分项能力”。旗舰 C1 确实全面,价格也全面。开源 M 的综合分低了几分,但输入价格只有 C1 的 3%,而且长文本窗口很实用。开源 D 代码专项 9.6 分,比旗舰 C1 还高,意味着很多 code review、补全、单元测试生成场景,根本没有必要付旗舰模型的溢价。

换句话说,所谓“神坛崩塌”,不是哪个模型突然变蠢了,而是开源模型的进步让“绑定单一模型”的性价比越来越难看。

1.2 绑定单一模型为什么越来越危险

我在给一个客服系统团队做技术方案的时候,见过一个非常典型的失误。他们最初只接一个闭源大模型,所有会话、总结、工单分类全部走同一个入口。请求量涨起来之后,月度 API 费用翻了将近三倍。更麻烦的是,有些简单咨询明明用一个便宜小模型就能回答,但因为没有路由层,代码里到处是硬编码的模型名,想改都改不动。

这种情况不是个例。绑定单一模型的风险主要体现在四个地方:

  • 成本不可控:能力强等于单位价格高,小请求没必要用大炮打蚊子。
  • 性能单点:某个模型服务端抖动、限流或者临时下线,你的核心链路跟着瘫痪。
  • 供应商策略变化:模型调价、接口版本变更、老版本退役,都需要业务代码跟着改。
  • 能力错配:同一模型不可能在所有任务类型上都最强,总会有专项短板。

我自己的经验是,上个月某模型网关出现过一次持续六小时的服务异常。因为路由层在,流量在两分钟内切到了备选模型,业务侧基本无感,只有日志里多了一行 fallback_used: true。如果还是单模型直连,那个上午大概率要写故障报告了。

1.3 动态分流不是“负载均衡”

很多人听到“路由中枢”会下意识觉得这就是负载均衡。真不是一回事。负载均衡是把同样的请求分给多台相同能力的机器;动态分流是给不同请求找不同的模型后端,背后是一套“任务类型 + 成本 + 延迟 + 质量”的约束匹配。

打个比方:旗舰 C1 像三甲医院的全科专家,什么病都能看,但挂号费高、排队久;开源 M 像社区门诊,常见问题处理得又快又便宜;开源 D 像骨科专科大夫,碰上手部骨折这类问题比全科专家还专业。好医院不会让专家看所有感冒,也不会让门诊医生硬接开颅手术。动态分流要做的事情,就是让每个请求找到合适的科室。

接下来就是我整个方案的核心:三层动态分流中枢。第一层管“进来的是什么”,第二层管“该去哪”,第三层管“去了之后效果怎么样以及出了问题怎么办”。

2. 三层动态分流中枢的架构拆解:输入、决策、执行各管一段

2.1 入口路由层:先搞清楚请求长什么样

入口层的主职责不是转发,而是“标准化”。业务方不需要关心背后接了几套模型、各自 API 格式有什么区别,只需要对接一个统一入口。

我用的技术方案是 Python + 一个轻量 ASGI Web 框架,对外暴露 OpenAI 风格的接口,例如 POST /v1/chat/completions。业务代码把 base_url 指到这里,传入 messages、temperature、stream 等标准字段,剩下的交给中枢处理。

请求进入后,入口层会做一次特征提取,生成一个标准化信封包:

  • 请求类型:从 message 内容里用关键词规则或模型分类,识别出代码生成、数学推理、通用对话、长文档总结等任务类型。
  • 输入规模:估算 token 数量,决定要不要走长文本模型或者先做本地裁剪。
  • 用户标签:不同业务线、不同用户等级可以挂不同的预算和延迟要求。
  • 流式标识:是否流式输出直接决定后面的降级策略能不能用。
  • 超时预期:请求方希望多快拿到结果,这会影响模型候选排序。

这一步很关键,因为后面策略层的所有判断都建立在几个结构化字段上,不需要反复读原始 message。

2.2 策略决策层:每个请求该送向哪个模型

策略层是整个中枢的心脏,它维护一张“模型注册表”,里面记录每个模型的类型、价格、上下文上限、能力评分、健康状况。决策过程也不是简单查表,我会分三步走:

  1. 硬规则优先:有些场景不需要计算,规则直接拍板。例如代码 review 默认走开源 D,涉及私密数据的请求强制走本地部署模型,绝不送公共 API。
  2. 加权评分:对剩余候选模型做综合打分,维度包括质量分、成本效率分、速度分。
  3. 候选排序:输出主选、备选、兜底三个结果,交给执行层处理。

打个比方,决策层像一台自动售票机:你知道你要去哪里(任务类型)、愿意花多少钱(预算)、想多快上车(延迟要求),机器会告诉你该买哪趟车,同时告诉你没赶上这趟车时换哪辆。

2.3 执行与兜底层:出问题时要有人接住

执行层的职责是保证“答出来”和“答得好”之间的平衡。它负责做真实的 API 调用、超时控制、重试、降级切换、输出校验和日志埋点。

我第一次做这套系统时最大的失误,是只做了“主模型失败后切备选”的降级,没有做超时分级。结果遇到某个模型持续卡顿,请求堆积,最后把整个网关都拖垮了。后来我把超时拆成了连接超时、首包超时、总超时三个等级,又加上了并发池和令牌桶限流,才算是把稳定性拉回来。

执行层的核心逻辑其实可以用一句话概括:只给主选有限次机会,不行就快速切备用,再不行就找兜底,所有过程都要留痕迹。

3. 中心策略引擎实战:从模型注册表到路由规则的代码实现

3.1 模型注册表:把模型信息全部收进配置

这一步要做的第一件事,就是用 YAML 定义模型注册表。我通常叫它 model.yaml,里面记录所有可用模型的外部形态。下面是精简版示例:

version: 2 models: c1: provider: closed-c1 base_url: https://api.example-c1.com/v1 api_key_env: MODEL_C1_KEY context_limit: 200000 price_in: 60 # 元 / 百万 token price_out: 120 tags: [flagship, reasoning, agent] capabilities: code: 0.94 math: 0.91 writing: 0.88 p95_latency_ms: 12000 concurrency: 8 open-m: provider: open-m base_url: https://api.example-m.com/v1 api_key_env: MODEL_M_KEY context_limit: 128000 price_in: 2 price_out: 6 tags: [lightweight, long_context, chat] capabilities: code: 0.82 math: 0.84 writing: 0.86 p95_latency_ms: 4000 concurrency: 20 open-d: provider: open-d base_url: https://api.example-d.com/v1 api_key_env: MODEL_D_KEY context_limit: 64000 price_in: 3.5 price_out: 9 tags: [code, review, unittest] capabilities: code: 0.96 math: 0.78 writing: 0.70 p95_latency_ms: 6000 concurrency: 12

每个字段都不是摆设。api_key_env 表示密钥从环境变量里读,避免把 Key 写进配置文件;capabilities 是给评分器用的能力基线;p95_latency_ms 来自历史日志滚动更新,你每隔一段时间要根据监控数据改一次。

路由策略单独放在 router.yaml:

rules: code_review: prefer: open-d fallback: [open-m, c1] math_reasoning: prefer: c1 fallback: [open-m, open-d] generic_chat: prefer: open-m fallback: [open-d, c1] scoring: weights: quality: 0.5 cost: 0.3 speed: 0.2 budget_ceiling: 0.15 timeout_ms: 30000 cache_ttl_seconds: 600

这里 weights 是全局参数,如果你某个业务线极度敏感、要求质量优先,可以把 quality 拉到 0.7。如果业务是面向 C 端的轻交互,speed 权重可以提到 0.4。灵活性的前提是策略都收拢在配置里,而不是散落在代码的 if else 里。

3.2 路由函数:规则、权重、打分怎么组合

决策层拿到 RequestEnvelope 之后,会调用一个 select_model 函数。它的完整逻辑可以简化成下面这一段 Python 代码:

def select_model(envelope, registry, router_config): # 第一步:硬规则优先 for rule_name, rule_conf in router_config["rules"].items(): if envelope.request_kind == rule_name: return rule_conf["prefer"], rule_conf["fallback"] # 第二步:按打分排序 candidates = [] for model_id, model_conf in registry["models"].items(): if estimate_tokens(envelope) > model_conf["context_limit"]: continue score = 0.0 weights = router_config["scoring"]["weights"] score += weights["quality"] * model_conf["capabilities"].get(envelope.task_tag, 0.5) score += weights["cost"] * cost_efficiency(model_conf, router_config) score += weights["speed"] * speed_score(model_conf, router_config["scoring"]["timeout_ms"]) candidates.append((score, model_id)) candidates.sort(reverse=True) primary = candidates[0][1] backup = candidates[1][1] last = candidates[-1][1] return primary, [backup, last]

有几个细节值得说明。

cost_efficiency 不是简单看价格绝对值,而是看价格对比区间。我会取注册表里所有候选模型的最高价作为分母,价格越低得分越高,这样旗舰 C1 在“成本维度”天然吃亏,但它在 quality 维度高,如果任务真的难,综合分还是能把它顶上去。

speed_score 基于 p95_latency 计算,越接近零延迟得分越高。这里要注意的是,延迟数据一定要用真实线上数据回填,不要用厂商文档里的宣称值。厂商给的只是理论边界,线上超时率才是真实体验。

3.3 降级链与流式请求的取舍

决策层给出主选、备选、兜底三个候选后,执行层按照降级链调用:

  1. 尝试主选模型。
  2. 如果发生超时、5xx、限流,做一次指数退避重试。
  3. 重试无效,切到备选模型。
  4. 备选模型失败,切到兜底模型。
  5. 兜底也失败,返回标准错误码 model_overloaded。

严格模式下,我建议重试次数最多 1 次。不要反复打同一个已超时的模型,那只会加剧拥塞。降级链的实现类似:

async def dispatch_with_fallback(ctx, candidates): last_error = None for model_id in candidates: try: async with deadline(ctx.timeout_ms): return await call_model(ctx, model_id) except TimeoutError as e: last_error = e continue except RateLimitError as e: last_error = e await sleep_with_jitter() continue except BadResponseError as e: last_error = e continue raise ModelOverloaded(last_error)

流式请求是另一套逻辑。SSE 流一旦开始向客户端发送数据,你就没法中途换个模型继续接上,因为两边生成的 token 文本大概率对不上。我自己的做法是:流式请求只有在“首包发出前”才允许主备切换;首包发出后模型锁定,再出错只能断流,由客户端决定是否重试。

所以在入口层识别 stream=True 之后,执行层就要把它标记为“锁定模式”,不能走通用的 race mode 或降级逻辑。

4. 三大系统跨平台一键套件:容器编排加三个启动脚本

4.1 为什么用容器封装,而不是裸装 Python

这套中枢依赖 Python 包、YAML 配置、缓存服务,如果每个开发者的机器环境不同,光是安装依赖就能消耗一个下午。我更倾向于把所有东西装进一个容器镜像,宿主机只需要有容器引擎和编排工具。

容器方案带来的好处很直接:

  • 跨平台一致:Windows、macOS、Linux 跑的是同一个镜像,行为一致。
  • 依赖隔离:Python 版本、pip 包互相不污染。
  • 升级简单:改代码后重新构建镜像即可,不用在每台机器上手动操作。

硬件要求不算高,一台 4GB 内存的机器就能跑通核心网关,因为真正的模型推理都在远端 API 服务上,网关本身只做路由和转发。

4.2 目录结构与编排配置

我习惯把整套中枢做成一个名为 portal 的独立工程,目录大概长这样:

portal/ ├── app/ │ ├── main.py # 网关入口 │ ├── router.py # 路由核心 │ ├── dispatcher.py # 执行与降级 │ └── health.py # 健康检查 ├── config/ │ ├── model.yaml │ └── router.yaml ├── docker-compose.yml ├── .env.example ├── start.sh ├── start.ps1 ├── start.bat └── README.md

docker-compose.yml 是核心编排文件:

services: portal: build: . ports: - "8080:8080" env_file: - .env volumes: - ./config:/app/config:ro environment: - CACHE_REDIS_URL=redis://cache:6379/0 depends_on: - cache cache: image: redis:7-alpine restart: unless-stopped

我没有把密钥写进 yml,而是放到 .env 里统一管理。.env.example 长这样:

MODEL_C1_KEY= MODEL_M_KEY= MODEL_D_KEY= CACHE_REDIS_URL=redis://cache:6379/0 ROUTER_PORT=8080 LOG_LEVEL=INFO

实际使用的时候,复制 .env.example 为 .env,填上各家模型 API Key,然后启动即可。这里有个小提醒:.env 必须加入 .gitignore,否则密钥很容易跟着仓库一起泄漏。

4.3 Windows、macOS、Linux 三套启动脚本

三大系统各给一个脚本,macOS 和 Linux 共用 start.sh,Windows 用户可以用 start.bat 或 start.ps1。

start.sh 的逻辑如下:

#!/usr/bin/env bash set -euo pipefail command -v docker >/dev/null 2>&1 || { echo "[ERROR] 未检测到 Docker,请先安装容器引擎"; exit 1; } docker compose version >/dev/null 2>&1 || { echo "[ERROR] 需要 Docker Compose V2"; exit 1; } if [ ! -f .env ]; then cp .env.example .env echo "[INFO] 已从 .env.example 创建 .env,请填写模型 API Key 后重新运行。" exit 0 fi docker compose up -d --build for i in $(seq 1 30); do if curl -fs http://localhost:8080/healthz >/dev/null 2>&1; then echo "[OK] 中枢已启动,入口: http://localhost:8080" exit 0 fi sleep 1 done echo "[ERROR] 服务启动超时,请查看 docker compose logs" exit 1

start.ps1 是 PowerShell 版本,适合 Windows 用户:

$ErrorActionPreference = "Stop" if (-not (Get-Command docker -ErrorAction SilentlyContinue)) { Write-Host "[ERROR] 未检测到 Docker,请先安装容器引擎" exit 1 } docker compose version | Out-Null if (-not (Test-Path .env)) { Copy-Item .env.example .env Write-Host "[INFO] 已创建 .env,请填写模型 API Key 后重新运行。" exit 0 } docker compose up -d --build for ($i = 0; $i -lt 30; $i++) { try { Invoke-RestMethod http://localhost:8080/healthz -TimeoutSec 2 | Out-Null Write-Host "[OK] 中枢已启动,入口: http://localhost:8080" exit 0 } catch { Start-Sleep -Seconds 1 } } Write-Host "[ERROR] 服务启动超时,请查看 docker compose logs" exit 1

如果连 PowerShell 都不想用,我再给一个 CMD 版本:

@echo off where docker >nul 2>nul if errorlevel 1 ( echo [ERROR] 未检测到 Docker,请先安装容器引擎 exit /b 1 ) docker compose version >nul 2>nul if errorlevel 1 ( echo [ERROR] 需要 Docker Compose V2 exit /b 1 ) if not exist .env ( copy .env.example .env >nul echo [INFO] 已创建 .env,请填写模型 API Key 后重新运行。 exit /b 0 ) docker compose up -d --build for /l %%i in (1,1,30) do ( curl -fs http://localhost:8080/healthz >nul 2>nul if not errorlevel 1 ( echo [OK] 中枢已启动,入口: http://localhost:8080 exit /b 0 ) timeout /t 1 /nobreak >nul ) echo [ERROR] 服务启动超时,请查看 docker compose logs exit /b 1

三份脚本做的事情完全一致:检查环境、创建 .env、启动容器、等待健康检查通过。健康检查接口不只是网关活着,还会返回各模型的最近探活状态,方便确认某条链路是否可用。这样做的好处是,克隆工程后,从零到跑起来只需要三步:填 Key、跑脚本、发请求。

5. 落地一周后:那些榜单上看不见的坑和参数经验

5.1 成本不是按价目表算的,token 膨胀才是大头

很多团队看到价目表之后只会简单算:旗舰贵,开源便宜。实际情况远没有那么简单。如果你不控制输入 token,一个看似便宜的模型也可能产生巨额账单。

我举一个估算例子。假设某天有 20 万次请求,平均每次输入 5000 token、输出 500 token:

  • 全部打旗舰 C1:输入 10 亿 token(按 60 元/百万)算 60000 元,输出 1 亿 token(按 120 元/百万)算 12000 元,合计 72000 元。
  • 分流后,简单请求占 60% 走开源 M(输入 2 元、输出 6 元),复杂请求 40% 走旗舰 C1:输入成本 = 6 亿 token ÷ 100万 × 2 元 + 4 亿 token ÷ 100万 × 60 元 = 1200 + 24000 ≈ 25200 元,输出成本 = 6000 万 token ÷ 100万 × 6 元 + 4000 万 token ÷ 100万 × 120 元 = 360 + 4800 = 5160 元,合计约 30360 元。

这还只是模型单价上的差距。真正需要警惕的是系统提示词反复拼接、历史消息无限折叠、工具调用结果不裁剪带来的 token 膨胀。我见过一个项目,每次请求塞进 8000 token 的指令文本,实际用户问题只有 200 token,成本浪费非常夸张。入口层做输入压缩和历史摘要,是比路由更为优先的省钱手段。

5.2 路由命中率怎么盘点:日志和指标缺一不可

没有观测的路由中枢等于盲飞。我在每次请求完成后都会输出一条 JSON 日志,包含路由决策和实际结果:

{ "ts": "2025-10-16T10:00:01Z", "request_id": "req_10001", "kind": "code_review", "selected_model": "open-d", "fallback_used": false, "latency_ms": 890, "est_cost": 0.0021 }

每周我都会做一次统计,看看路由分布情况。示意结果可能是这样:

任务类型请求量主要命中模型平均延迟平均成本
代码 review45%open-d890ms0.0021
通用对话35%open-m420ms0.0008
数学推理15%c12100ms0.0090
长文档总结5%open-m3500ms0.0030

如果某个候选模型连续两周都没有出现在命中分布里,我会考虑直接把它下线,省掉健康检查和注册表维护的成本。相反,如果某个模型频繁成为 fallback_used=true 的备胎,说明主选模型稳定性有问题或者策略打分没匹配现实。

5.3 兜底模型的选择:不求最聪明,只求不摆烂

兜底模型是一个很容易被忽视的设计点。很多人的第一反应是“兜底当然选能力最强的”,实际上恰恰相反。兜底模型出现在异常链路里,用户状态已经很差了,这时候最重要的是快速给出一个合理响应,而不是追求完美答案。

我自己的兜底选择标准是:稳定性大于速度,速度大于效果。宁可模型答得平庸,也不能让它无限转圈或者连续抛错。更关键的是,兜底链路不要无限重试。兜底失败就应该快速失败,返回标准化的 model_overloaded 错误码,让调用方决定是重试还是展示友好提示。无限重试只会把网关线程池全部占满。

还有一个流式场景的经验:如果客户端要求 SSE 流式返回,兜底策略会自动降级为“首包前可切换”。首包一旦发出,不再做任何模型替换。否则即使能接上,前后文风格和格式也大概率是乱的,对用户感知来说反而是事故。

5.4 一组可复用的调参起点

根据不同业务形态,参数肯定要微调,但这里有一组我验证过还算稳妥的初始值:

参数初始值说明
quality / cost / speed 权重0.5 / 0.3 / 0.2质量优先兼顾成本速度
非流式超时30 秒超过即触发降级
流式首包超时5 秒首包太慢直接切换
race mode关闭低危任务可开启
语义缓存 TTL600 秒相同问题请求直接命中
降级重试次数1 次避免雪崩
日志级别INFO含路由决策与成本估算

灰度接入建议分三步:先切 10% 流量跑三到五天,对比 P95 延迟、错误率和平均成本;确认数据没有恶化后再切到 50%;稳定一周后放量到 100%。不要第一天就全量切,毕竟模型评分是静态值,线上真实表现才说了算。

最后说一点我的个人习惯。这套中枢真正值钱的地方,不是那几段路由代码,而是每周一早上花二十分钟看路由分布表的动作。模型榜单每个月都在洗牌,你的路由表也要跟着洗。哪家模型突然降价了,哪家新版本把数学能力提上来了,改一行 YAML 就能让流量切过去,这种“不慌”的感觉,才是做分流中枢最大的回报。

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进