用 LiteLLM 搭一个按任务自动分流的多模型网关

这篇能做出什么

做完之后，你会有一个跑在本机的 HTTP 服务，对外暴露和 OpenAI 一样的 /v1/chat/completions 接口。调用方只管发消息，网关自己判断这次请求该用哪一档模型：

举个具体例子。同一段前端代码：

• "把这段注释翻译成英文" → instant
• "把这个接口的返回字段整理成表格" → mini
• "这个页面每次切换 Tab 都会重新请求，帮我分析根因并给出重构方案" → reasoning

除了分流，你还会有三样东西：

1. 失败回退：instant 超时或报错 → 自动降级到 mini；mini 挂了 → 升到 reasoning；全挂则返回结构化错误。

2. 用量观测：每条请求打一行 JSON 日志（档位、真实模型、延迟、token、估算成本），另有一个 /stats 接口按档位汇总调用量、P50/P95 延迟和累计成本。

3. 预算兜底：调用方可以用 x-priority: cheap 强制走最便宜档。

前置条件清单

• Python 3.10 以上（代码里用了 str | None 这种写法）
• 一个 OpenRouter API Key，或者任意一家 OpenAI 兼容服务的 Key
• 本地能起 uvicorn 的端口
• 想清楚三件事：三档分别对应哪三个真实模型 ID、每档的超时可接受多长、每档的 max_tokens 上限

关于模型 ID：本文里出现的 gpt-5.5-instant、gpt-5.5-mini、gpt-5.5 都是占位名。真实的模型 ID、可用区域、支持的参数、价格，一律以 OpenRouter 或各家的官方模型页为准。价格和上下文长度这类信息变化很快，写死在教程里意义不大，正确的做法是写进配置文件、随官方调整。

安装依赖：

```bash

python -m venv .venv && source .venv/bin/activate

pip install litellm fastapi "uvicorn[standard]" pyyaml

export OPENROUTER_API_KEY="sk-or-你的key"

```

项目结构：

```

gateway/

config.yaml # 模型清单 + 回退规则

router_builder.py # 把 yaml 变成 LiteLLM Router

policy.py # 路由分类器（纯规则，无网络调用）

obs.py # 日志与统计

app.py # FastAPI 入口

```

第一步：配置文件里定义三档

核心思想：档位名是别名，模型 ID 只在一处出现。业务代码永远只说"我要 mini 档"，从不说具体模型名。将来换供应商、换模型版本，只改这个文件。

gateway/config.yaml：

```yaml

model_list:

• model_name: tier-instant

litellm_params:

model: openrouter/openai/gpt-5.5-instant # ← 换成官方模型页上的真实 ID

timeout: 12

max_retries: 1

model_info:

tier: instant

• model_name: tier-mini

litellm_params:

model: openrouter/openai/gpt-5.5-mini

timeout: 45

max_retries: 2

model_info:

tier: mini

• model_name: tier-reasoning

litellm_params:

model: openrouter/openai/gpt-5.5

timeout: 150

max_retries: 1

推理档常需要额外的"思考预算"类参数，

字段名和取值范围以官方文档为准，写错会直接 400

extra_body:

reasoning:

effort: medium

model_info:

tier: reasoning

router_settings:

routing_strategy: simple-shuffle

num_retries: 2

allowed_fails: 3

cooldown_time: 30

timeout: 180

fallbacks:

• tier-instant: ["tier-mini"]
• tier-mini: ["tier-reasoning"]
• tier-reasoning: ["tier-mini"]

```

两个细节值得说明：

为什么 instant 的 timeout 只有 12 秒？ 走 instant 的都是"用户正盯着屏幕等"的场景。它要是 30 秒还没回来，就算最后答对了，体验也已经崩了。这时候宁可降级到 mini 换个更靠谱的模型重试，也不要继续等。

fallbacks 的 key 必须是 model_name（别名），不是底层的 model。 这是最常见的配置错误，写错不会报错，只会静默失效。

第二步：把配置装进 LiteLLM Router

gateway/router_builder.py：

```python

import yaml

import litellm

from litellm import Router

让 LiteLLM 自动丢弃目标模型不支持的参数

例如某些推理模型不接受 temperature，不丢就会 400

litellm.drop_params = True

def build_router(config_path: str = "config.yaml") -> Router:

with open(config_path, "r", encoding="utf-8") as f:

cfg = yaml.safe_load(f)

rs = cfg.get("router_settings", {}) or {}

Router 的参数名以官方文档为准，这里只用最稳的几个

router = Router(

model_list=cfg["model_list"],

fallbacks=rs.get("fallbacks"),

num_retries=rs.get("num_retries", 2),

allowed_fails=rs.get("allowed_fails", 3),

cooldown_time=rs.get("cooldown_time", 30),

routing_strategy=rs.get("routing_strategy", "simple-shuffle"),

timeout=rs.get("timeout", 180),

set_verbose=False,

)

return router

```

注释里说的"以官方文档为准"不是客套话。LiteLLM 迭代很快，Router 的构造参数、fallbacks 的嵌套格式、extra_body 的透传方式都改过。你的 linter 报参数不存在时，第一件事是去翻当前版本的文档，而不是硬猜。

第三步：写路由分类器

分类器是整个网关里唯一"有业务判断"的地方，也是最值得反复打磨的地方。先用纯规则，别一上来就上模型分类——规则分类零延迟、零成本、可解释、能打日志复盘。等规则真的不够用了，再考虑用小模型做兜底。

gateway/policy.py：

```python

import re

from dataclasses import dataclass

TIER_INSTANT = "tier-instant"

TIER_MINI = "tier-mini"

TIER_REASONING = "tier-reasoning"

VALID_TIERS = (TIER_INSTANT, TIER_MINI, TIER_REASONING)

命中就倾向推理档

REASONING_PATTERNS = [

r"为什么", r"根因", r"证明", r"推导", r"逐步分析",

r"重构", r"架构", r"评审", r"权衡", r"对比.{0,6}方案",

r"traceback", r"stack ?trace", r"debug", r"报错.{0,10}分析",

]

典型"短平快"信号

SIMPLE_PATTERNS = [

r"^翻译", r"改写", r"润色", r"提取", r"分类", r"打标签",

r"转成\s*JSON", r"总结成一句话", r"起个标题", r"格式化",

]

各档的默认输出上限，防止 reasoning 档被滥用生成超长文本

DEFAULT_MAX_TOKENS = {

TIER_INSTANT: 512,

TIER_MINI: 1536,

TIER_REASONING: 4096,

}

@dataclass

class RouteDecision:

tier: str

model_group: str

max_tokens: int

reason: str

def _text_of(messages: list) -> str:

"""把所有消息的文本拼起来，兼容纯字符串和分段内容两种格式。"""

parts = []

for m in messages:

content = m.get("content")

if isinstance(content, str):

parts.append(content)

elif isinstance(content, list):

for seg in content:

if isinstance(seg, dict) and seg.get("type") == "text":

parts.append(seg.get("text", ""))

return "\n".join(parts)

def classify(

messages: list,

*,

priority: str = "balanced",

forced: str | None = None,

retry_count: int = 0,

has_tools: bool = False,

) -> RouteDecision:

0) 调用方明确指定，优先级最高

if forced in VALID_TIERS:

return RouteDecision(forced, forced, DEFAULT_MAX_TOKENS[forced], "调用方指定")

text = _text_of(messages)

n = len(text)

1) 成本优先：直接压到最便宜档

if priority == "cheap":

return RouteDecision(TIER_INSTANT, TIER_INSTANT, 512, "priority=cheap")

2) 重试即升级：第一次没答好，第二次加档重来

if retry_count >= 1:

return RouteDecision(

TIER_REASONING, TIER_REASONING, 4096,

f"第 {retry_count + 1} 次请求，自动升级",

)

3) 强推理信号

hits = [p for p in REASONING_PATTERNS if re.search(p, text, re.I)]

if hits or n > 8000 or has_tools:

why = []

if hits:

why.append(f"命中 {hits[:3]}")

if n > 8000:

why.append(f"超长输入 {n} 字符")

if has_tools:

why.append("带工具调用")

return RouteDecision(TIER_REASONING, TIER_REASONING, 4096, "；".join(why))

4) 短平快信号

simple = [p for p in SIMPLE_PATTERNS if re.search(p, text.strip(), re.I)]

if simple and n < 1500:

return RouteDecision(

TIER_INSTANT, TIER_INSTANT, 512,

f"命中 {simple[:2]}，输入仅 {n} 字符",

)

5) 默认落中档

return RouteDecision(TIER_MINI, TIER_MINI, 1536, f"默认中档，输入 {n} 字符")

```

有几个设计决定值得解释：

"重试即升级"是整套方案里性价比最高的一条。 用户点了重试，本身就是最强的"上次答得不好"信号，比任何分类器都准。第一次走 instant 花掉的时间和钱没有浪费，它帮你在真正需要质量的请求上省下了推理档的开销。

has_tools=True 直接进推理档，因为工具调用一旦参数格式错了，整个链路就断，省这点钱不划算。这个阈值你可以按自己的失败率调整。

n > 8000 直接进推理档是启发式，不是定理。长输入不一定难，但它一定贵，所以让它落到一个输出质量更稳的档位更安全。

分类器必须可观测。 每次决策都写进 reason 字段，这样你才能回答"为什么昨天成本涨了 40%"这类问题。

第四步：日志与用量统计

gateway/obs.py：

```python

import json

import logging

import threading

import litellm

logger = logging.getLogger("gateway.usage")

logger.setLevel(logging.INFO)

_handler = logging.StreamHandler()

_handler.setFormatter(logging.Formatter("%(message)s"))

logger.addHandler(_handler)

_lock = threading.Lock()

STATS: dict[str, dict] = {} # tier -> {calls, errors, latency_ms[], cost_usd}

def log_event(payload: dict) -> None:

"""打一行 JSON，方便直接喂给日志系统或 jq 统计。"""

logger.info(json.dumps(payload, ensure_ascii=False))

def estimate_cost(resp) -> float:

try:

LiteLLM 内置了各家的价格表；函数名与返回口径以官方文档为准

return float(litellm.completion_cost(completion_response=resp) or 0.0)

except Exception:

价格表里没有这个模型、或响应里没有 usage 时会走到这里

return 0.0

def record(request_id: str, decision, resp, latency_ms: int) -> None:

usage = getattr(resp, "usage", None)

cost = estimate_cost(resp)

with _lock:

s = STATS.setdefault(

decision.tier,

{"calls": 0, "errors": 0, "latency_ms": [], "cost_usd": 0.0},

)

s["calls"] += 1

s["latency_ms"].append(latency_ms)

s["cost_usd"] += cost

log_event({

"event": "route_ok",

"request_id": request_id,

"tier": decision.tier,

"model": getattr(resp, "model", None),

"reason": decision.reason,

"latency_ms": latency_ms,

"prompt_tokens": getattr(usage, "prompt_tokens", None),

"completion_tokens": getattr(usage, "completion_tokens", None),

"cost_usd": round(cost, 6),

})

def record_failure(request_id: str, tier: str, error: str) -> None:

with _lock:

s = STATS.setdefault(

tier, {"calls": 0, "errors": 0, "latency_ms": [], "cost_usd": 0.0}

)

s["errors"] += 1

log_event({"event": "route_failed", "request_id": request_id,

"tier": tier, "error": error[:500]})

```

关于成本估算，说三句实话：它是估算不是账单，价格表由 LiteLLM 维护、可能滞后；流式响应、缓存命中、部分供应商的计费口径会让数字对不上；completion_cost 的签名和返回口径以官方文档为准。它的价值在于相对比较——instant 档的成本占比是不是在涨，reasoning 档是不是被误触发了，这些趋势足够可靠。

第五步：拼成 FastAPI 网关

gateway/app.py：

```python

import time

import uuid

from fastapi import FastAPI, Header, HTTPException

from pydantic import BaseModel

from . import obs

from .policy import classify

from .router_builder import build_router

app = FastAPI(title="Multi-tier LLM Gateway")

router = build_router("config.yaml")

class ChatRequest(BaseModel):

messages: list

tier: str | None = None # 调用方可强制指定档位

max_tokens: int | None = None

temperature: float | None = None

stream: bool = False

retry_count: int = 0 # 客户端重试时把它 +1

@app.get("/healthz")

def healthz():

return {"ok": True}

@app.post("/v1/chat/completions")

async def chat(req: ChatRequest, x_priority: str = Header(default="balanced")):

if req.stream:

raise HTTPException(400, "示例先只做非流式，见'下一步建议'")

decision = classify(

req.messages,

priority=x_priority,

forced=req.tier,

retry_count=req.retry_count,

has_tools=any(m.get("role") == "tool" for m in req.messages),

)

request_id = uuid.uuid4().hex[:12]

started = time.perf_counter()

try:

Router 会按 fallbacks 自动重试和降级

resp = await router.acompletion(

model=decision.model_group,

messages=req.messages,

max_tokens=req.max_tokens or decision.max_tokens,

temperature=req.temperature,

metadata={"request_id": request_id, "tier": decision.tier},

)

except Exception as exc:

走到这里说明 Router 的重试和回退都已经用尽

obs.record_failure(request_id, decision.tier, str(exc))

raise HTTPException(

503, f"上游模型均不可用，请稍后重试（request_id={request_id}）"

)

latency_ms = round((time.perf_counter() - started) * 1000)

obs.record(request_id, decision, resp, latency_ms)

body = resp.model_dump() if hasattr(resp, "model_dump") else dict(resp)

body["x_route"] = {

"tier": decision.tier,

"model": getattr(resp, "model", None),

"reason": decision.reason,

"latency_ms": latency_ms,

"request_id": request_id,

}

return body

@app.get("/stats")

def stats():

out = {}

with obs._lock:

for tier, s in obs.STATS.items():

lat = sorted(s["latency_ms"])

n = len(lat)

out[tier] = {

"calls": s["calls"],

"errors": s["errors"],

"p50_ms": lat[n // 2] if n else None,

"p95_ms": lat[min(n - 1, int(n * 0.95))] if n else None,

"cost_usd": round(s["cost_usd"], 6),

}

return out

```

x_route 字段是刻意加的。调用方从响应里就能看到这次走了哪一档、为什么，排查问题时不用去翻服务端日志。

第六步：跑起来，做冒烟测试

```bash

uvicorn gateway.app:app --port 8080 --reload

```

先测最便宜的路径：

```bash

cat > req_cheap.json <<'JSON'

{"messages": [{"role": "user", "content": "把这句话改写得更礼貌：明天之前把报表发我。"}]}

JSON

curl -s http://127.0.0.1:8080/v1/chat/completions \

-H "Content-Type: application/json" \

-d @req_cheap.json | python -m json.tool

```

看 x_route.tier，应该是 tier-instant，reason 里写着"命中改写"。

再测推理路径：

```bash

cat > req_hard.json <<'JSON'

{"messages": [{"role": "user", "content": "这个脚本的 traceback 显示 KeyError: 'user_id'，帮我分析根因并给出重构方案。"}]}

JSON

curl -s http://127.0.0.1:8080/v1/chat/completions \

-H "Content-Type: application/json" \

-d @req_hard.json | python -m json.tool

```

这次应该落到 tier-reasoning，并且响应时间明显更长。

测一下强制省钱：

```bash

curl -s http://127.0.0.1:8080/v1/chat/completions \

-H "Content-Type: application/json" \

-H "x-priority: cheap" \

-d @req_hard.json | python -m json.tool

```

同一个难问题，应该落到 instant 档。

最后看汇总：

```bash

curl -s http://127.0.0.1:8080/stats | python -m json.tool

```

常见坑与排错

1. 回退没生效。 十有八九是 fallbacks 里写了底层 model 名字而不是 model_name 别名。回退配置不匹配时不会报错，只是不工作，一定要用上面"拔网线"的方式实测：把某个档的 API Key 临时改错，看请求是否自动落到下一档。

2. 参数被目标模型拒绝。 推理模型经常不吃 temperature、top_p、frequency_penalty。全局设 litellm.drop_params = True 能挡掉大部分，但 extra_body 里的自定义参数不会被自动清理，得自己按档位区别对待。

3. 超时和重试相乘。 timeout 是单次尝试的时间，num_retries 是重试次数，两者叠起来才是最坏延迟。instant 档设 12 秒超时 + 2 次重试，最坏可能等 36 秒以上——这已经违背了"instant"的初衷。快档就应该少重试，让回退去承担质量兜底。

4. 429 限流下的连锁雪崩。 一个模型被限流，回退把流量全压到下一档，下一档再被压垮。cooldown_time 和 allowed_fails 就是干这个用的：失败几次就把这个模型临时踢出池子，冷却期过了再放回来。这两个值和你的配额有关，需要按实际流量调。

5. 成本数字对不上账单。 前面说过，completion_cost 是估算。别拿它做财务口径，拿它做趋势和比例：哪个档占了总成本的多少、每次模型 ID 变更后成本曲线有没有跳变。

6. 分类器把流量全吸到一个档。 上线第一天一定先跑"影子模式"：只记录 classify() 的结果，不真的改路由，看分布是否符合预期。如果 90% 的请求都落到 reasoning 档，那你的规则写得太激进了。

7. 关键词匹配误伤。 "帮我翻译这段代码的报错信息"——既有"翻译"又有"报错"。规则顺序很重要：强推理信号要先于短平快信号判断。宁可多花点钱，也别在需要动脑的问题上糊弄。

8. 模型 ID 和 provider 前缀写错。 LiteLLM 用前缀决定走哪家，比如 openrouter/、openai/、azure/。前缀写错通常报"找不到模型"或者认证失败，具体前缀规则以官方文档为准。

9. Key 泄进 Git。 别把 Key 写进 config.yaml。让 LiteLLM 从环境变量读，或者用 os.environ/VAR_NAME 这种写法（这是 Proxy 配置的语法，SDK 场景下直接省略 api_key 让它读环境变量更省事）。

10. 流式没做。 上面代码遇到 stream=true 直接返回 400。流式需要把 Router 的异步生成器转成 SSE 逐块吐出去，还要在 finally 里补记用量（流式响应的 token 数常常要等最后一个 chunk 才有）。这是下一步该做的事。

下一步建议

接流式。 这是唯一挡在生产可用性前面的硬需求。思路是把 router.acompletion(stream=True) 返回的异步迭代器包成 StreamingResponse，用 finally 块保证日志一定落盘。

让分类器学会自我修正。 把每条 route_ok 日志里的 tier、reason、latency_ms、以及后续有没有触发重试（retry_count > 0 的那条请求）关联起来。重试率最高的那类请求，就是分类器判断错了的样本。拿这些样本去调规则，比拍脑袋加关键词有效得多。

加预算熔断。 维护一个"本月已花费"的计数器，超过阈值就让 classify 强制返回 instant 档，并给调用方返回一个 x_budget_warning 头。成本失控的时候，能降级比不能降级强。

上 LiteLLM Proxy。 现在这套是你自己写的薄网关。当你要做虚拟 Key、团队级配额、按用户的用量看板时，LiteLLM 自带的 Proxy 模式（litellm --config config.yaml）能省很多活，它和本文的 model_list 格式基本兼容。代价是路由逻辑要换个方式挂上去，具体挂法以官方文档为准。

最后一条，也是最重要的：先把三档的模型 ID 和超时定下来，再动手写代码。 这套架构的价值不在于代码多精巧，而在于你把"什么任务值得花多少钱"这件事，从每个人的直觉变成了一个可检查、可调整、可回滚的配置文件。