前些天把一个跑了半年的内部工具从“只接一家模型服务商”改成“同时接四五家”，改到一半差点想骂人。各家API长得实在太不像了，有的用messages数组，有的用prompt字符串，有的返回choices[0].message.content，有的把内容藏在content[0].text里。改到第三天我实在忍不了，开始认真研究litellm这个开源方案，最后干脆把整个项目里的所有模型调用全部切了过去。这篇文章就从我自己的角度聊聊litellm——它到底解决什么问题、核心机制长什么样、怎么在一个真实项目里落地，以及我实测下来觉得必须注意的细节。如果你正在做多模型接入、模型切换、成本统计这类事，这篇文章应该能帮你少走几步弯路。1. 模型接口分裂带来的运维噩梦1.1 不是“多个模型”，是“多套协议”先说痛点。大多数团队做AI功能，起步时都只接一家模型服务商，代码里调用逻辑高度耦合。到了后面要接第二家、第三家做效果对比或者降本时，问题就来了：每家服务商的HTTP请求格式、鉴权方式、超时约定、错误返回结构，几乎没有一致的。我给一个简化但很真实的例子。假设现有代码里封装了一个call_model()函数，里面用的是某厂商的请求体：resp = requests.post( "https://api.example-a.com/v1/chat/completions", headers={"Authorization": f"Bearer {api_key}"}, json={ "model": "demo-chat", "messages": [{"role": "user", "content": "你好"}] } ) data = resp.json() answer = data["choices"][0]["message"]["content"]换到另一家服务商，请求路径变了，messages这个字段名变成了prompt，返回内容也从choices嵌套变成了顶层字段。最让人崩溃的是错误码体系，第一家超时返回408，第二家用504表示限流，第三家干脆所有错误都返回400，靠error.type字段区分。这种差异放在三五次调用的Demo里不是事，放到生产环境就是纯消耗。1.2 为什么“自己写个适配层”看起来香，做起来臭很多团队的第一反应是写一个自己的适配层，给上层提供统一接口，底层分别对接各家API。这个方案不是不行，我早期也是这么干的。但做深了就会发现问题：第一，每接一家新服务商，你就要维护一套协议转换代码。算上流式、非流式、工具调用、视觉输入、嵌入模型，这些变体乘起来，维护成本几乎是指数增长。第二，各家API版本更新频繁。某个字段废弃、某个路径加了版本号，你的适配层就得跟着改，但凡有一家升级后没在文档里写清楚，你会花很长时间排查“为什么同样参数，之前好的现在不行了”。第三，错误处理、重试策略、超时控制这些横切逻辑，散落在各个适配代码里，不同服务商的逻辑还互相冲突。比如某家服务商对短连接特别敏感，过快的并发会触发限流，而另一家根本不在乎并发，这就让你的重试策略永远没法统一。所以后来我在内部团队里总结过一句话：自研适配层不是不行，而是“如果你们的业务核心不是模型接入，就别在这上面押注人力”。litellm这类工具存在的意义，就是把这块重复劳动收敛掉。1.3 litellm是什么：一个收敛所有差异的中间层litellm本质上做的事情很简单：把各家模型服务商的API格式、鉴权、返回结构、错误处理，统一成一套OpenAI风格的接口——因为目前大部分开发者最熟悉的就是这套格式。你只需要按OpenAI的调用方式写一份代码，通过不同配置切换底层服务商。它有两种用法，一种是直接在Python代码里当库调用，一种是起一个代理服务，把HTTP请求转发给后端各家服务商。两种模式我们后面都会讲到，先说核心原理。一句话概括：litellm不是帮你“选模型”，而是帮你“管接入”。2. litellm的核心机制：入参、出参、成本、路由2.1 统一入参：你的代码永远只写一种格式litellm最核心的抽象，是把所有模型服务商的输入参数映射成标准结构。你的应用侧代码只需要写一套：from litellm import completion resp = completion( model="provider-b/demo-chat", messages=[ {"role": "system", "content": "你是一个严谨的助手"}, {"role": "user", "content": "介绍一下跨平台开发的基本思路"} ], temperature=0.7 ) print(resp["choices"][0]["message"]["content"])注意model参数的写法：斜杠前面是服务商标识，斜杠后面是该服务商下的具体模型名。litellm看到provider-b/demo-chat，就知道该走哪套协议、哪个鉴权配置。你在messages里写的内容，它会根据目标服务商自动转换成对方要求的格式。比如某家厂商只接受prompt字符串，它会取数组里的文本拼成一个字符串传过去；某家厂商要求的角色枚举不一样，它也会自动映射。这就是第一个“为什么值得用”的理由：你的业务代码不再关心后端是谁。今天用A厂商，明天换B厂商，只需改配置或者改模型名的前缀，代码一行不动。2.2 统一出参：流式、非流式、工具调用都不再各说各话入参统一只是第一步。返回结构的差异才是生产环境里最折磨人的。我举几个实际见过的差异：服务商A返回长文本时，网关强加了长度截断，你拿到文本发现后半截没了。服务商B的流式返回是data:前缀的SSE格式，但字段名叫delta.text；服务商C的流式返回字段叫choices[0].delta.content。服务商D的“工具调用”实现，是在返回里塞了一段JSON字符串，需要你手动解析。litellm做的事情，是把这些输出全部转成一套标准结构。流式输出时，你的代码统一拿choices[0].delta：from litellm import completion resp = completion( model="provider-b/demo-chat", messages=[{"role": "user", "content": "讲个笑话"}], stream=True ) for chunk in resp: delta = chunk["choices"][0]["delta"].get("content") if delta: print(delta, end="", flush=True)不管底层服务商怎么命名流式字段，你在应用侧拿到的永远是同一个schema。这点在前后端联调时价值尤其大：前端只要对接一套SSE格式，后端换模型时前端完全感知不到。2.3 内置的容量管理：重试、超时、限流、熔断说句实在话，很多人一开始用litellm只是图它接口统一，但真正让我觉得“回不去”的，是它内置的攻击型故障处理能力。在没有中间层的情况下，重试逻辑是散落在业务代码里的。接地气的写法一般是try-except包一层，遇到异常time.sleep几秒再试一次，试三次就放弃。这种写法最大的问题在于：你没有区分“服务商暂时过载”和“你的参数有问题”，明明是被限流，你还在那傻乎乎地重试，结果越试越被限流。litellm的completion()支持一组运行时参数：from litellm import completion resp = completion( model="provider-b/demo-chat", messages=[{"role": "user", "content": "你好"}], timeout=30, num_retries=3, retry_strategy="exponential_backoff", max_retries_per_request=5 )它会在请求失败时根据状态码区分：限流类错误执行指数退避重试，鉴权类错误第一时间返回而不是傻等，超时类错误按可配置策略处理。内部还会记录每个服务商的健康状态，连续失败次数超过阈值后，可以自动把流量切到备用模型。对生产系统来说，这一层能力比“统一格式”更值钱，因为它直接关系到系统的可用性。2.4 成本与用量统计：不再等月底账单才知道花了多少多模型接入之后，成本可视化的需求几乎是必然出现的。你总不能等财务看到账单才反应过来某个模型的调用量异常高。litellm本身带有成本记录功能，每次调用它会根据模型定价计算大概费用。通过回调可以输出到日志系统：from litellm import completion def track_cost(data): # data里有response_cost、total_tokens等信息 print(f"本次调用费用: {data['response_cost']:.4f} 元") completion( model="provider-b/demo-chat", messages=[{"role": "user", "content": "你好"}], callback_handler=track_cost )这在多模型对比评测时特别有用——你可以同一批问题跑三家模型，最后直接拉一张费用对比表，方便做技术选型。3. 实战：从零把一个跨厂商调用项目切到litellm3.1 安装与第一轮热身安装很简单，纯Python环境：pip install litellm我建议装的时候顺便装一下litellm[proxy]扩展，因为后面起代理模式会用到：pip install "litellm[proxy]"环境变量方面，不同服务商的API Key有各自的约定命名。举个例子：export PROVIDER_A_API_KEY="sk-example-key-a" export PROVIDER_B_API_KEY="sk-example-key-b"litellm默认会从环境变量里读取对应前缀的key。如果你不想用环境变量，也可以在代码里通过api_key参数显式传入：resp = completion( model="provider-b/demo-chat", messages=[{"role": "user", "content": "hello"}], api_key="sk-example-key-b" )3.2 用一套代码同时调用不同服务商的模型为了验证统一性，我在测试项目里写了一个循环，分别调用两家服务商的模型，完全复用同一段业务逻辑：from litellm import completion models_to_test = [ "provider-a/demo-chat", "provider-b/demo-chat", ] def ask_question(question, model_name): resp = completion( model=model_name, messages=[{"role": "user", "content": question}], temperature=0.5 ) return resp["choices"][0]["message"]["content"] for model_name in models_to_test: answer = ask_question("用一句话解释什么是消息队列", model_name) print(f"[{model_name}] => {answer}")结果就是，除了模型名不同，其余代码完全一致。这一小段代码让我彻底体会到了“接入层收敛”的甜头——以前这个循环得写两份完全不同的请求逻辑。3.3 异步场景下的正确姿势真实系统里，模型调用很少是同步裸调用的。对话机器人、批量处理任务、Agent工具链，几乎都是并发场景。litellm的异步接口是acompletion，写法跟同步版本几乎一致：import asyncio from litellm import acompletion async def generate_all(): tasks = [ acompletion( model="provider-a/demo-chat", messages=[{"role": "user", "content": f"问题{i}"}] ) for i in range(5) ] results = await asyncio.gather(*tasks, return_exceptions=True) for idx, result in enumerate(results): if isinstance(result, Exception): print(f"第{idx}个任务失败: {result}") else: print(result["choices"][0]["message"]["content"]) asyncio.run(generate_all())几点经验：不要为每个请求单独创建Session。用异步模式时维护好连接池，频繁创建销毁Session会白白损耗性能。并发数要和服务商限额对齐。比如某家服务商按每分钟请求数限流，你的并发从10提到20，触发限流的概率会指数上升，异步调用本身不会帮你避开这个坑。务必设置timeout。我在线上见过不少“请求挂起半小时不返回”的情况，异步任务没有超时控制，一个慢请求能拖住整个批次。3.4 流式输出接前端做聊天类产品基本逃不开流式。前端需要打字机效果，后端就只能SSE一个一个吐字。litellm的流式接口往上对接FastAPI这类Web框架，比我预想中顺滑：from fastapi import FastAPI from fastapi.responses import StreamingResponse from litellm import completion app = FastAPI() def generate_stream(question: str): resp = completion( model="provider-b/demo-chat", messages=[{"role": "user", "content": question}], stream=True ) for chunk in resp: delta = chunk["choices"][0]["delta"].get("content") if delta: yield f"data: {delta}\n\n" @app.get("/chat") def chat(question: str): return StreamingResponse(generate_stream(question), media_type="text/event-stream")注意yield出来的内容是字符串，前端EventSource解析时按SSE协议处理即可。litellm把底层各家流式格式统一之后，这里的前后端协议就完全稳定了：你再也不用关心底下那家服务商用的是data:还是event:还是裸文本。提示：流式场景下，如果你想让前端的“打字机”效果包含标点自然停顿，可以自己在yield循环里做缓冲切分，不要一个字一个字吐，否则网络开销会明显放大。4. 从“库”到“服务”：Proxy模式的价值4.1 当你有多个团队、多个项目时，库模式就不够用了litellm的库模式适合单个服务里做接入收敛。但当你所在的技术中心同时有五个业务系统都要接模型时，问题就变了：每个系统各配各的API Key，成本统计时根本分不清是哪条业务线花的。每家业务团队都自己写一份接入代码，即便都是基于litellm，配置风格也五花八门。想统一做限流、审计、密钥轮换时，没有集中控制点。这时候litellm的Proxy模式就派上用场了。它相当于在你公司内网里起一个模型网关，所有业务系统不再直接访问各家服务商，而是访问这个网关，由网关统一转发、统计、限流、鉴权。4.2 部署一个网关：配置比想象中更简单Proxy模式本质上是一个基于FastAPI的应用。安装完成后，一条命令就能启动：litellm --config ./config.yaml --port 4000config.yaml是核心配置文件。下面是一个最小可用的配置示例：model_list: - model_name: unified-chat litellm_params: model: provider-b/demo-chat api_key: os.environ/PROVIDER_B_API_KEY - model_name: unified-chat-fast litellm_params: model: provider-a/demo-chat api_key: os.environ/PROVIDER_A_API_KEY这个配置暴露了两个模型名（unified-chat和unified-chat-fast）。业务团队只需要知道这两个名字，而背后的实际供应商、API Key、模型版本全部都封装在网关侧。启动后，网关提供一个OpenAI兼容的HTTP接口。业务系统可以像调用OpenAI一样调用：curl http://localhost:4000/v1/chat/completions \ -H "Content-Type: application/json" \ -H "Authorization: Bearer sk-litellm-proxy-key" \ -d '{ "model": "unified-chat", "messages": [{"role": "user", "content": "你好"}] }'注意，这个sk-litellm-proxy-key是网关自己颁发的虚拟Key，各家服务商真正的API Key不会暴露给业务系统。4.3 路由策略：同一模型名背后可以做多少文章Proxy模式最好用的点，是可以在网关层做模型路由。比如你可以配置一个unified-chat，让它背后串联三个服务商：正常情况下走A厂商，A厂商触发限流后切到B厂商，B厂商也失败再切C厂商。对于业务方来说，他们完全感知不到这三个切换过程。也可以按预算路由：同一个模型名，根据你的配置把80%的流量导向便宜的服务商，20%导向效果更好的服务商，先小流量验证效果。这些策略在配置里写起来也不复杂：router_settings: routing_strategy: "least-busy" model_group_alias: unified-chat: - provider-b/demo-chat - provider-a/demo-chat提示：路由策略虽然好，但别忽略一个细节——不同服务商的模型能力可能有差异，同一个问题问两家得到的结果可能不一样。如果你的业务对输出一致性要求高，路由切换前最好先跑一轮评测确认可接受。4.4 网关模式下的密钥管理变化很多团队上网关最直接的收益，其实是密钥管理。以前业务系统的后端配置里到处都是各家服务商的API Key，一旦某把Key需要轮换，要跑去每套系统里改，而且可能改漏。上了网关之后，密钥只在网关这一侧管理，业务侧统一用虚拟Key访问。虚拟Key还可以做细粒度控制：给A业务线发一把只允许调用unified-chat的Key，给B业务线发一把允许调用全部模型的Key；某把Key突然调用量异常时，直接吊销即可，不需要动全局配置。5. 真机实测：那些文档没写透的坑5.1 超时设置的“三层”逻辑我踩过的第一个坑，是超时设置不是一层而是三层。litellm的请求超时机制，最外层是timeout参数，控制的是从发出请求到拿到最终响应的完整时间；中间层是服务商SDK自身的超时；再往里一层是底层HTTP客户端的连接超时。如果你只设置了timeout=30，但某个服务商在连接阶段就hang住了，30秒内请求永远不会发出去，你这个超时设定未必能等到那一刻才触发。所以实际配置时，我建议：resp = completion( model="provider-b/demo-chat", messages=[{"role": "user", "content": "你好"}], timeout=30, connect_timeout=5, read_timeout=60, write_timeout=60 )connect_timeout短一些，尽早暴露网络不可达的问题；read_timeout长一些，因为大模型生成长文本确实需要时间；外层timeout作为兜底。5.2 重试机制的“重试陷阱”litellm默认对某些错误码做重试，这本来是好功能，但有一个场景会变成坏事：你请求的是生成类任务，重发一次等于给用户多扣一次钱。比如两家服务商都出现过这种情况：请求超时了，但后端实际上已经生成了内容并计费。重试之后，用户拿到了答案，账单上却出现了两次计费。处理思路有两个：对非幂等场景（比如对话生成），关闭自动重试或把重试次数降到1，改用人工可观察的方式处理。对批量任务场景，记录每次请求的唯一ID，在重试前先查询该请求是否已经成功生成。litellm的请求ID是可以通过litellm_log回调拿到并落库的。重试前先查库，能有效避免重复计费。5.3 连接池与高并发参数在跑压力测试时，我发现默认参数下的并发能力不太够。因为litellm底层依赖httpx，你的连接池配置直接决定了它能支撑多少并发。一个常见的配置方式：import litellm litellm.telemetry = False litellm.suppress_debug_info = True litellm.set_verbose = False # 为所有请求设置共享的连接参数 import httpx client = httpx.Client( timeout=httpx.Timeout(60.0, connect=5.0), limits=httpx.Limits(max_connections=100, max_keepalive_connections=20) )注意：不要把max_connections调到天际。后端服务商大多有并发限制，连接池开得再大，触发限流后反而会出现一堆等待重试的请求堆积在内存里。5.4 自定义模型的接入边界litellm对主流服务商的支持很完善，但你可能会遇到一些“非主流”的模型：内部部署的开源模型、某个小众服务商的推理接口、或者你自己封装的一个特殊服务。这类模型走litellm的自定义provider机制也可以接，但边界要提前搞清楚。最基本的接入方式是用CustomLLM：from litellm import CustomLLM class MyCustomModel(CustomLLM): async def acompletion(self, model, messages, **kwargs): # 这里写你自己服务的调用逻辑 result = await call_my_internal_service(messages) # 返回litellm标准结构 return { "choices": [ {"message": {"role": "assistant", "content": result}} ] } custom_model = MyCustomModel() litellm.custom_provider_map = [ {"provider": "my-company", "custom_handler": custom_model} ]这样就能通过model="my-company/my-model-name"来调用了。但注意，自定义provider的很多内置能力会打折扣：比如成本统计需要你自己提供模型定价信息，流式支持需要你自己实现SSE逻辑，重试策略大概率也需要自己处理。所以如果只是内部实验，自定义provider完全够用；如果是核心链路，建议先用litellm原生支持的成熟服务商，把自定义provider当作补充手段。5.5 关于Debug日志：开与关都很重要litellm本身带了比较完整的debug信息，但对生产环境来说这些信息既是宝也是噪音。开发阶段，我建议开着：import litellm litellm.set_verbose = True这样每次请求都会打印出完整的请求体、响应体、耗时、费用预估值，排查问题非常直观。上了生产环境以后，建议显式关掉：litellm.set_verbose = False litellm.telemetry = False litellm.suppress_debug_info = True否则大量请求日志会刷屏，而且在日志系统里占用的空间成本不低。6. 一些个人体会用litellm把项目里的模型调用全部梳理完之后，我最大的感受是：中间层这种思路，价值不在于你少写了几行代码，而在于你多了一个“控制点”。有控制点，才可以统一设置超时、统一管理密钥、统一统计成本、统一做模型切换。这些东西如果分散在各个业务代码里，等团队规模和调用量上来之后，几乎没法收拾。如果你目前只接了一家模型服务商，代码量也不大，我不建议立刻上litellm——简单就是最大的优势。但如果你已经开始接第二家、第三家，或者你已经在为不同模型间的切换、成本分摊、限流处理而头疼，那么花一两天时间把litellm引入进来，很可能是这个阶段投入产出比最高的一次重构。最后分享一个特别小的技巧：我习惯在litellm的模型配置里给每个模型加一个说明字段，比如用途、负责团队、成本归属部门。等模型数量多到十几个的时候，这个字段会帮你省很多沟通成本。这类元数据在config里只是几行，价值却很实在。
前些天把一个跑了半年的内部工具从“只接一家模型服务商”改成“同时接四五家”，改到一半差点想骂人。各家API长得实在太不像了，有的用messages数组，有的用prompt字符串，有的返回choices[0].message.content，有的把内容藏在content[0].text里。改到第三天我实在忍不了，开始认真研究litellm这个开源方案，最后干脆把整个项目里的所有模型调用全部切了过去。这篇文章就从我自己的角度聊聊litellm——它到底解决什么问题、核心机制长什么样、怎么在一个真实项目里落地，以及我实测下来觉得必须注意的细节。如果你正在做多模型接入、模型切换、成本统计这类事，这篇文章应该能帮你少走几步弯路。1. 模型接口分裂带来的运维噩梦1.1 不是“多个模型”，是“多套协议”先说痛点。大多数团队做AI功能，起步时都只接一家模型服务商，代码里调用逻辑高度耦合。到了后面要接第二家、第三家做效果对比或者降本时，问题就来了：每家服务商的HTTP请求格式、鉴权方式、超时约定、错误返回结构，几乎没有一致的。我给一个简化但很真实的例子。假设现有代码里封装了一个call_model()函数，里面用的是某厂商的请求体：resp = requests.post( "https://api.example-a.com/v1/chat/completions", headers={"Authorization": f"Bearer {api_key}"}, json={ "model": "demo-chat", "messages": [{"role": "user", "content": "你好"}] } ) data = resp.json() answer = data["choices"][0]["message"]["content"]换到另一家服务商，请求路径变了，messages这个字段名变成了prompt，返回内容也从choices嵌套变成了顶层字段。最让人崩溃的是错误码体系，第一家超时返回408，第二家用504表示限流，第三家干脆所有错误都返回400，靠error.type字段区分。这种差异放在三五次调用的Demo里不是事，放到生产环境就是纯消耗。1.2 为什么“自己写个适配层”看起来香，做起来臭很多团队的第一反应是写一个自己的适配层，给上层提供统一接口，底层分别对接各家API。这个方案不是不行，我早期也是这么干的。但做深了就会发现问题：第一，每接一家新服务商，你就要维护一套协议转换代码。算上流式、非流式、工具调用、视觉输入、嵌入模型，这些变体乘起来，维护成本几乎是指数增长。第二，各家API版本更新频繁。某个字段废弃、某个路径加了版本号，你的适配层就得跟着改，但凡有一家升级后没在文档里写清楚，你会花很长时间排查“为什么同样参数，之前好的现在不行了”。第三，错误处理、重试策略、超时控制这些横切逻辑，散落在各个适配代码里，不同服务商的逻辑还互相冲突。比如某家服务商对短连接特别敏感，过快的并发会触发限流，而另一家根本不在乎并发，这就让你的重试策略永远没法统一。所以后来我在内部团队里总结过一句话：自研适配层不是不行，而是“如果你们的业务核心不是模型接入，就别在这上面押注人力”。litellm这类工具存在的意义，就是把这块重复劳动收敛掉。1.3 litellm是什么：一个收敛所有差异的中间层litellm本质上做的事情很简单：把各家模型服务商的API格式、鉴权、返回结构、错误处理，统一成一套OpenAI风格的接口——因为目前大部分开发者最熟悉的就是这套格式。你只需要按OpenAI的调用方式写一份代码，通过不同配置切换底层服务商。它有两种用法，一种是直接在Python代码里当库调用，一种是起一个代理服务，把HTTP请求转发给后端各家服务商。两种模式我们后面都会讲到，先说核心原理。一句话概括：litellm不是帮你“选模型”，而是帮你“管接入”。2. litellm的核心机制：入参、出参、成本、路由2.1 统一入参：你的代码永远只写一种格式litellm最核心的抽象，是把所有模型服务商的输入参数映射成标准结构。你的应用侧代码只需要写一套：from litellm import completion resp = completion( model="provider-b/demo-chat", messages=[ {"role": "system", "content": "你是一个严谨的助手"}, {"role": "user", "content": "介绍一下跨平台开发的基本思路"} ], temperature=0.7 ) print(resp["choices"][0]["message"]["content"])注意model参数的写法：斜杠前面是服务商标识，斜杠后面是该服务商下的具体模型名。litellm看到provider-b/demo-chat，就知道该走哪套协议、哪个鉴权配置。你在messages里写的内容，它会根据目标服务商自动转换成对方要求的格式。比如某家厂商只接受prompt字符串，它会取数组里的文本拼成一个字符串传过去；某家厂商要求的角色枚举不一样，它也会自动映射。这就是第一个“为什么值得用”的理由：你的业务代码不再关心后端是谁。今天用A厂商，明天换B厂商，只需改配置或者改模型名的前缀，代码一行不动。2.2 统一出参：流式、非流式、工具调用都不再各说各话入参统一只是第一步。返回结构的差异才是生产环境里最折磨人的。我举几个实际见过的差异：服务商A返回长文本时，网关强加了长度截断，你拿到文本发现后半截没了。服务商B的流式返回是data:前缀的SSE格式，但字段名叫delta.text；服务商C的流式返回字段叫choices[0].delta.content。服务商D的“工具调用”实现，是在返回里塞了一段JSON字符串，需要你手动解析。litellm做的事情，是把这些输出全部转成一套标准结构。流式输出时，你的代码统一拿choices[0].delta：from litellm import completion resp = completion( model="provider-b/demo-chat", messages=[{"role": "user", "content": "讲个笑话"}], stream=True ) for chunk in resp: delta = chunk["choices"][0]["delta"].get("content") if delta: print(delta, end="", flush=True)不管底层服务商怎么命名流式字段，你在应用侧拿到的永远是同一个schema。这点在前后端联调时价值尤其大：前端只要对接一套SSE格式，后端换模型时前端完全感知不到。2.3 内置的容量管理：重试、超时、限流、熔断说句实在话，很多人一开始用litellm只是图它接口统一，但真正让我觉得“回不去”的，是它内置的攻击型故障处理能力。在没有中间层的情况下，重试逻辑是散落在业务代码里的。接地气的写法一般是try-except包一层，遇到异常time.sleep几秒再试一次，试三次就放弃。这种写法最大的问题在于：你没有区分“服务商暂时过载”和“你的参数有问题”，明明是被限流，你还在那傻乎乎地重试，结果越试越被限流。litellm的completion()支持一组运行时参数：from litellm import completion resp = completion( model="provider-b/demo-chat", messages=[{"role": "user", "content": "你好"}], timeout=30, num_retries=3, retry_strategy="exponential_backoff", max_retries_per_request=5 )它会在请求失败时根据状态码区分：限流类错误执行指数退避重试，鉴权类错误第一时间返回而不是傻等，超时类错误按可配置策略处理。内部还会记录每个服务商的健康状态，连续失败次数超过阈值后，可以自动把流量切到备用模型。对生产系统来说，这一层能力比“统一格式”更值钱，因为它直接关系到系统的可用性。2.4 成本与用量统计：不再等月底账单才知道花了多少多模型接入之后，成本可视化的需求几乎是必然出现的。你总不能等财务看到账单才反应过来某个模型的调用量异常高。litellm本身带有成本记录功能，每次调用它会根据模型定价计算大概费用。通过回调可以输出到日志系统：from litellm import completion def track_cost(data): # data里有response_cost、total_tokens等信息 print(f"本次调用费用: {data['response_cost']:.4f} 元") completion( model="provider-b/demo-chat", messages=[{"role": "user", "content": "你好"}], callback_handler=track_cost )这在多模型对比评测时特别有用——你可以同一批问题跑三家模型，最后直接拉一张费用对比表，方便做技术选型。3. 实战：从零把一个跨厂商调用项目切到litellm3.1 安装与第一轮热身安装很简单，纯Python环境：pip install litellm我建议装的时候顺便装一下litellm[proxy]扩展，因为后面起代理模式会用到：pip install "litellm[proxy]"环境变量方面，不同服务商的API Key有各自的约定命名。举个例子：export PROVIDER_A_API_KEY="sk-example-key-a" export PROVIDER_B_API_KEY="sk-example-key-b"litellm默认会从环境变量里读取对应前缀的key。如果你不想用环境变量，也可以在代码里通过api_key参数显式传入：resp = completion( model="provider-b/demo-chat", messages=[{"role": "user", "content": "hello"}], api_key="sk-example-key-b" )3.2 用一套代码同时调用不同服务商的模型为了验证统一性，我在测试项目里写了一个循环，分别调用两家服务商的模型，完全复用同一段业务逻辑：from litellm import completion models_to_test = [ "provider-a/demo-chat", "provider-b/demo-chat", ] def ask_question(question, model_name): resp = completion( model=model_name, messages=[{"role": "user", "content": question}], temperature=0.5 ) return resp["choices"][0]["message"]["content"] for model_name in models_to_test: answer = ask_question("用一句话解释什么是消息队列", model_name) print(f"[{model_name}] => {answer}")结果就是，除了模型名不同，其余代码完全一致。这一小段代码让我彻底体会到了“接入层收敛”的甜头——以前这个循环得写两份完全不同的请求逻辑。3.3 异步场景下的正确姿势真实系统里，模型调用很少是同步裸调用的。对话机器人、批量处理任务、Agent工具链，几乎都是并发场景。litellm的异步接口是acompletion，写法跟同步版本几乎一致：import asyncio from litellm import acompletion async def generate_all(): tasks = [ acompletion( model="provider-a/demo-chat", messages=[{"role": "user", "content": f"问题{i}"}] ) for i in range(5) ] results = await asyncio.gather(*tasks, return_exceptions=True) for idx, result in enumerate(results): if isinstance(result, Exception): print(f"第{idx}个任务失败: {result}") else: print(result["choices"][0]["message"]["content"]) asyncio.run(generate_all())几点经验：不要为每个请求单独创建Session。用异步模式时维护好连接池，频繁创建销毁Session会白白损耗性能。并发数要和服务商限额对齐。比如某家服务商按每分钟请求数限流，你的并发从10提到20，触发限流的概率会指数上升，异步调用本身不会帮你避开这个坑。务必设置timeout。我在线上见过不少“请求挂起半小时不返回”的情况，异步任务没有超时控制，一个慢请求能拖住整个批次。3.4 流式输出接前端做聊天类产品基本逃不开流式。前端需要打字机效果，后端就只能SSE一个一个吐字。litellm的流式接口往上对接FastAPI这类Web框架，比我预想中顺滑：from fastapi import FastAPI from fastapi.responses import StreamingResponse from litellm import completion app = FastAPI() def generate_stream(question: str): resp = completion( model="provider-b/demo-chat", messages=[{"role": "user", "content": question}], stream=True ) for chunk in resp: delta = chunk["choices"][0]["delta"].get("content") if delta: yield f"data: {delta}\n\n" @app.get("/chat") def chat(question: str): return StreamingResponse(generate_stream(question), media_type="text/event-stream")注意yield出来的内容是字符串，前端EventSource解析时按SSE协议处理即可。litellm把底层各家流式格式统一之后，这里的前后端协议就完全稳定了：你再也不用关心底下那家服务商用的是data:还是event:还是裸文本。提示：流式场景下，如果你想让前端的“打字机”效果包含标点自然停顿，可以自己在yield循环里做缓冲切分，不要一个字一个字吐，否则网络开销会明显放大。4. 从“库”到“服务”：Proxy模式的价值4.1 当你有多个团队、多个项目时，库模式就不够用了litellm的库模式适合单个服务里做接入收敛。但当你所在的技术中心同时有五个业务系统都要接模型时，问题就变了：每个系统各配各的API Key，成本统计时根本分不清是哪条业务线花的。每家业务团队都自己写一份接入代码，即便都是基于litellm，配置风格也五花八门。想统一做限流、审计、密钥轮换时，没有集中控制点。这时候litellm的Proxy模式就派上用场了。它相当于在你公司内网里起一个模型网关，所有业务系统不再直接访问各家服务商，而是访问这个网关，由网关统一转发、统计、限流、鉴权。4.2 部署一个网关：配置比想象中更简单Proxy模式本质上是一个基于FastAPI的应用。安装完成后，一条命令就能启动：litellm --config ./config.yaml --port 4000config.yaml是核心配置文件。下面是一个最小可用的配置示例：model_list: - model_name: unified-chat litellm_params: model: provider-b/demo-chat api_key: os.environ/PROVIDER_B_API_KEY - model_name: unified-chat-fast litellm_params: model: provider-a/demo-chat api_key: os.environ/PROVIDER_A_API_KEY这个配置暴露了两个模型名（unified-chat和unified-chat-fast）。业务团队只需要知道这两个名字，而背后的实际供应商、API Key、模型版本全部都封装在网关侧。启动后，网关提供一个OpenAI兼容的HTTP接口。业务系统可以像调用OpenAI一样调用：curl http://localhost:4000/v1/chat/completions \ -H "Content-Type: application/json" \ -H "Authorization: Bearer sk-litellm-proxy-key" \ -d '{ "model": "unified-chat", "messages": [{"role": "user", "content": "你好"}] }'注意，这个sk-litellm-proxy-key是网关自己颁发的虚拟Key，各家服务商真正的API Key不会暴露给业务系统。4.3 路由策略：同一模型名背后可以做多少文章Proxy模式最好用的点，是可以在网关层做模型路由。比如你可以配置一个unified-chat，让它背后串联三个服务商：正常情况下走A厂商，A厂商触发限流后切到B厂商，B厂商也失败再切C厂商。对于业务方来说，他们完全感知不到这三个切换过程。也可以按预算路由：同一个模型名，根据你的配置把80%的流量导向便宜的服务商，20%导向效果更好的服务商，先小流量验证效果。这些策略在配置里写起来也不复杂：router_settings: routing_strategy: "least-busy" model_group_alias: unified-chat: - provider-b/demo-chat - provider-a/demo-chat提示：路由策略虽然好，但别忽略一个细节——不同服务商的模型能力可能有差异，同一个问题问两家得到的结果可能不一样。如果你的业务对输出一致性要求高，路由切换前最好先跑一轮评测确认可接受。4.4 网关模式下的密钥管理变化很多团队上网关最直接的收益，其实是密钥管理。以前业务系统的后端配置里到处都是各家服务商的API Key，一旦某把Key需要轮换，要跑去每套系统里改，而且可能改漏。上了网关之后，密钥只在网关这一侧管理，业务侧统一用虚拟Key访问。虚拟Key还可以做细粒度控制：给A业务线发一把只允许调用unified-chat的Key，给B业务线发一把允许调用全部模型的Key；某把Key突然调用量异常时，直接吊销即可，不需要动全局配置。5. 真机实测：那些文档没写透的坑5.1 超时设置的“三层”逻辑我踩过的第一个坑，是超时设置不是一层而是三层。litellm的请求超时机制，最外层是timeout参数，控制的是从发出请求到拿到最终响应的完整时间；中间层是服务商SDK自身的超时；再往里一层是底层HTTP客户端的连接超时。如果你只设置了timeout=30，但某个服务商在连接阶段就hang住了，30秒内请求永远不会发出去，你这个超时设定未必能等到那一刻才触发。所以实际配置时，我建议：resp = completion( model="provider-b/demo-chat", messages=[{"role": "user", "content": "你好"}], timeout=30, connect_timeout=5, read_timeout=60, write_timeout=60 )connect_timeout短一些，尽早暴露网络不可达的问题；read_timeout长一些，因为大模型生成长文本确实需要时间；外层timeout作为兜底。5.2 重试机制的“重试陷阱”litellm默认对某些错误码做重试，这本来是好功能，但有一个场景会变成坏事：你请求的是生成类任务，重发一次等于给用户多扣一次钱。比如两家服务商都出现过这种情况：请求超时了，但后端实际上已经生成了内容并计费。重试之后，用户拿到了答案，账单上却出现了两次计费。处理思路有两个：对非幂等场景（比如对话生成），关闭自动重试或把重试次数降到1，改用人工可观察的方式处理。对批量任务场景，记录每次请求的唯一ID，在重试前先查询该请求是否已经成功生成。litellm的请求ID是可以通过litellm_log回调拿到并落库的。重试前先查库，能有效避免重复计费。5.3 连接池与高并发参数在跑压力测试时，我发现默认参数下的并发能力不太够。因为litellm底层依赖httpx，你的连接池配置直接决定了它能支撑多少并发。一个常见的配置方式：import litellm litellm.telemetry = False litellm.suppress_debug_info = True litellm.set_verbose = False # 为所有请求设置共享的连接参数 import httpx client = httpx.Client( timeout=httpx.Timeout(60.0, connect=5.0), limits=httpx.Limits(max_connections=100, max_keepalive_connections=20) )注意：不要把max_connections调到天际。后端服务商大多有并发限制，连接池开得再大，触发限流后反而会出现一堆等待重试的请求堆积在内存里。5.4 自定义模型的接入边界litellm对主流服务商的支持很完善，但你可能会遇到一些“非主流”的模型：内部部署的开源模型、某个小众服务商的推理接口、或者你自己封装的一个特殊服务。这类模型走litellm的自定义provider机制也可以接，但边界要提前搞清楚。最基本的接入方式是用CustomLLM：from litellm import CustomLLM class MyCustomModel(CustomLLM): async def acompletion(self, model, messages, **kwargs): # 这里写你自己服务的调用逻辑 result = await call_my_internal_service(messages) # 返回litellm标准结构 return { "choices": [ {"message": {"role": "assistant", "content": result}} ] } custom_model = MyCustomModel() litellm.custom_provider_map = [ {"provider": "my-company", "custom_handler": custom_model} ]这样就能通过model="my-company/my-model-name"来调用了。但注意，自定义provider的很多内置能力会打折扣：比如成本统计需要你自己提供模型定价信息，流式支持需要你自己实现SSE逻辑，重试策略大概率也需要自己处理。所以如果只是内部实验，自定义provider完全够用；如果是核心链路，建议先用litellm原生支持的成熟服务商，把自定义provider当作补充手段。5.5 关于Debug日志：开与关都很重要litellm本身带了比较完整的debug信息，但对生产环境来说这些信息既是宝也是噪音。开发阶段，我建议开着：import litellm litellm.set_verbose = True这样每次请求都会打印出完整的请求体、响应体、耗时、费用预估值，排查问题非常直观。上了生产环境以后，建议显式关掉：litellm.set_verbose = False litellm.telemetry = False litellm.suppress_debug_info = True否则大量请求日志会刷屏，而且在日志系统里占用的空间成本不低。6. 一些个人体会用litellm把项目里的模型调用全部梳理完之后，我最大的感受是：中间层这种思路，价值不在于你少写了几行代码，而在于你多了一个“控制点”。有控制点，才可以统一设置超时、统一管理密钥、统一统计成本、统一做模型切换。这些东西如果分散在各个业务代码里，等团队规模和调用量上来之后，几乎没法收拾。如果你目前只接了一家模型服务商，代码量也不大，我不建议立刻上litellm——简单就是最大的优势。但如果你已经开始接第二家、第三家，或者你已经在为不同模型间的切换、成本分摊、限流处理而头疼，那么花一两天时间把litellm引入进来，很可能是这个阶段投入产出比最高的一次重构。最后分享一个特别小的技巧：我习惯在litellm的模型配置里给每个模型加一个说明字段，比如用途、负责团队、成本归属部门。等模型数量多到十几个的时候，这个字段会帮你省很多沟通成本。这类元数据在config里只是几行，价值却很实在。