摘要
解决传统 HTTP 请求在 LLM 长文本生成时导致的客户端长时间阻塞与 TTFT 极高的问题。适用于需为 AI 应用提供打字机效果、优化交互体验的后端开发工程师。读完可获得基于 FastAPI 与 SSE（Server-Sent Events）技术的流式响应实现代码、协议规范及生产环境避坑指南。

关键词
SSE、流式输出、FastAPI、打字机效果、TTFT、大模型推理

1. 背景与问题
   大模型（LLM）生成完整回复通常耗时数秒至数十秒，传统请求-响应模型导致用户界面长时间无响应，体验极差。后端开发者需在服务端与客户端之间建立单向推送通道，将生成内容切片后实时推送到前端。SSE 技术基于原生 HTTP 协议，开销极低，是实现单向流式输出的行业标准方案，能有效压低 TTFT（首 Token 延迟），实现丝滑的逐字输出效果。

2. 核心概念与环境
   2.1 依赖与版本
   组件 版本要求 作用
   Python >= 3.9 运行环境
   fastapi >= 0.100.0 提供异步 Web 框架
   uvicorn >= 0.23.0 ASGI Web 服务器
   httpx >= 0.24.0 异步 HTTP 客户端，用于流式请求上游 LLM
   openai >= 1.0.0 大模型 SDK（模拟流式调用）
   以下代码为最小化演示实现，生产环境需接入真实的鉴权、限流与日志追踪组件。

2.2 关键术语
SSE（Server-Sent Events）：基于 HTTP 协议的服务器推送技术。响应头需设为 Content-Type: text/event-stream，数据格式要求为 data: \n\n。

TTFT（Time To First Token）：从用户发起请求到接收到第一个字符的时间，决定打字机效果的“启动”速度。

ASGI：异步服务器网关接口，FastAPI 的基础，支持高并发流式处理。

Event Stream 格式：SSE 的标准数据结构，每行以 data: 开头，以双换行符 \n\n 结尾。

3. 核心内容
   3.1 可运行代码：FastAPI SSE 端点与流式转发
   python

file: sse_streaming.py

pip install fastapi uvicorn httpx openai

import asyncio
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
import uvicorn

app = FastAPI(title=“AI Streaming API”)
client = AsyncOpenAI(api_key=“your_api_key”)

async def generate_stream(prompt: str, request: Request):
try:
# 向上游大模型发起流式请求
response = await client.chat.completions.create(
model=“gpt-3.5-turbo”,
messages=[{“role”: “user”, “content”: prompt}],
stream=True,
)

    async for chunk in response:
        # 检查客户端是否已断开，若断开则终止上游生成，节约资源
        if await request.is_disconnected():
            break
            
        content = chunk.choices[0].delta.content
        if content:
            # SSE 规范：以 data: 开头，以 \n\n 结尾
            yield f"data: {content}\n\n"
            
except asyncio.CancelledError:
    # 捕获任务取消异常（客户端强行关闭连接）
    pass
except Exception as e:
    yield f"data: [ERROR] {str(e)}\n\n"
finally:
    # 发送结束标志
    yield "data: [DONE]\n\n"

@app.post(“/api/chat/stream”)
async def chat_stream(request: Request):
body = await request.json()
prompt = body.get(“prompt”, “”)

# 核心：返回 StreamingResponse，配置流式响应头
return StreamingResponse(
    generate_stream(prompt, request),
    media_type="text/event-stream",
    headers={
        "Cache-Control": "no-cache",
        "Connection": "keep-alive",
        "X-Accel-Buffering": "no",  # 禁用 Nginx 缓冲，防止流式被合并
    }
)

if name == “main”:
uvicorn.run(app, host=“0.0.0.0”, port=8000)
入口命令：

bash
python sse_streaming.py
测试命令（使用 curl 验证流式效果）：

bash
curl -N -X POST http://localhost:8000/api/chat/stream
-H “Content-Type: application/json”
-d ‘{“prompt”: “用一句话解释什么是SSE”}’
预期输出（终端会以打字机速度逐行打印）：

text
data: SSE

data: （

data: Server

data: -

data: Sent

data: Events

data: ）

data: 是

data: 一种

data: 基于

data: HTTP

data: 协议

data: 的

data: 服务器

data: 单向

data: 推送

data: 技术

data: 。

data: [DONE]
3.2 SSE vs WebSocket 选型对比
维度 SSE WebSocket
通信方向 单向（Server -> Client） 双向全双工（Bidirectional）
底层协议 纯 HTTP/HTTPS 独立的 ws/wss 协议（握手后升级）
实现复杂度 低（无需引入额外网关，基于标准 HTTP） 高（需处理心跳、重连、状态同步）
资源消耗 极低（单个连接复用一个 HTTP 流） 较高（长连接占用文件描述符）
适用场景 AI 文本流式生成、新闻订阅、实时日志 聊天室、即时通讯、协同编辑（高频双向交互）
3.3 避坑清单
风险点 触发条件 处置方式
前端收到数据不渲染 缺少 \n\n 分隔符，或内容未以 data: 开头 严格检查 yield 格式，必须为 f"data: {content}\n\n"
代理服务器下失去流式效果 Nginx / 云负载均衡默认开启缓冲 响应头设置 X-Accel-Buffering: no，或在 Nginx 设置 proxy_buffering off;
客户端断开后持续计费 后端未侦测到断开，继续请求大模型 在生成器中通过 await request.is_disconnected() 及时中断上游连接
CORS 跨域拦截 前端与后端域名不一致，且未配置预检请求 FastAPI 中引入 CORSMiddleware，并允许 Content-Type: application/json 及流式响应头
4. 常见问题与排错
现象：调用接口后，内容全部生成完毕才一次性显示在前端，失去了打字机效果。
原因：中间存在代理服务器（如 Nginx、Cloudflare）对响应进行了缓冲（Buffering）。
解决：确保 FastAPI 响应头包含 X-Accel-Buffering: no，并检查运维层的代理配置是否禁用了缓冲。

现象：服务端已停止生成，但前端连接未关闭，页面卡死。
原因：SSE 流未发送结束标志，或连接被闲置超时。
解决：在流的末尾显式发送 data: [DONE]\n\n，前端 JS 的 onmessage 中检测到该标识后主动调用 eventSource.close()。

现象：并发测试时，后端抛出 RuntimeError: Cannot call “send” once a close message has been sent.
原因：ASGI 服务器中，客户端已断开连接，但后台协程仍在尝试 yield 数据。
解决：捕获 asyncio.CancelledError，并在循环中定期检查 request.is_disconnected()。

现象：流式响应中部分中文字符被截断显示为乱码。
原因：字符编码错误或代理服务器对字节流进行了切分。
解决：确保 FastAPI 中的字符以 UTF-8 编码生成，且上游 LLM 返回的 chunk 解码正确（OpenAI SDK 自动处理）。

5. 优化建议与边界
   断连释放：LLM 生成属于高算力消耗操作。客户端关闭页面时必须能触发服务端中断，防止资源浪费（如应用在 Vercel 等服务上会导致 Function 超时费用）。

心跳机制：若大模型思考时间过长（如逻辑推理），网络中断可能静默断开。可每隔 15 秒发送注释行 : ping\n\n 维持连接。

适用边界：SSE 适合单向数据流。若业务场景要求前端在上传大文件的同时接收实时语音流（双向高频），应优先选用 WebSocket。SSE 在 HTTP/1.1 下有 6 个并发连接限制（针对同域名），但在 HTTP/2 下限制消除。

6. 求职与面试关联
   SSE 与 WebSocket 的核心区别是什么？
   思路：SSE 基于 HTTP 单向推送，轻量易用；WebSocket 是全双工协议，需独立握手，适合高频双向交互。

为什么大模型对话系统普遍采用 SSE 而不是 WebSocket？
思路：LLM 输出通常是单向的（Server -> Client 文本流），使用 SSE 无需维护双向状态，HTTP 天然支持重连，网络兼容性更好。

SSE 的数据格式规范要求是什么？
思路：Content-Type 为 text/event-stream；数据体需遵循 data: \n\n 的双换行结构；支持事件类型及重连间隔定义。

如何解决 Nginx 代理下的 SSE 缓冲问题？
思路：服务端响应头注入 X-Accel-Buffering: no；Nginx 配置中设置 proxy_buffering off;。

流式响应场景下，如何实现客户端的断连检测以节省资源？
思路：在生成器的每次循环中调用 request.is_disconnected()，或在框架中捕获协程的 CancelledError，及时中止对上游大模型的请求。

SSE 的自动重连机制是如何工作的？
思路：底层 EventSource 客户端在连接断开后默认发起重连。服务端可通过 retry: <毫秒> 字段控制重连间隔。

总结
SSE 基于标准 HTTP 协议实现单向流式推送，无需引入 WebSocket 网关，是 AI 对话场景的首选方案。

FastAPI 的 StreamingResponse 结合 httpx 异步客户端，可高效透传大模型的 chunk 数据。

数据规范必须严格遵守 data: xxx\n\n 的双换行结构，缺失将导致前端无法解析。

流式效果的成败在于代理层配置，必须通过 X-Accel-Buffering: no 禁用 Nginx 等组件的缓冲。

必须实现客户端断连检测（request.is_disconnected()），防止流式连接断开后无谓消耗大模型算力。