ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

litellm fastapi sse api 集成简单说明:用 TaoToken 统一 Key 打通 Ollama 流式输出

litellm fastapi sse api 集成简单说明:用 TaoToken 统一 Key 打通 Ollama 流式输出 1. 为什么要在 FastAPI 里做 SSE 流式接口如果你正在做 AI 应用大概率会遇到一个很具体的需求前端要一个字一个字地往外蹦内容而不是等模型全部生成完再一次性返回。这就是 SSEServer-Sent Events的典型场景。SSE 是一种基于 HTTP 的服务器推送技术浏览器用EventSource就能接收比 WebSocket 轻量得多特别适合大模型这种「单向流式输出」的场景。但真正落地时麻烦往往不在 SSE 本身而在上游模型的接入。你可能本地跑着 Ollama里面有 qwen2、gemma2 这些模型同时业务又需要调用云端模型。如果每个模型都单独写一套调用逻辑、单独管理 Key代码会迅速变成一团乱麻。litellm 就是来解决这个问题的——它用统一的 OpenAI 兼容接口去调用上百种模型包括 Ollama 本地模型。再配合 TaoToken 统一 Key/API 通道你就能用一套凭证管理多模型调用不用在代码里到处塞不同的 base_url 和 api_key。这篇文章要解决的就是这条链路Ollama 提供本地模型 → litellm 做统一代理 → FastAPI 暴露 SSE 流式接口 → curl 验证端到端跑通。适合已经会一点 Python、想快速把流式接口搭起来的人。我会给出可直接复制的配置片段和路由代码也会把常见的报错对照着讲清楚。先说清楚整体数据流这样后面看代码不会迷路。客户端发起GET /stream?promptxxx请求FastAPI 路由收到后通过 litellm 的 OpenAI 兼容客户端向代理服务发起streamTrue的请求代理服务根据 model_name 路由到对应的上游本地 Ollama 或云端模型把生成结果以 chunk 的形式回传FastAPI 再把每个 chunk 包装成 SSE 事件格式data: {...}\n\n推给客户端。整条链路里litellm 负责「统一接口 路由」TaoToken 负责「统一 Key 和通道管理」FastAPI 负责「对外暴露 SSE」。这里有个容易混淆的点litellm 既可以作为 Python 库直接import使用也可以作为独立代理服务proxy运行。做多模型统一管理时推荐用 proxy 模式因为它把模型配置、Key、限流都集中到一个配置文件里业务代码只需要认一个 base_url。下面第二节就先把这个前置条件搭好。2. TaoToken 与 litellm proxy 前置准备在写 FastAPI 代码之前得先把「统一入口」准备好。这一步的核心是两件事拿到 TaoToken 的 API Key以及把 litellm proxy 跑起来并指向正确的上游。先说 TaoToken 这边。它的作用是给你一个统一的 API 通道和 Key让你不用为每个模型单独申请凭证。你需要去控制台创建一个 API Key这个 Key 后面会作为 litellm 的 master_key 或者上游 api_key 使用。创建入口在 API Keys 页面登录后就能生成。拿到形如sk-xxxx的字符串后先存好后面配置文件里要用。关于接入方式和文档建议先扫一眼官方接入文档里面有针对不同框架的说明。地址是 https://taotoken.net/api 文档页在 https://taotoken.net/doc 。如果你后面要接 Claude Code 这类编码工具还有专门的 Coding Plan 可以参考 https://taotoken.net/coding-plan 。接下来是 litellm proxy 的配置。litellm 支持静态 YAML 配置也支持数据库动态模式。对于大多数场景静态配置就够了。下面这份配置同时挂了两个 Ollama 模型并预留了通过 TaoToken 走云端模型的位置model_list: - model_name: qwen2 litellm_params: model: ollama/qwen2:1.5b api_base: http://localhost:11434 api_key: demo rpm: 60 - model_name: gemma2 litellm_params: model: ollama/gemma2:2b api_base: http://localhost:11434 api_key: demo rpm: 60 - model_name: cloud-chat litellm_params: model: openai/gpt-4o-mini api_base: https://taotoken.net/api api_key: sk-你的TaoToken密钥 router_settings: routing_strategy: usage-based-routing-v2 general_settings: master_key: sk-1234这份配置里有几个关键字段要解释。model_name是对外暴露的名字业务代码里model填的就是它litellm_params.model才是真正的上游模型标识ollama/前缀告诉 litellm 走 Ollama 适配器api_base指向 Ollama 默认的 11434 端口。第三个cloud-chat条目演示了如何通过 TaoToken 的 API 地址接入云端模型api_base填https://taotoken.net/apiapi_key填你申请的 Key。这样本地模型和云端模型就在同一个代理下统一管理了。启动 proxy 的命令很简单litellm --config ./config.yaml --port 4000如果你要用数据库动态模式需要额外设置环境变量export STORE_MODEL_IN_DBTrue并在配置里加上database_url。不过对于本文的流式接口演示静态配置完全够用先不引入数据库依赖减少出错点。启动成功后访问http://localhost:4000应该能看到 litellm 的欢迎页。这一步验证通过说明代理层已经就绪可以进入 FastAPI 的编写了。注意 master_key 设成sk-1234后业务代码调用时也要带上这个 Key否则会返回 401。3. 可复制的 FastAPI SSE 路由与配置现在进入核心部分写 FastAPI 路由。这里我用 litellm 的 OpenAI 兼容客户端来发请求因为 litellm proxy 暴露的就是 OpenAI 格式的接口直接用openai库最省事。SSE 的封装有两种常见写法一种是sse_starlette的EventSourceResponse一种是 FastAPI 原生的StreamingResponse两种我都会给出来。先装依赖pip install fastapi uvicorn openai sse-starlette然后是完整的应用代码。注意base_url指向 litellm proxy 的 4000 端口api_key用配置里的 master_keyfrom fastapi import FastAPI from fastapi.responses import StreamingResponse from fastapi.middleware.cors import CORSMiddleware from sse_starlette.sse import EventSourceResponse import openai import asyncio app FastAPI() client openai.OpenAI( api_keysk-1234, base_urlhttp://localhost:4000 ) app.add_middleware( CORSMiddleware, allow_origins[*], allow_credentialsTrue, allow_methods[*], allow_headers[*], ) def sse_format(message: str): return fdata: {message}\n\n app.get(/stream) async def stream_openai(prompt: str): async def generate(): response client.chat.completions.create( modelgemma2, streamTrue, messages[{role: user, content: prompt}] ) for chunk in response: choice chunk.choices[0] yield choice.model_dump_json() await asyncio.sleep(0.05) return EventSourceResponse(generate()) app.get(/streamv2) async def openai_stream(prompt: str): messages [{role: user, content: prompt}] async def stream_response(): response client.chat.completions.create( modelgemma2, messagesmessages, streamTrue ) for chunk in response: choice chunk.choices[0] yield sse_format(choice.model_dump_json()) await asyncio.sleep(0.05) return StreamingResponse(stream_response(), media_typetext/event-stream) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)两个路由的区别值得说清楚。/stream用EventSourceResponse它会自动帮你处理 SSE 的格式和心跳你只需要 yield 数据内容/streamv2用原生StreamingResponse需要自己拼data: ...\n\n的格式但控制更灵活。实际项目里我更推荐EventSourceResponse因为它对连接断开、心跳保活处理得更稳。关于model参数这里填的是gemma2对应 litellm 配置里的model_name。如果你想切到 qwen2只改这一个字符串就行其他代码完全不用动——这就是统一代理的价值。如果要用 TaoToken 的云端模型把model改成cloud-chat即可。还有一个细节choice.model_dump_json()是 Pydantic v2 的方法能把 chunk 对象序列化成 JSON 字符串。如果你用的是旧版 Pydantic改成choice.json()。这个 JSON 里包含delta.content字段前端解析时取这个字段就能拿到增量文本。启动服务uvicorn main:app --host 0.0.0.0 --port 8000到这里配置和代码都齐了。下一节用 curl 实际验证流式返回确认整条链路真的通了。4. 用 curl 验证流式返回与成功结果代码写完不代表链路通了必须实际发请求验证。SSE 的验证用 curl 最直观因为你能看到数据是一段一段吐出来的而不是一次性返回。先确认 litellm proxy 在跑4000 端口FastAPI 也在跑8000 端口。然后执行curl -N http://localhost:8000/stream?prompt用一句话介绍你自己-N参数很关键它关闭 curl 的缓冲让你能实时看到每个 SSE 事件。如果一切正常你会看到类似这样的输出一行一行往外冒data: {id:chatcmpl-xxx,object:chat.completion.chunk,created:1730000000,model:gemma2,choices:[{index:0,delta:{role:assistant,content:我},finish_reason:null}]} data: {id:chatcmpl-xxx,object:chat.completion.chunk,created:1730000000,model:gemma2,choices:[{index:0,delta:{content:是},finish_reason:null}]} data: {id:chatcmpl-xxx,object:chat.completion.chunk,created:1730000000,model:gemma2,choices:[{index:0,delta:{content:一个},finish_reason:null}]} data: {id:chatcmpl-xxx,object:chat.completion.chunk,created:1730000000,model:gemma2,choices:[{index:0,delta:{},finish_reason:stop}]}看到delta.content逐字出现最后有一个finish_reason: stop的结束事件就说明流式链路完全打通了。每个data:行之间用空行分隔这是 SSE 协议规定的格式。再验证一下/streamv2curl -N http://localhost:8000/streamv2?prompt你好输出格式应该和上面一致因为两个路由最终产出的 SSE 事件结构是相同的区别只在内部实现。如果你想直接验证 litellm proxy 这一层可以绕过 FastAPI 直接打 4000 端口curl -N http://localhost:4000/v1/chat/completions \ -H Authorization: Bearer sk-1234 \ -H Content-Type: application/json \ -d { model: gemma2, stream: true, messages: [{role: user, content: 你好}] }这个请求能返回流式数据说明 litellm 和 Ollama 之间的连接没问题如果这个通了但 FastAPI 那个不通问题就在 FastAPI 层。分层验证是排查问题的好习惯。浏览器端验证也很简单用EventSourceconst es new EventSource(http://localhost:8000/stream?prompt你好); es.onmessage (e) { const chunk JSON.parse(e.data); const text chunk.choices[0]?.delta?.content || ; process.stdout.write(text); }; es.onerror () es.close();注意EventSource只支持 GET 请求这也是为什么上面的路由都用app.get。如果你的 prompt 很长建议改成 POST fetch的流式读取方式避免 URL 长度限制。5. 常见报错排查对照流式接口的坑大多集中在连接和格式上下面按真实报错逐个对照。401 Unauthorized。这个最常见通常是 litellm 的 master_key 和业务代码里的 api_key 不一致。检查配置文件里general_settings.master_key的值和 FastAPI 里openai.OpenAI(api_key...)是否完全一致。如果错误来自上游比如 TaoToken 或 Ollama检查对应条目的api_key字段。Ollama 本地模型其实不校验 Key填demo占位即可但字段不能缺。local proxy failed / Connection refused。报错信息里带local proxy或Connection refused说明 litellm 找不到上游。先确认 Ollama 是否在 11434 端口运行用curl http://localhost:11434/api/tags测试。如果 Ollama 没启动litellm 转发时会直接失败。另外确认api_base写的是http://localhost:11434而不是httpsOllama 默认是 http。reading choices / NoneType object has no attribute choices。这个报错说明返回的 chunk 结构和你预期的不一样chunk.choices是 None。常见原因是上游返回了错误信息而不是正常的流式 chunk比如模型名写错、上游超时。建议在循环里加一层判断for chunk in response: if not chunk.choices: continue choice chunk.choices[0] yield choice.model_dump_json()同时把 litellm 的日志级别调高litellm --config ./config.yaml --detailed_debug能看到实际的上游响应。OAuth / authentication_error。如果你接的是需要 OAuth 的云端服务报错里会出现 OAuth 相关字样。用 TaoToken 统一通道时认证走的是 API Key不涉及 OAuth 流程所以这类报错一般出现在直连某些官方 SDK 的场景。确认你用的是api_key而不是 token 刷新机制。SSE 数据一次性返回没有流式效果。curl 里如果所有数据同时出现通常是中间有缓冲。检查是否加了-N如果用 Nginx 反代需要关闭proxy_bufferingFastAPI 这边确认返回的是StreamingResponse或EventSourceResponse而不是普通JSONResponse。模型名找不到 / model not found。litellm 报这个错说明请求的model不在model_list的model_name里。注意model_name和litellm_params.model是两个概念业务代码里填的是前者。改完配置要重启 litellm proxy 才生效。排查时记住分层思路先 curl 4000 端口验证 litellm再 curl 8000 端口验证 FastAPI最后看浏览器。哪一层断了就查哪一层比盲目改代码高效得多。6. 把统一 Key 通道用起来链路跑通之后真正省心的地方在于扩展。你新增一个模型只需要在 litellm 配置的model_list里加一个条目业务代码一行都不用改。比如想加一个通过 TaoToken 接入的云端模型复制cloud-chat那段改个model_name就行。多模型切换从「改代码」变成了「改配置」这是统一通道最实际的价值。对于需要长期跑编码任务或 Agent 的场景可以考虑 TaoToken 的 Coding Plan它在调用额度和通道稳定性上更适合持续性的工作负载具体可以看 https://taotoken.net/coding-plan 。如果你只是想先验证模型对话效果直接用模型对话页面试就行https://taotoken.net/chat 。日常管理 Key 和查看用量在控制台https://taotoken.net/console 创建和轮换 Key 在 https://taotoken.net/api-keys 。最后给一个实用建议把 litellm 的配置文件和 FastAPI 的 base_url、api_key 都放到环境变量里别硬编码在代码中。生产环境里 Key 泄露的代价很高用.env加python-dotenv是最低成本的防护。另外 SSE 连接建议加超时和心跳EventSourceResponse支持ping参数长时间生成时能避免中间层把空闲连接掐断。这些细节在本地测试时看不出来一上生产就会暴露提前处理能省不少事。
返回列表