用异步 Python 搭建流式推理接口
介绍通过异步生成器接收模型输出，再用 FastAPI StreamingResponse 向客户端逐块传输的实现方式。文章还强调流式接口的错误处理，以及 asyncio 对高并发等待场景的适用性。
AI 深度解读
把模型回答攒齐再返回，用户可能要盯着空白页面等 5–30 秒。流式接口能让内容逐步出现，但给请求加上 stream=True，距离一个可靠的服务还有几步。
作者用 Python、asyncio 和 FastAPI 串起了一条实现路径：接收模型增量输出，再通过 SSE 推给客户端。真正值得看的，是断连、超时和慢客户端这些边界。不过，文中的代码更适合作为实现骨架，部分生产可用性的判断仍需补证据。
为什么流式接口适合用 asyncio
这类接口的大部分时间都在等上游模型返回数据，属于 IO 密集型任务。asyncio 可以在等待期间切换到其他连接，避免为每个会话单独创建线程。作者认为，单个进程可以承载数百个并发流式会话，但没有提供硬件配置、负载模型和压测结果。
整个链路可以理解为两段：异步生成器从模型服务逐块读取内容，FastAPI 的 StreamingResponse 再把这些内容发送给客户端。异步生成器每次 yield 一小块数据，下游就有机会立即处理。
下面这段解决的是上游读取：请求 OpenAI 的 Chat Completions API，解析 SSE 数据行，提取文本增量。
import json
import httpx
from typing import AsyncGenerator
API_KEY = "your-api-key-here"
async def stream_inference(
prompt: str,
model: str = "gpt-4o",
) -> AsyncGenerator[str, None]:
"""Stream tokens from an upstream LLM API."""
async with httpx.AsyncClient(timeout=60.0) as client:
async with client.stream(
"POST",
"https://api.openai.com/v1/chat/completions",
headers={"Authorization": f"Bearer {API_KEY}"},
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"stream": True,
},
) as response:
response.raise_for_status()
async for line in response.aiter_lines():
if not line.startswith("data: "):
continue
chunk = line[6:]
if chunk == "[DONE]":
break
try:
data = json.loads(chunk)
content = data["choices"][0]["delta"].get("content", "")
if content:
yield content
except (json.JSONDecodeError, KeyError):
continue
重点看 client.stream() 和 response.aiter_lines()：前者以流式方式发起请求，后者逐行读取响应，不必先收完整个响应体。data: 后面是 JSON，[DONE] 表示上游结束，文本来自 choices[0]["delta"]。
作者特别提醒，不要在事件循环里直接用同步的 requests 读取流。即使设置 stream=True，同步等待仍会阻塞事件循环，拖累其他连接。
这里还有一个术语差别：代码交付的是文本增量块，不保证一块恰好对应一个模型 token。后续设计缓冲区和前端渲染时，都应该按「块」理解。
浏览器迟迟收不到内容，可能卡在代理层
下游接口是 POST /inference/stream，请求包含 prompt 和 model，默认模型为 gpt-4o；空白提示词返回 400，错误信息为 Empty prompt。
FastAPI 通过 StreamingResponse 返回数据，并设置 media_type="text/event-stream"。包装函数把每块文本编码为 data: {"token": ...}\n\n，末尾发送 data: [DONE]\n\n。两个换行用于分隔 SSE 事件，不能随便省掉。
作者强调了两个响应头：Cache-Control: no-cache 和 X-Accel-Buffering: no。后者用于 nginx，避免代理攒够缓冲区后才批量转发。应用层已经逐块输出，代理层仍然可能把它变成一次性返回。
不过，原文把 no-cache 解释成禁止代理缓冲并不准确。缓存控制和响应缓冲是不同机制，不能只凭这个响应头判断流式链路是否畅通。
下面的命令用于检查接口是否逐步返回：
curl -N -X POST http://localhost:8000/inference/stream \
-H "Content-Type: application/json" \
-d '{"prompt": "Explain async/await in three sentences"}'
要看的是 -N，它关闭 curl 自己的输出缓冲。如果内容延迟后一起出现，应继续检查 nginx 的 X-Accel-Buffering 和响应的 text/event-stream 类型。
前端接入还有一个原文没展开的点：这是 POST 接口，浏览器原生 EventSource 不能直接携带这样的 POST 请求体，需要用 fetch 读取响应流并处理 SSE 事件边界。
断连、卡住和消费过慢，要分别处理
流式接口的可靠性，取决于异常发生后能否及时停止和释放资源。
第一类是用户离开页面。作者依靠 FastAPI/Starlette 将断连传播为 asyncio.CancelledError，在生成器中捕获，停止继续读取上游，避免为已经离开的用户消耗资源。
方向成立，但示例在 finally 中仍然 yield 结束标记。断连或生成器关闭时，不能假设还能继续发送数据，也不能把正常完成、取消和失败都当作同一种结束。
第二类是模型迟迟不返回。示例中的 httpx.AsyncClient(timeout=60.0) 之外，作者又设计了两层限制：total_timeout=120.0 控制全程，token_timeout=10.0 限制等待下一块输出的时间。
实现通过 loop.time() 计算截止时间，再用 asyncio.wait_for(gen.__anext__(), timeout=min(token_timeout, remaining)) 等待下一块。这样既能限制长请求，也能发现输出中途停滞。不过，这实际是块间超时，是否适合设为 10 秒，要结合模型的响应行为判断。
第三类是客户端读得慢。作者用有界 asyncio.Queue(maxsize=buffer_size) 分离生产和消费，默认 buffer_size=20。当 20 个未消费块占满队列，await queue.put(chunk) 就暂停生产者，这就是背压：下游处理不过来，上游读取也要放慢。
它限制的是队列中的块数，并非精确限制内存字节数。原文用 10 000-token 长回答说明风险，但没有给内存测试；直接串接异步生成器本身也可能传递背压，额外队列是否必要，要看实际链路。
此外，生产者在 finally 中执行 await queue.put(None)，如果消费者已经退出且队列满了，清理过程可能卡住。这段代码还需要补齐取消路径。
我的判断
作者选对了关注点：异步读取、代理缓冲、断连清理、双层超时和慢消费者，确实比只展示一个 StreamingResponse 更接近实际问题。
但它没有提供可复现的对照数据。「首个 token 不到一秒」「单进程数百并发」都应视为文中的概括，不能直接作为项目承诺。流式传输能减少等待完整回答的时间，却不能保证模型何时产出第一块内容；避免代理超时也取决于代理配置和输出间隔。
代码还有一个整合缺口：路由实际只调用了 stream_inference，后面的 stream_with_timeout、buffered_stream 没有接进去。因此，照抄路由不会自动获得全部保护。上游异常如果发生在响应已经开始之后，也不能指望再切换成普通 HTTP 错误响应。
我的取舍是先保留直接生成器链路，明确完成、失败和取消的语义，再按需求增加队列。作者末尾提到认证和限流，但没有实现，这部分也需要项目自行补齐。
能拿来做什么
- 为前端 AI 对话页接入该 POST 接口时，用
fetch读取响应流，按 SSE 事件边界解析内容，并把用户停止操作接到请求取消上，验证服务端是否同步停止读取上游。 - 部署到 nginx 后，先用
curl -N检查真实访问入口，再检查页面渲染，避免把代理缓冲误判成模型慢或前端更新不及时。 - 将总时长和块间等待分别配置，并让上游失败、正常完成、用户取消有可区分的处理结果，避免页面把截断内容当成完整回答。
- 如果增加生产者队列，先验证慢连接和队列满时的取消清理，再用
httpx.AsyncClient编写并发流测试，记录首块延迟、完成耗时与内存变化。