LiteLLM 适配 176 个大模型:构建统一 LLM Gateway 实践
做 AI 应用最痛苦的事之一:每家 LLM 厂商的 SDK 都不一样。OpenAI 一套、Anthropic 一套、Gemini 一套、国内厂商又各一套。你的代码里到处是
if model.startswith("gpt")的分支。LiteLLM 用一个 OpenAI 兼容的接口,把 176 个模型统一成一种调用方式。本文是 IHUI-AI 在apps/ai-service/app/core/llm_gateway.py落地 LLM Gateway 的工程总结。
一、为什么需要 LLM Gateway
1.1 没有 Gateway 的痛苦
# 早期代码:到处是 if-else
def call_llm(model: str, messages: list):
if model.startswith("gpt"):
from openai import OpenAI
client = OpenAI()
return client.chat.completions.create(model=model, messages=messages)
elif model.startswith("claude"):
import anthropic
client = anthropic.Anthropic()
return client.messages.create(model=model, messages=messages)
elif model.startswith("gemini"):
import google.generativeai as genai
genai.configure(api_key=os.getenv("GEMINI_KEY"))
# 完全不同的 API 形状
...
elif model.startswith("deepseek"):
# 又一个 SDK
...
# 176 个模型 = 176 套分支
问题:
- 接口不统一:每家 SDK 的入参/出参/流式格式都不同
-
错误处理散乱:OpenAI 抛
RateLimitError,Anthropic 抛APITimeoutError,没法统一 catch - 成本统计难:每家计价单位不同(token 定义都不一样)
- 故障转移难:GPT-4 限流了想切 Claude,要写一堆切换逻辑
1.2 LLM Gateway 的四大价值
| 价值 | 说明 |
|---|---|
| 统一接口 | 所有模型用 OpenAI 兼容格式调用,一套代码全覆盖 |
| 成本控制 | 统一计费、预算告警、租户配额 |
| 故障转移 | 主模型挂了自动切备用,业务无感 |
| 审计合规 | 所有调用记录:谁、何时、调了什么、花了多少、结果 |
二、LiteLLM 设计原理
LiteLLM 的核心思想:所有 LLM 都能被映射成 OpenAI Chat Completions API。
2.1 三层架构
┌─────────────────────────────────────┐
│ 你的应用(OpenAI 兼容调用) │
└──────────────┬──────────────────────┘
│ uniform interface
┌──────────────▼──────────────────────┐
│ LiteLLM Router / Proxy │
│ - 路由策略 │
│ - 成本控制 │
│ - 故障转移 │
└──────────────┬──────────────────────┘
│ translate
┌──────────────▼──────────────────────┐
│ 各厂商 SDK / API │
│ OpenAI | Anthropic | Gemini | ... │
└─────────────────────────────────────┘
2.2 统一调用示例
from litellm import completion
# 调 OpenAI
response = completion(model="gpt-4o", messages=[{"role": "user", "content": "Hi"}])
# 调 Anthropic(完全一样的代码,只改 model 名)
response = completion(model="claude-3-5-sonnet-20241022", messages=[...])
# 调 Gemini
response = completion(model="gemini/gemini-1.5-pro", messages=[...])
# 调 DeepSeek
response = completion(model="deepseek/deepseek-chat", messages=[...])
# 调本地 Ollama
response = completion(model="ollama/llama3", messages=[...], api_base="http://localhost:11434")
返回的 response 对象结构完全一致(choices[0].message.content),无需关心底层差异。
三、176 个模型的适配分类
LiteLLM 支持的 176 个模型按厂商分类:
3.1 国际厂商
| 厂商 | model 前缀 | 代表模型 |
|---|---|---|
| OpenAI | gpt-* |
gpt-4o, gpt-4o-mini, o1-preview |
| Anthropic | claude-* |
claude-3-5-sonnet, claude-3-opus |
gemini/* |
gemini-1.5-pro, gemini-1.5-flash | |
| Mistral | mistral/* |
mistral-large, mistral-small |
| Cohere | command-r-* |
command-r, command-r-plus |
| Azure | azure/<deploy> |
Azure 托管的 OpenAI |
| AWS | bedrock/* |
Bedrock 托管的多厂商 |
3.2 国内厂商(IHUI-AI 重点)
| 厂商 | model 前缀 | 代表模型 |
|---|---|---|
| DeepSeek | deepseek/* |
deepseek-chat, deepseek-coder |
| 通义 Qwen | dashscope/* |
qwen-max, qwen-plus, qwen-turbo |
| 智谱 GLM | zhipu/* |
glm-4, glm-4-plus, glm-4-flash |
| 字节豆包 | volcengine/* |
doubao-pro, doubao-lite |
| 月之暗面 | moonshot/* |
moonshot-v1-8k, moonshot-v1-32k |
| MiniMax | minimax/* |
abab6.5, abab6.5s |
| 讯飞星火 | spark/* |
spark-v3.5, spark-v4 |
| 百度文心 | wenxin/* |
ernie-bot-4, ernie-bot-turbo |
3.3 开源自托管
| 厂商 | model 前缀 | 代表模型 |
|---|---|---|
| Ollama | ollama/* |
llama3, qwen2, mistral |
| vLLM | vllm/* |
任意 HuggingFace 模型 |
| LlamaCpp | llamacpp/* |
本地 GGUF 模型 |
| LM Studio | lm_studio/* |
LM Studio 桌面端 |
国内场景下,IHUI-AI 默认推荐 DeepSeek(性价比高)+ Qwen(阿里云生态)+ GLM(国产合规),配合 Ollama 做本地开发。
四、路由策略
LLM Gateway 的核心是 router,决定「这次调用走哪个模型」。
4.1 Router 配置
# apps/ai-service/app/core/llm_gateway.py
from litellm import Router
router = Router(model_list=[
{
"model_name": "default", # 业务侧用的别名
"litellm_params": {
"model": "deepseek/deepseek-chat",
"api_key": os.getenv("DEEPSEEK_KEY"),
},
},
{
"model_name": "default", # 同名 = 同组,故障转移候选
"litellm_params": {
"model": "gpt-4o-mini",
"api_key": os.getenv("OPENAI_KEY"),
},
},
{
"model_name": "premium",
"litellm_params": {
"model": "claude-3-5-sonnet-20241022",
"api_key": os.getenv("ANTHROPIC_KEY"),
},
},
{
"model_name": "premium",
"litellm_params": {
"model": "gpt-4o",
"api_key": os.getenv("OPENAI_KEY"),
},
},
])
# 业务侧:用别名,router 决定实际走哪个
response = router.completion(model="default", messages=[...])
4.2 路由策略对比
| 策略 | 配置 | 适合场景 |
|---|---|---|
| 成本优先 | routing_strategy="cost-based" |
免费用户、内部测试 |
| 性能优先 | routing_strategy="latency-based-routing" |
付费用户、实时场景 |
| 故障转移 | fallbacks=["premium"] |
高可用要求 |
| 地域优先 | 自定义权重 | 合规、延迟优化 |
| 轮询 | routing_strategy="simple-shuffle" |
负载均衡 |
4.3 故障转移实战
router = Router(
model_list=[...],
fallbacks=[
{"default": ["premium"]}, # default 全挂 → 切 premium
{"premium": ["default"]}, # premium 全挂 → 降级 default
],
routing_strategy="latency-based-routing",
num_retries=2,
retry_after=1,
timeout=30,
)
# DeepSeek 限流 → 自动切 GPT-4o-mini
response = await router.acompletion(model="default", messages=[...])
实测:DeepSeek 高峰期限流率约 5%,有 fallback 后业务侧完全无感。
五、成本优化
5.1 缓存
相同 prompt + model 的结果缓存(Redis):
from litellm import completion
from litellm.caching import Cache
litellm.cache = Cache(
type="redis",
host=os.getenv("REDIS_HOST"),
port=6379,
ttl=3600, # 1 小时
)
# 相同输入第二次走缓存,0 成本
response = completion(model="gpt-4o", messages=[...], caching=True)
适合:知识库 QA(同一问题反复问)、分类标签(同一段文本分类稳定)。不适合:创意写作、对话(每次都不同)。
5.2 批处理
多个独立请求合并成一个 batch,部分厂商支持降价 50%:
from litellm import batch_completion
responses = batch_completion(
model="gpt-4o-mini",
messages=[
[{"role": "user", "content": "翻译: Hello"}],
[{"role": "user", "content": "翻译: World"}],
[{"role": "user", "content": "翻译: Goodbye"}],
],
)
OpenAI 的 Batch API 异步批处理,24 小时内返回,价格减半。
5.3 模型降级
简单任务用小模型,复杂任务才用大模型:
def smart_route(messages: list) -> str:
total_tokens = sum(estimate_tokens(m["content"]) for m in messages)
last_msg = messages[-1]["content"]
# 简单问候/闲聊 → 小模型
if total_tokens < 100 and len(last_msg) < 50:
return "default" # deepseek-chat / gpt-4o-mini
# 长文本/复杂推理 → 大模型
if total_tokens > 5000 or "分析" in last_msg or "代码" in last_msg:
return "premium" # claude / gpt-4o
return "default"
model = smart_route(messages)
response = await router.acompletion(model=model, messages=messages)
实测:接入 smart_route 后,平均成本下降 60%,因为 80% 的请求其实只需要小模型。
5.4 token 估算
import tiktoken
def estimate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
pricing = {
"gpt-4o": {"input": 2.5e-6, "output": 10e-6},
"gpt-4o-mini": {"input": 0.15e-6, "output": 0.6e-6},
"claude-3-5-sonnet":{"input": 3e-6, "output": 15e-6},
"deepseek-chat": {"input": 0.14e-6, "output": 0.28e-6},
}
p = pricing.get(model, {"input": 1e-6, "output": 2e-6})
return input_tokens * p["input"] + output_tokens * p["output"]
DeepSeek 比 GPT-4o 便宜约 18 倍,这是国内 SaaS 的成本优势所在。
六、流式响应(SSE)+ 客户端断连
6.1 流式输出
from litellm import acompletion
async def stream_chat(model: str, messages: list):
response = await acompletion(
model=model,
messages=messages,
stream=True,
)
async for chunk in response:
delta = chunk.choices[0].delta.content
if delta:
yield f"data: {json.dumps({'content': delta})}\n\n"
yield "data: [DONE]\n\n"
6.2 客户端断连处理
用户关浏览器时,SSE 连接断开,但 LLM 还在生成(继续计费)。要监听断开事件取消请求:
from fastapi import Request
from fastapi.responses import StreamingResponse
import asyncio
@app.post("/chat/stream")
async def chat_stream(request: Request, body: ChatBody):
async def event_stream():
task = asyncio.create_task(stream_chat(body.model, body.messages))
try:
async for chunk in task:
if await request.is_disconnected():
task.cancel() # 客户端断了,取消 LLM 调用
break
yield chunk
except asyncio.CancelledError:
task.cancel()
raise
return StreamingResponse(event_stream(), media_type="text/event-stream")
这一步能省下 10-20% 的 token 成本(用户中途关掉的请求不会继续跑)。
七、限流与配额
7.1 per-tenant / per-user / per-model
from litellm import Router
router = Router(
model_list=[...],
router_budget_limit={
"default": {"tenant_free": 1000, "tenant_pro": 100000}, # tokens/day
},
routing_strategy="usage-based-routing-v2",
)
# 每次调用前 router 检查租户今日用量
response = await router.acompletion(
model="default",
messages=[...],
metadata={"tenant_id": tenant_id, "tenant_tier": "free"},
)
超额时 router 抛 BudgetExceededError,业务侧返回 429。
7.2 速率限制
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
@app.post("/chat")
@limiter.limit("30/minute") # 每 IP 每分钟 30 次
async def chat(request: Request, body: ChatBody):
...
八、IHUI-AI 实战:llm_gateway.py
apps/ai-service/app/core/
├── llm_gateway.py # LiteLLM router 封装
├── llm_config.py # 模型配置(从 DB 加载)
├── llm_cost.py # 成本统计
└── llm_audit.py # 审计日志
8.1 统一入口
# apps/ai-service/app/core/llm_gateway.py
from litellm import Router
from litellm.exceptions import BudgetExceededError, RateLimitError
import structlog
logger = structlog.get_logger()
class LLMGateway:
def __init__(self):
self.router = Router(model_list=self._load_models())
self.router.fallbacks = self._load_fallbacks()
async def chat(
self,
messages: list,
model_alias: str = "default",
tenant_id: str = "",
user_id: str = "",
stream: bool = False,
) -> dict:
metadata = {"tenant_id": tenant_id, "user_id": user_id}
try:
response = await self.router.acompletion(
model=model_alias,
messages=messages,
stream=stream,
metadata=metadata,
)
# 记录审计
await self._audit(tenant_id, user_id, model_alias, response)
return response
except BudgetExceededError:
logger.warning("budget_exceeded", tenant=tenant_id)
raise HTTPException(429, "今日额度已用完")
except RateLimitError:
logger.warning("rate_limited", model=model_alias)
raise HTTPException(429, "模型繁忙,请稍后")
async def _audit(self, tenant_id, user_id, model, response):
# 写入 llm_calls 表,记录 token / 成本 / 状态
await db.execute(
"INSERT INTO llm_calls (tenant_id, user_id, model, input_tokens, output_tokens, cost_usd, status) VALUES ($1,$2,$3,$4,$5,$6,$7)",
tenant_id, user_id, model,
response.usage.prompt_tokens,
response.usage.completion_tokens,
estimate_cost(model, response.usage.prompt_tokens, response.usage.completion_tokens),
"success",
)
gateway = LLMGateway()
8.2 模型配置从 DB 加载
admin 后台可以动态增删模型,无需重启:
async def _load_models(self) -> list:
rows = await db.fetch("SELECT * FROM llm_models WHERE enabled = TRUE")
return [{
"model_name": row["alias"], # 业务别名,如 "default"
"litellm_params": {
"model": row["provider_model"], # 实际模型名,如 "deepseek/deepseek-chat"
"api_key": decrypt(row["api_key_encrypted"]),
"api_base": row.get("api_base"),
},
} for row in rows]
九、监控:token / 错误率 / 延迟 / 成本
9.1 监控仪表盘
-- 今日各模型 token 消耗
SELECT model,
COUNT(*) AS calls,
SUM(input_tokens) AS input_toks,
SUM(output_tokens) AS output_toks,
SUM(cost_usd) AS cost
FROM llm_calls
WHERE created_at::date = CURRENT_DATE
GROUP BY model ORDER BY cost DESC;
-- 错误率
SELECT model,
COUNT(*) FILTER (WHERE status != 'success') * 100.0 / COUNT(*) AS error_rate
FROM llm_calls
WHERE created_at > NOW() - INTERVAL '1 hour'
GROUP BY model;
-- P95 延迟
SELECT model,
PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY duration_ms) AS p95
FROM llm_calls
GROUP BY model;
9.2 告警
- 单租户 1 小时成本 > $10 → 告警(可能有滥用)
- 某模型错误率 > 10% → 告警(可能厂商故障,触发 fallback)
- P95 延迟 > 10s → 告警(模型过载)
十、与 MCP 协议的协同
LiteLLM 把「选哪个模型」收敛成 model="default",而 MCP 把「调哪个工具」收敛成 tools=[...]。两者结合:
async def smart_agent(messages: list, tenant_id: str):
# 1. 用 LiteLLM 选模型(成本/性能路由)
model = await gateway.pick_model(messages, tenant_id)
# 2. 用 MCP 拉工具(按租户权限)
tools = await mcp_registry.discover_tools(tenant_id)
# 3. 统一调用
response = await gateway.chat(
messages=messages,
model_alias=model,
tools=tools, # LiteLLM 自动转成厂商的 tool calling 格式
)
return response
LiteLLM 内部会把 OpenAI 格式的 tools 翻译成 Anthropic 的 tools、Gemini 的 function_declarations 等,业务侧只写一份。
更进一步:可以把「模型选择」本身做成一个 MCP 工具,让 agent 自己根据任务复杂度选模型(简单任务选便宜模型,复杂任务选强模型)。
十一、参考资料
- LiteLLM 官方文档
- LiteLLM 支持的 176 个模型清单
- LiteLLM Router 策略
- LiteLLM Proxy Server
- IHUI-AI 源码:
apps/ai-service/app/core/llm_gateway.py
总结
LLM Gateway 不是「多此一举」,而是 AI 应用从 demo 走向生产的必经之路。LiteLLM 用 OpenAI 兼容接口统一 176 个模型,解决了:
- 接口碎片化:一套代码调所有厂商
- 成本失控:缓存 + 批处理 + 模型降级,成本砍 60%
- 故障频发:fallback 自动切备用,业务无感
- 审计缺失:每次调用记录 token/成本,精准计费
IHUI-AI 的实践配置:DeepSeek(默认,性价比)+ GPT-4o-mini(fallback)+ Claude 3.5(premium),配合 Redis 缓存和 smart_route,单次对话平均成本控制在 $0.001 以内。
下一篇讲 Monorepo 8 端同步开发,看所有这些服务如何在一个仓库里协同构建。
This article was originally published on the IHUI AI Blog. Follow us on GitHub for more AI engineering content.
Top comments (0)