FastAPI流式响应StreamingResponse完全指南

📂 关联资源：WebSocket实时通信 · 异步编程深度解析

流式响应是什么？

核心对比

传统HTTP是全量等包模式，服务器处理完所有数据才一次性返回，大文件、AI对话这类场景会造成内存溢出或用户等待焦虑。而流式响应是边处理边推包模式，用异步生成器逐步发送数据：

✅ AI打字机：每生成一个字/词就推
✅ 大文件：分块读内存，避免一次性加载
✅ 实时日志：新日志写入队列就发
✅ 进度条：每完成一步更新状态

选SSE还是WebSocket？

技术	适用场景	复杂度
StreamingResponse	简单流式文本/二进制	⭐
SSE（基于前者）	标准化单向推 + 自动重连	⭐⭐
WebSocket	双向实时互动（聊天、游戏）	⭐⭐⭐

StreamingResponse基础与优化

极简入门版

先用异步生成器 + StreamingResponse 发个数字序列：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def num_stream():
    for i in range(1, 6):
        yield f"收到第{i}个数据块\n"
        await asyncio.sleep(0.5)  # 模拟处理延迟

@app.get("/simple-stream")
async def simple_stream():
    return StreamingResponse(
        num_stream(),
        media_type="text/plain",
        headers={
            "Cache-Control": "no-cache",  # 禁用浏览器/代理缓存
            "Connection": "keep-alive"    # 保持长连接
        }
    )

处理客户端断开

客户端中途刷新/关闭会触发 asyncio.CancelledError，记得在生成器里捕获并释放资源：

async def safe_num_stream():
    try:
        for i in range(1, 6):
            yield f"收到第{i}个数据块\n"
            await asyncio.sleep(0.5)
    except asyncio.CancelledError:
        print("客户端主动断开！")  # 打印日志、释放DB连接等
        raise  # 必须重新抛出，FastAPI才会正确清理

SSE服务器发送事件

SSE是标准化的单向流式响应，浏览器原生支持 EventSource API，自带重连、事件类型区分功能，无需额外依赖。

标准SSE格式

SSE数据必须以 data: 开头，每条数据结束用两个换行符 \n\n，还可选加 event: 区分事件、id: 标识断点、retry: 调整重连间隔（毫秒）。

import json
from datetime import datetime

async def sse_status_stream():
    # 先发重连间隔建议（10秒）
    yield "retry: 10000\n\n"
    
    for i in range(1, 11):
        # 普通数据事件
        normal_data = {"step": i, "progress": i*10, "time": datetime.now().isoformat()}
        yield f"data: {json.dumps(normal_data)}\n\n"
        
        # 每3步发一个状态事件
        if i % 3 == 0:
            status_data = {"state": "checkpoint_reached", "step": i}
            yield f"event: checkpoint\ndata: {json.dumps(status_data)}\n\n"
        
        await asyncio.sleep(0.8)

@app.get("/sse-status")
async def sse_status():
    return StreamingResponse(
        sse_status_stream(),
        media_type="text/event-stream",  # 必须是这个MIME类型
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"  # 关键！禁用Nginx缓冲
        }
    )

AI对话打字机效果

模拟或调用真实AI时，逐token推给前端，是提升AI产品体验的核心。

模拟本地AI打字机

async def mock_ai_stream(user_msg: str):
    ai_reply = f"您的问题是「{user_msg}」，让我为您逐步分析：首先，我们要理清需求边界；其次，梳理实现路径；最后，优化细节。"
    
    accumulated = ""
    for char in ai_reply:
        accumulated += char
        chunk = {
            "type": "token",
            "content": char,
            "full": accumulated  # 可选，方便前端直接更新完整内容
        }
        yield f"data: {json.dumps(chunk)}\n\n"
        await asyncio.sleep(0.02)  # 模拟AI生成速度

from fastapi import Request
@app.post("/mock-ai-chat")
async def mock_ai_chat(req: Request):
    body = await req.json()
    return StreamingResponse(
        mock_ai_stream(body.get("msg", "")),
        media_type="text/event-stream",
        headers={"X-Accel-Buffering": "no"}
    )

文件流与日志实时推送

大文件分块下载

用 aiofiles 异步读文件，避免阻塞事件循环，同时控制内存占用：

import aiofiles
from pathlib import Path

async def file_chunk_stream(file_path: str, chunk_size: int = 8192):
    p = Path(file_path)
    if not p.exists():
        yield b"File not found"
        return
    
    async with aiofiles.open(p, "rb") as f:
        while chunk := await f.read(chunk_size):
            yield chunk

@app.get("/download/{filename}")
async def download_file(filename: str):
    local_path = f"./data/{filename}"  # 实际项目需做安全校验
    return StreamingResponse(
        file_chunk_stream(local_path),
        media_type="application/octet-stream",
        headers={
            "Content-Disposition": f"attachment; filename={filename}"
        }
    )

生产环境关键配置

Nginx禁用缓冲

如果用Nginx做反向代理，必须禁用SSE/流式响应路径的缓冲，否则数据会被攒成大段才发给前端：

location /stream/ {
    proxy_pass http://uvicorn_backend;
    proxy_http_version 1.1;
    
    # 核心配置：禁用所有缓冲
    proxy_buffering off;
    proxy_cache off;
    proxy_set_header X-Accel-Buffering no;
    
    # 超时时间适配长连接
    proxy_read_timeout 300s;
    proxy_send_timeout 300s;
    keepalive_timeout 300s;
}

Uvicorn/Gunicorn参数

使用异步Worker（如 uvicorn.workers.UvicornWorker），增加超时时间，限制并发连接：

# Uvicorn直接启动（开发/小规模生产）
uvicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --timeout-keep-alive 300 --limit-concurrency 100

前端极简集成方案

SSE原生EventSource

浏览器原生支持，无需依赖库，自动重连（默认3秒）：

<!DOCTYPE html>
<html>
<body>
    <div id="output"></div>
    <script>
        const es = new EventSource("/sse-status");
        const output = document.getElementById("output");

        // 监听默认事件（无event: 前缀的）
        es.onmessage = (e) => {
            const data = JSON.parse(e.data);
            output.innerHTML += `<p>默认事件：进度 ${data.progress}%</p>`;
        };

        // 监听自定义checkpoint事件
        es.addEventListener("checkpoint", (e) => {
            const data = JSON.parse(e.data);
            output.innerHTML += `<p style="color:blue;">✅ 检查点 ${data.step} 达成！</p>`;
        });

        // 监听错误（自动重连前触发）
        es.onerror = () => console.log("连接出错，尝试重连...");
    </script>
</body>
</html>

📝 总结：FastAPI的StreamingResponse是构建流式应用的核心，配合SSE标准化格式和生产环境的缓冲/超时配置，能快速实现高可用的AI对话、实时日志、大文件下载等功能。

#FastAPI流式响应StreamingResponse完全指南

#目录

#流式响应是什么？

#核心对比

#选SSE还是WebSocket？

#StreamingResponse基础与优化

#极简入门版

#处理客户端断开

#SSE服务器发送事件

#标准SSE格式

#AI对话打字机效果

#模拟本地AI打字机

#文件流与日志实时推送

#大文件分块下载

#生产环境关键配置

#Nginx禁用缓冲

#Uvicorn/Gunicorn参数

#前端极简集成方案

#SSE原生EventSource