litellm fastapi sse api 集成简单说明：用 TaoToken 统一 Key 打通 Ollama 流式输出-拓冰网站定制

1. 为什么要在 FastAPI 里做 SSE 流式接口

如果你正在做 AI 应用，大概率会遇到一个很具体的需求：前端要一个字一个字地往外蹦内容，而不是等模型全部生成完再一次性返回。这就是 SSE（Server-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?prompt=xxx请求，FastAPI 路由收到后，通过 litellm 的 OpenAI 兼容客户端向代理服务发起stream=True的请求；代理服务根据 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/api，api_key填你申请的 Key。这样本地模型和云端模型就在同一个代理下统一管理了。

启动 proxy 的命令很简单：

litellm --config ./config.yaml --port 4000

如果你要用数据库动态模式，需要额外设置环境变量export STORE_MODEL_IN_DB='True'，并在配置里加上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_key：

from 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_key="sk-1234", base_url="http://localhost:4000" ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) def sse_format(message: str): return f"data: {message}\n\n" @app.get("/stream") async def stream_openai(prompt: str): async def generate(): response = client.chat.completions.create( model="gemma2", stream=True, 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( model="gemma2", messages=messages, stream=True ) 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_type="text/event-stream") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

两个路由的区别值得说清楚。/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 协议规定的格式。

再验证一下/streamv2：

curl -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 层。分层验证是排查问题的好习惯。

浏览器端验证也很简单，用EventSource：

const 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而不是https，Ollama 默认是 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_buffering；FastAPI 这边确认返回的是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参数，长时间生成时能避免中间层把空闲连接掐断。这些细节在本地测试时看不出来，一上生产就会暴露，提前处理能省不少事。

资讯详情