135 lines
5.8 KiB
Markdown
135 lines
5.8 KiB
Markdown
---
|
||
id: python.fastapi.streaming-testing
|
||
title: 流式输出（SSE）、生命周期、测试和部署
|
||
type: lesson
|
||
domain: python
|
||
stack: [fastapi]
|
||
tags: [SSE, StreamingResponse, lifespan, TestClient, pytest, uvicorn]
|
||
level: intermediate
|
||
prerequisites: [python.fastapi.deps-auth-errors]
|
||
est_hours: 3
|
||
related: [python.fastapi.scenario.streaming-chat-api, devops.nginx.https-stream-limit, python.fastapi.interview]
|
||
week: [5, 12]
|
||
order: 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 流式代理](../../devops/nginx/02-https-stream-limit.md)。
|
||
- 前端：`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 专题](../../devops/docker/index.md)。
|
||
|
||
## 案例
|
||
|
||
RAG 问答接口：lifespan 中创建共享的 `httpx.AsyncClient`（连接池、超时）；问答接口先检索，然后流式输出答案，最后发一个 `event: sources` 事件携带引用来源；测试用假的检索器和模型客户端，验证事件顺序。
|
||
|
||
## 代码示例
|
||
|
||
```python
|
||
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` 插件）：
|
||
|
||
```python
|
||
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 面试题](interview.md)：怎么实现流式输出、SSE 和 WebSocket 的区别、怎么测试依赖外部服务的接口、部署时 worker 数怎么定。
|