5.8 KiB
5.8 KiB
id, title, type, domain, stack, tags, level, prerequisites, est_hours, related, week, order
| id | title | type | domain | stack | tags | level | prerequisites | est_hours | related | week | order | |||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| python.fastapi.streaming-testing | 流式输出（SSE）、生命周期、测试和部署 | lesson | python |
|
|
intermediate |
|
3 |
|
|
3 |
流式输出（SSE）、生命周期、测试和部署
解释说明
SSE 流式输出
- 大模型逐字生成，用
StreamingResponse(异步生成器, media_type="text/event-stream")边生成边发送。 - 每个事件的格式：
data: 内容\n\n（以空行结束）；可以加event: 名字区分事件类型（如error）。内容里有换行时用 JSON 编码。 - 响应头一旦发出就不能再修改状态码：生成中途出错，只能在流里发送一个错误事件，前端据此提示。
- 客户端断开：用
await request.is_disconnected()检查，断开就停止调用模型，节省费用。 - 经过 Nginx 时要关闭代理缓冲（响应头
X-Accel-Buffering: no），见 Nginx 流式代理。 - 前端：
EventSource只支持 GET；POST 请求体（对话消息）通常用fetch+ReadableStream自己解析 SSE。
生命周期 lifespan：应用启动时创建共享资源（HTTP 客户端连接池、数据库连接池、加载模型配置），关闭时释放。不要每个请求新建 httpx.AsyncClient。
测试
TestClient(app)：同步写法测试，内部运行事件循环，最常用；httpx.AsyncClient(transport=ASGITransport(app=app))：在异步测试中使用；app.dependency_overrides：替换模型客户端、数据库、鉴权；- 测试里永远不要调用真实的模型 API——慢、花钱、结果不稳定。
部署：uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4；或 gunicorn + uvicorn worker。每个 worker 是独立进程，内存里的状态（缓存、限流计数）不共享，需要共享的放 Redis。容器化见 Docker 专题。
案例
RAG 问答接口：lifespan 中创建共享的 httpx.AsyncClient（连接池、超时）；问答接口先检索，然后流式输出答案，最后发一个 event: sources 事件携带引用来源；测试用假的检索器和模型客户端，验证事件顺序。
代码示例
import asyncio
import json
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from fastapi.testclient import TestClient
@asynccontextmanager
async def lifespan(app: FastAPI):
# 启动：创建共享资源（连接池复用，设置超时）
app.state.http = httpx.AsyncClient(timeout=httpx.Timeout(30, connect=5))
yield
# 关闭：释放资源
await app.state.http.aclose()
app = FastAPI(lifespan=lifespan)
async def fake_answer(question: str) -> AsyncIterator[str]:
for word in ["根据", "制度", "，", "年假", "为", "5", "天", "。"]:
await asyncio.sleep(0.01)
yield word
def sse(data, event: str | None = None) -> str:
head = f"event: {event}\n" if event else ""
return f"{head}data: {json.dumps(data, ensure_ascii=False)}\n\n"
@app.get("/ask")
async def ask(q: str, request: Request):
async def events() -> AsyncIterator[str]:
async for token in fake_answer(q):
if await request.is_disconnected():
return # 客户端断开，停止生成
yield sse({"delta": token})
yield sse([{"title": "员工手册", "section": "3.2"}], event="sources")
yield "data: [DONE]\n\n"
return StreamingResponse(events(), media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})
# 测试：with 语句会触发 lifespan 的启动和关闭
with TestClient(app) as client:
assert isinstance(app.state.http, httpx.AsyncClient)
with client.stream("GET", "/ask", params={"q": "年假几天"}) as r:
body = "".join(r.iter_text())
blocks = body.strip().split("\n\n")
answer = "".join(json.loads(b[6:])["delta"] for b in blocks if b.startswith("data: {"))
print(answer) # 根据制度，年假为5天。
print(blocks[-2].splitlines()[0], blocks[-1]) # event: sources data: [DONE]
print(app.state.http.is_closed) # True：关闭时已释放
pytest 中的异步测试写法（需要 pytest-asyncio 或 anyio 插件）：
import httpx
import pytest
@pytest.mark.anyio
async def test_health():
from app import app # 你的应用
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
r = await client.get("/healthz")
assert r.status_code == 200
易错点
- 流式响应中途出错仍想返回 500：做不到，响应头已发出。在流里发送错误事件。
- 不检查客户端断开：用户关掉页面，后端仍在生成并计费。
- 每个请求创建新的 HTTP 客户端：无法复用连接，延迟高，还可能耗尽文件描述符。
- 用
TestClient(app)但不用with：lifespan 不会执行，app.state上的资源不存在。 - 多 worker 部署后，内存缓存 / 限流 / 会话状态各进程不一致：共享状态放 Redis。
- 测试调用真实模型：CI 不稳定且产生费用。
面试常问
见 FastAPI 面试题：怎么实现流式输出、SSE 和 WebSocket 的区别、怎么测试依赖外部服务的接口、部署时 worker 数怎么定。