1. 项目概述:LiteLLM 是什么,它解决的不是“模型跑不起来”,而是“模型调用太重、太碎、太难管”
LiteLLM 这个名字刚看到时,很多人会下意识以为是个轻量级大语言模型——毕竟带个 “Lite” 前缀,又跟 LLM 挨着。我第一次在某开源社区刷到它时也这么想,顺手点开 README 就愣住了:它压根不训练模型,也不推理模型,甚至连一行模型权重加载代码都没有。它干的事儿特别“务虚”:在你和所有主流大模型 API 之间,插上一层薄如蝉翼、却稳如磐石的统一调度层。你可以把它理解成大模型世界的“万能充电头”——不管你的设备是 USB-C、Lightning 还是老式 Micro-USB,只要插上这个头,就能用同一根线、同一个电压、同一套指示灯来供电。而 LiteLLM 的“线”就是标准 OpenAI 兼容的 REST 接口,“电压”是你传进去的 messages 数组,“指示灯”就是返回的 completion 字段。
它真正瞄准的痛点,是当前大模型应用开发中那种令人窒息的“碎片化疲劳”。比如你写一个客服对话系统,初期用 OpenAI 的 gpt-3.5-turbo,成本可控;但客户突然要求支持中文更强的 Qwen2-72b,你得去阿里云平台申请密钥、读一遍新文档、改掉所有openai.ChatCompletion.create调用,再加一套错误重试逻辑;过两周又说要接入本地部署的 Llama3-8B,你又得研究 Ollama 的/api/chat接口格式、处理流式响应的 chunk 解析……最后你的代码里散落着七八种不同风格的if model_provider == 'xxx'分支,日志里全是response['choices'][0]['message']['content']和response['message']['content']的混搭,测试用例写到第三版就没人敢动了。LiteLLM 就是来终结这种状态的——它强制你只写一次调用逻辑,后面所有模型切换、密钥管理、重试策略、负载均衡、甚至费用统计,都由它在后台默默完成。
它的核心价值不在“快”,而在“省心”和“可扩展”。一个刚毕业的实习生,用它三分钟就能把本地 Ollama 的 Llama3 和云端 Anthropic 的 Claude 3 Opus 写进同一个completion()函数里做 A/B 测试;一个运维老手,靠它把公司所有模型 API 密钥集中托管在 Vault 里,前端服务完全不碰敏感信息;一个架构师,则用它作为模型网关,在流量高峰时自动把 70% 的请求切到便宜的 Mixtral,30% 留给高保真的 GPT-4 Turbo,整个过程对业务代码零侵入。所以如果你正在为“模型太多、接口太乱、切换太烦”而深夜改 if-else,LiteLLM 就是你该立刻打开终端敲pip install litellm的那个工具。它不替代你的模型,它让你终于能把注意力从“怎么调通”回到“怎么用好”。
2. 核心设计思路与方案选型逻辑:为什么是“协议转换器”,而不是“模型代理”或“推理框架”
LiteLLM 的设计哲学,可以用一句话概括:不做减法,只做加法;不碰模型,只管管道。这决定了它和同类工具(比如 vLLM、Text Generation Inference、Ollama)有本质区别。vLLM 是在 GPU 显存里做极致优化,目标是让单卡吞吐翻倍;TGI 是在模型加载和 KV Cache 管理上死磕,目标是降低延迟;Ollama 则是把模型打包成 Docker 那样的镜像,目标是简化本地部署。而 LiteLLM 的战场,是在 HTTP 请求发出前的最后一毫秒,和响应返回后的第一微秒之间。
为什么选择这条“中间件”路线?我们拆解三个关键决策点:
2.1 不封装模型,只封装协议:规避硬件与生态绑定风险
很多团队早期会尝试自己写一个“通用模型 SDK”,把 OpenAI、Anthropic、Google 的接口都封装一遍。但很快就会发现,这种 SDK 会迅速腐化:Anthropic 加了个system字段,你得改;Google Vertex AI 更新了认证方式,你得加新类;Azure OpenAI 强制要求api-version头,你得在每个请求里塞。LiteLLM 的解法极其聪明——它不自己维护任何模型适配器,而是直接复用各厂商官方 SDK 的底层 HTTP 客户端(比如openai、anthropic、google-generativeai),只在它们之上加一层“协议翻译层”。当你调用litellm.completion(model="gpt-3.5-turbo", messages=...)时,LiteLLM 并不自己发请求,而是把你的参数翻译成标准 OpenAI 格式,再原封不动交给openai.OpenAI().chat.completions.create()执行。这意味着:它永远和官方 SDK 同步更新,你永远不用操心某个模型字段的兼容性问题。我去年在某金融项目里用它对接 Azure OpenAI,微软突然把api-version从2023-05-15升级到2024-02-01,我们连代码都没动,只升级了openai包,LiteLLM 就自动适配了——因为它是“借势”,不是“硬扛”。
2.2 强制 OpenAI 兼容接口:用事实标准终结接口战争
LiteLLM 的所有对外 API,都严格遵循 OpenAI 的/v1/chat/completions规范。这不是为了讨好 OpenAI,而是因为这个规范已经成了事实上的行业通用语。几乎所有大模型平台(包括非 OpenAI 系的)都在自己的 API 文档里写着“兼容 OpenAI 格式”。LiteLLM 把这个“兼容”做成了铁律:无论后端是 Groq 的 LPU、Fireworks 的推理集群,还是你自己用 vLLM 跑的私有模型,只要它暴露一个 OpenAI 兼容的 endpoint,LiteLLM 就能无缝接入。这种设计带来两个巨大好处:一是前端开发极度简单——你用openai包写的 demo,换行改成litellm,就能跑通所有模型;二是生态工具链零迁移成本,LangChain、LlamaIndex、DSPy 这些主流框架,原本就深度集成 OpenAI,现在只需改一个导入路径,就能获得多模型支持。我见过最夸张的案例,是某教育 SaaS 公司,他们用 LangChain 构建的作文批改 pipeline,原来只跑 GPT-4,上线 LiteLLM 后,三天内就完成了向 Qwen2、GLM-4、DeepSeek-Coder 的全模型覆盖,而 LangChain 的 chain 代码一行没改。
2.3 轻量级进程内运行:拒绝引入复杂基础设施
有些团队一听到“统一网关”,第一反应就是上 Kubernetes + Envoy + Istio。LiteLLM 坚决拒绝这条路。它的核心是一个纯 Python 库,启动时只初始化一个全局的litellm.Router实例,所有模型调用都在当前 Python 进程内完成路由、重试、日志。没有额外的 HTTP 服务进程,没有需要单独部署的 sidecar,没有复杂的 service mesh 配置。这种设计对中小团队极其友好:你不需要专门招一个“模型平台工程师”,运维同学不用学新的监控指标,开发同学调试时print(response)看到的就是最终结果,而不是一堆中间代理的日志。当然,它也提供了litellm --port 4000的简易 HTTP server 模式,但这只是为那些必须用 curl 或 Postman 测试的场景准备的“临时插座”,不是它的主干。我在某电商公司的推荐系统里把它嵌入到 FastAPI 服务中,整个模型路由模块就占 20 行代码,监控指标直接打到 Prometheus,和订单服务的日志混在一起,运维同学说“比查 Redis 慢日志还简单”。
3. 核心功能解析与实操要点:从“能用”到“用稳”的关键细节
LiteLLM 的安装和基础调用简单到令人发指:pip install litellm,然后import litellm; litellm.completion(...)。但真正让它从玩具变成生产级工具的,是那些藏在文档角落、却决定系统生死的配置项。下面这些,是我踩过坑、调过参、线上跑过半年才敢写出来的硬核要点。
3.1 模型路由(Router):不只是负载均衡,更是业务策略引擎
LiteLLM 的Router不是简单的 round-robin 轮询。它支持基于模型能力、价格、延迟、甚至自定义规则的智能路由。比如你有一组模型:gpt-4-turbo(贵、强)、claude-3-haiku(便宜、快)、llama3-70b(本地、可控)。你可以这样配置:
from litellm import Router router = Router( model_list=[ { "model_name": "high_quality", "litellm_params": { "model": "gpt-4-turbo", "api_key": os.getenv("OPENAI_API_KEY"), "max_retries": 3 } }, { "model_name": "fast_response", "litellm_params": { "model": "claude-3-haiku", "api_key": os.getenv("ANTHROPIC_API_KEY"), "max_retries": 2 } } ], routing_strategy="usage-based" )这里routing_strategy="usage-based"是关键——它会让 LiteLLM 根据每个模型的历史调用量(按 token 计费)自动分配请求,确保总成本不超预算。更狠的是,你可以写自定义函数:
def route_based_on_content(messages): # 检查用户消息是否含“代码”、“debug”等关键词 content = " ".join([m.get("content", "") for m in messages]) if "code" in content.lower() or "debug" in content.lower(): return "llama3-70b" # 代码类请求走本地大模型 else: return "fast_response" # 其他走便宜的 Haiku router = Router( model_list=[...], routing_strategy="custom", custom_routing_strategy=route_based_on_content )提示:自定义路由函数必须是纯函数,不能有副作用(比如修改全局变量)。我曾在一个项目里误把数据库查询塞进去,导致每次路由都触发一次 DB 连接,QPS 直接崩到 5。正确做法是把业务规则缓存到内存字典里,定期刷新。
3.2 重试与熔断:别让单点故障拖垮整个系统
默认情况下,LiteLLM 对网络超时、5xx 错误会自动重试 3 次。但这远远不够。真正的生产环境需要分层重试策略:
- 第一层:快速失败——对
429 Too Many Requests(限流)和401 Unauthorized(密钥失效),必须立即失败并告警,重试毫无意义; - 第二层:智能重试——对
503 Service Unavailable,可以重试,但要指数退避(1s, 2s, 4s); - 第三层:熔断降级——如果某个模型连续 5 次失败,自动将其从路由池中剔除 60 秒,避免雪崩。
LiteLLM 通过fallbacks和num_retries组合实现:
router = Router( model_list=[...], fallbacks=[ # 当主模型失败时,依次降级 {"model": "high_quality", "fallbacks": ["fast_response"]}, {"model": "fast_response", "fallbacks": ["llama3-70b"]} ], num_retries=1 # Router 层只重试 1 次,把重试逻辑交给下游模型 )注意:
num_retries设为 1 是经验之谈。设太高会导致请求在 Router 层堆积,设为 0 则失去兜底能力。我们在线上压测发现,1 次重试 + 3 级 fallback 的组合,在 99.99% 的故障场景下都能恢复,且平均延迟增加不到 200ms。
3.3 Token 统计与成本追踪:让每一分钱都花得明白
LiteLLM 会在每次completion()返回的response中,自动注入response._hidden_params字段,里面包含精确的prompt_tokens、completion_tokens、total_tokens,以及按当前模型价格计算出的cost(单位:美元)。这玩意儿对财务审计简直是救命稻草。你可以轻松写出这样的监控逻辑:
def log_cost(response): cost = response._hidden_params.get("cost", 0) model = response._hidden_params.get("model", "unknown") if cost > 0.01: # 单次请求超 1 美分,发企业微信告警 send_alert(f"高成本请求:{model},${cost:.4f}") response = router.completion(model="high_quality", messages=...) log_cost(response)更绝的是,它支持litellm.token_counter()独立调用,让你在请求发出前就预估成本,做实时预算控制:
# 用户提交长文本前,先估算 token 数和成本 prompt = "..." * 1000 estimated_tokens = litellm.token_counter(model="gpt-4-turbo", text=prompt) estimated_cost = estimated_tokens * 0.01 / 1000 # gpt-4-turbo $0.01/1k prompt tokens if estimated_cost > 0.5: # 超过 50 美分,提示用户精简 raise ValueError("输入过长,请减少至 500 字以内")4. 完整实操流程:从零搭建一个支持三模型的生产级问答服务
现在我们动手做一个真实可用的服务:一个 Web API,接收用户问题,自动选择最优模型回答,并记录完整日志和成本。整个过程不依赖任何外部服务,全部在一台 16G 内存的服务器上跑通。
4.1 环境准备与依赖安装
我们用 Poetry 管理依赖,确保环境干净:
# 创建项目目录 mkdir llm-router-demo && cd llm-router-demo # 初始化 poetry poetry init -n poetry add litellm fastapi uvicorn python-dotenv poetry add --group dev pytest black # 创建 .env 文件,填入你的 API 密钥 echo "OPENAI_API_KEY=sk-..." > .env echo "ANTHROPIC_API_KEY=sk-ant-api03-..." >> .env echo "GROQ_API_KEY=gsk_..." >> .env注意:LiteLLM 支持从环境变量自动读取密钥,无需在代码里硬编码。这是安全底线,我见过太多项目因为把密钥写进 Git 而被扫库。
4.2 核心路由配置(router_config.py)
这是整个系统的“大脑”,我们配置三个模型:OpenAI 的 GPT-4 Turbo(质量优先)、Anthropic 的 Claude 3 Haiku(速度优先)、Groq 的 Llama3-70b(本地推理优先):
import os from litellm import Router # 从环境变量读取密钥,失败则抛异常 def get_env_var(key: str) -> str: value = os.getenv(key) if not value: raise ValueError(f"Missing required env var: {key}") return value # 模型列表,按性能和成本排序 model_list = [ { "model_name": "gpt-4-turbo", "litellm_params": { "model": "gpt-4-turbo", "api_key": get_env_var("OPENAI_API_KEY"), "api_base": "https://api.openai.com/v1", "max_retries": 2, "timeout": 60 } }, { "model_name": "claude-3-haiku", "litellm_params": { "model": "claude-3-haiku-20240307", "api_key": get_env_var("ANTHROPIC_API_KEY"), "api_base": "https://api.anthropic.com/v1", "max_retries": 2, "timeout": 30 } }, { "model_name": "llama3-70b-groq", "litellm_params": { "model": "groq/llama3-70b-8192", "api_key": get_env_var("GROQ_API_KEY"), "api_base": "https://api.groq.com/openai/v1", "max_retries": 1, "timeout": 45 } } ] # 初始化 Router,启用 usage-based 路由和 fallback router = Router( model_list=model_list, routing_strategy="usage-based", num_retries=1, fallbacks=[ {"model": "gpt-4-turbo", "fallbacks": ["claude-3-haiku"]}, {"model": "claude-3-haiku", "fallbacks": ["llama3-70b-groq"]} ] )4.3 FastAPI 服务(main.py)
把 Router 封装成 Web API,加入完整的错误处理和日志:
from fastapi import FastAPI, HTTPException, BackgroundTasks from pydantic import BaseModel import logging import time from datetime import datetime from router_config import router # 配置日志 logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler()] ) logger = logging.getLogger("llm-router") app = FastAPI(title="LiteLLM Router API", version="1.0") class ChatRequest(BaseModel): messages: list[dict] model: str | None = None # 可选,指定模型名,否则自动路由 @app.post("/v1/chat/completions") async def chat_completions(request: ChatRequest, background_tasks: BackgroundTasks): start_time = time.time() try: # 调用 LiteLLM Router response = await router.acompletion( model=request.model or "gpt-4-turbo", # 默认走高质量 messages=request.messages, stream=False ) # 计算耗时和成本 duration = time.time() - start_time cost = response._hidden_params.get("cost", 0) model_used = response._hidden_params.get("model", "unknown") # 记录结构化日志 logger.info( f"SUCCESS - model={model_used} " f"duration={duration:.2f}s " f"prompt_tokens={response.usage.prompt_tokens} " f"completion_tokens={response.usage.completion_tokens} " f"cost=${cost:.6f}" ) # 异步发送成本统计到监控系统(此处模拟) background_tasks.add_task(log_to_monitoring, model_used, cost, duration) return response.dict() except Exception as e: duration = time.time() - start_time error_msg = str(e) logger.error(f"ERROR - duration={duration:.2f}s error={error_msg}") raise HTTPException(status_code=500, detail=f"LLM call failed: {error_msg}") def log_to_monitoring(model: str, cost: float, duration: float): """模拟发送指标到 Prometheus 或 Datadog""" # 实际项目中这里会调用 statsd 或 prometheus_client pass # 健康检查端点 @app.get("/health") def health_check(): return {"status": "ok", "timestamp": datetime.now().isoformat()}4.4 启动与测试
创建start.sh脚本一键启动:
#!/bin/bash poetry run uvicorn main:app --host 0.0.0.0 --port 8000 --reload赋予执行权限并运行:
chmod +x start.sh ./start.sh用 curl 测试:
curl -X POST "http://localhost:8000/v1/chat/completions" \ -H "Content-Type: application/json" \ -d '{ "messages": [{"role": "user", "content": "用三句话解释量子纠缠"}] }'你会看到返回的 JSON 中,model字段显示实际调用的模型(比如gpt-4-turbo),usage字段有精确的 token 数,_hidden_params里有cost字段。同时终端日志会打印出类似:
2024-06-15 14:22:33,456 - llm-router - INFO - SUCCESS - model=gpt-4-turbo duration=2.34s prompt_tokens=28 completion_tokens=152 cost=$0.001234实操心得:第一次运行时,如果遇到
AuthenticationError,90% 的概率是密钥没生效。用print(os.getenv("OPENAI_API_KEY"))在代码里 debug,别信.env文件的“存在感”。另外,Groq 的llama3-70b模型在免费 tier 有速率限制,如果返回429,换llama3-8b测试更稳妥。
5. 常见问题与排查技巧实录:那些文档里不会写的“血泪教训”
LiteLLM 的文档写得非常清晰,但真实世界里的坑,往往藏在文档的留白处。以下是我在三个不同规模项目中,反复遇到、反复解决的典型问题,附带可直接复制的排查命令和修复方案。
5.1 问题:模型调用成功,但返回内容为空或格式错乱
现象:response.choices[0].message.content是空字符串,或者返回的是{"error": "invalid_request_error"}这样的原始错误体,而不是 LiteLLM 封装后的标准格式。
排查步骤:
- 首先确认是不是流式(stream)模式的问题——LiteLLM 的
stream=True返回的是异步生成器,必须用async for chunk in response:循环消费,直接print(response)会看到<async_generator object ...>。 - 如果是
stream=False,检查response对象的类型:print(type(response))。正常应是litellm.types.CompletionResponse,如果打印出来是dict,说明你调用的是底层 SDK(比如openai.ChatCompletion.create),没走 LiteLLM 的路由。 - 最常见的原因是
messages格式不合规。LiteLLM 要求messages必须是[{"role": "user", "content": "..."}, ...],而有些模型(如早期的 Llama2)不支持system角色。LiteLLM 默认会把system消息合并到第一个user消息里,但如果合并逻辑出错,就会导致内容丢失。
修复方案:
强制禁用 system 消息合并,用drop_params=True让 LiteLLM 原样透传:
response = litellm.completion( model="llama3-70b-groq", messages=[{"role": "system", "content": "You are a helpful assistant"}, {"role": "user", "content": "Hello"}], drop_params=True # 关键!跳过 LiteLLM 的消息预处理 )注意:
drop_params=True是双刃剑,它绕过了 LiteLLM 的标准化,意味着你要自己保证消息格式符合目标模型的要求。只在调试时用,生产环境建议用litellm.set_verbose=True开启详细日志,看它到底把消息转成了什么。
5.2 问题:Router 路由失效,所有请求都打到同一个模型
现象:配置了routing_strategy="usage-based",但观察日志发现 100% 的请求都去了gpt-4-turbo,claude-3-haiku完全没被调用。
根本原因:usage-based策略依赖litellm内部的model_cost_map,这个 map 默认只包含 OpenAI、Anthropic 等主流模型的价格。如果你用了自定义模型(比如my-local-llama3),LiteLLM 不知道它的价格,就会把它排除在路由池外。
验证方法:
from litellm import model_cost print(model_cost) # 查看当前已知的所有模型价格 # 如果没看到你的模型名,就证实了问题修复方案:
手动注入模型成本:
from litellm import model_cost # 添加自定义模型价格:$0.0005/1k tokens model_cost["my-local-llama3"] = { "input_cost_per_token": 0.0000005, "output_cost_per_token": 0.0000005, "max_tokens": 8192 } # 然后在 Router 配置中使用这个模型名 router = Router( model_list=[{ "model_name": "my-local-llama3", "litellm_params": { "model": "ollama/llama3", "api_base": "http://localhost:11434/v1" } }], routing_strategy="usage-based" )5.3 问题:高并发下出现ConnectionResetError或Timeout
现象:单请求测试一切正常,但用locust压测 50 QPS 时,大量请求报ConnectionResetError: [Errno 104] Connection reset by peer或ReadTimeout。
原因分析:LiteLLM 默认使用httpx.AsyncClient,其连接池大小是有限的。当并发请求数超过连接池上限(默认 10),后续请求就会排队等待,超时后直接失败。这不是 LiteLLM 的 bug,而是 HTTP 客户端的通用行为。
解决方案:
显式配置httpx客户端,增大连接池:
import httpx from litellm import Router # 创建自定义 httpx client client = httpx.AsyncClient( limits=httpx.Limits( max_connections=100, # 总连接数 max_keepalive_connections=20, # 保持长连接数 keepalive_expiry=60.0 # 长连接存活时间 ), timeout=httpx.Timeout(60.0, connect=10.0, read=50.0) ) router = Router( model_list=[...], client=client # 关键!把自定义 client 传给 Router )实测数据:在 32G 内存的服务器上,
max_connections=100可稳定支撑 80 QPS,平均延迟 < 1.5s。超过这个值,建议横向扩展服务实例,而不是继续堆连接数。
5.4 问题排查速查表
| 问题现象 | 最可能原因 | 快速验证命令 | 修复方案 |
|---|---|---|---|
AuthenticationError | 环境变量未加载或密钥错误 | echo $OPENAI_API_KEY | 检查.env文件路径,用poetry run python -c "import os; print(os.getenv('KEY'))"验证 |
返回{"error": "model_not_found"} | 模型名拼写错误或未在model_list中注册 | print(router.model_list) | 检查model_name和litellm_params.model是否一致,注意大小写 |
ImportError: No module named 'anthropic' | 缺少对应厂商 SDK | pip list | grep anthropic | pip install anthropic,LiteLLM 不自动安装这些依赖 |
日志中cost=0.0 | 模型未在model_cost_map中 | from litellm import model_cost; print('gpt-4-turbo' in model_cost) | 手动model_cost["model_name"] = {...}注入价格 |
RuntimeWarning: coroutine 'acompletion' was never awaited | 混淆了同步/异步调用 | 检查函数是否加了async,调用是否用了await | 同步用litellm.completion(),异步用await litellm.acompletion() |
6. 进阶技巧与生产就绪建议:让 LiteLLM 从“能跑”到“敢上生产”
LiteLLM 的基础功能开箱即用,但要让它真正扛住生产环境的千锤百炼,还需要几个关键加固点。这些不是文档里的“可选项”,而是我在线上系统崩溃三次后,亲手写进 SOP 的“必选项”。
6.1 密钥轮转与动态加载:告别重启服务
生产环境的密钥不可能永久有效,OpenAI 的密钥会过期,Anthropic 的密钥可能被误删。LiteLLM 支持运行时热更新密钥,无需重启服务:
from litellm import update_model_list # 新密钥来了,动态更新 new_model_list = [ { "model_name": "gpt-4-turbo", "litellm_params": { "model": "gpt-4-turbo", "api_key": "sk-new-key-xxxx", # 新密钥 "api_base": "https://api.openai.com/v1" } } ] # 热更新,Router 会自动 reload update_model_list(new_model_list)更进一步,可以结合 HashiCorp Vault,写一个定时任务每小时拉取最新密钥:
import asyncio from litellm import update_model_list async def refresh_keys_from_vault(): while True: try: # 从 Vault 获取密钥(伪代码) new_keys = await vault_client.read_secret("llm/api-keys") new_model_list = [{ "model_name": "gpt-4-turbo", "litellm_params": { "model": "gpt-4-turbo", "api_key": new_keys["openai_key"] } }] update_model_list(new_model_list) logger.info("Keys refreshed from Vault") except Exception as e: logger.error(f"Failed to refresh keys: {e}") await asyncio.sleep(3600) # 每小时一次 # 在 FastAPI startup 事件中启动 @app.on_event("startup") async def startup_event(): asyncio.create_task(refresh_keys_from_vault())6.2 模型健康度主动探测:别等用户投诉才报警
LiteLLM 的熔断是被动的(等失败才触发),我们可以加一层主动探测,每 30 秒用一个“探针请求”测试每个模型:
import asyncio from litellm import completion async def probe_model(model_name: str): try: response = await completion( model=model_name, messages=[{"role": "user", "content": "hi"}], timeout=10 ) return response.usage.total_tokens > 0 except Exception as e: logger.warning(f"Probe failed for {model_name}: {e}") return False async def health_probe_loop(): while True: for model in ["gpt-4-turbo", "claude-3-haiku", "llama3-70b-groq"]: is_healthy = await probe_model(model) if not is_healthy: # 发送告警,比如企业微信机器人 send_alert(f"Model {model} unhealthy!") await asyncio.sleep(30) @app.on_event("startup") async def startup_event(): asyncio.create_task(health_probe_loop())6.3 成本预警与自动降级:把预算控制写进代码
最后,也是最重要的——把财务风控做成代码逻辑。我们可以在每次请求前,检查本月已用成本是否接近预算:
import os from litellm import get_total_cost # 从环境变量读取月度预算(单位:美元) MONTHLY_BUDGET = float(os.getenv("MONTHLY_BUDGET", "1000")) def should_allow_request() -> bool: total_cost = get_total_cost() # LiteLLM 内置的累计成本统计 if total_cost > MONTHLY_BUDGET * 0.9: # 超过 90%,触发降级 logger.warning(f"Budget alert: ${total_cost:.2f}/${MONTHLY_BUDGET} used") return False return True @app.post("/v1/chat/completions") async def chat_completions(request: ChatRequest): if not should_allow_request(): # 自动降级到最便宜的模型 request.model = "claude-3-haiku" # ... rest of the logic我个人在实际操作中的体会是:LiteLLM 最大的价值,不是它帮你省了多少 token,而是它把原本分散在各个团队、各个服务、各个脚本里的“模型调用”这件事,收束成一个可监控、可审计、可策略化的中心节点。当你能在 Grafana 里看到一张“各模型调用量 vs 成本”折线图,在 Slack 里收到“Claude 3 Sonnet 响应延迟突增 200ms”的告警,在财务报表里精确列出“Qwen2-72b 占比 37%”时,你就知道,这场大模型应用的基建之战,你已经赢了一半。剩下的,就是不断往这个“中枢”里注入更精细的策略、更实时的数据、更人性化的反馈——而 LiteLLM,始终是你最可靠的那块基石。