FastAPI StreamingResponse在Next.js fetch中被缓冲的排查与修复

一、缓冲的常见原因及对应因素排查 #

FastAPI/Starlette行为 #

FastAPI的StreamingResponse原生支持流式输出，但需注意：

• 若使用旧版Uvicorn（<0.18.0），可能存在流式处理的底层bug，建议升级到最新稳定版
• 异步生成器需确保真正异步yield，你的代码中sleep(0.5)已满足这一点，排除此问题

浏览器fetch()行为 #

fetch本身支持流式读取，但：

• 部分浏览器对text/plain类型的小chunk会自动合并缓冲，直到缓冲区满或响应结束才交付
• 你的getReader()+decoder.decode({stream: true})代码是正确的，但需结合媒体类型优化

text/plain vs text/event-stream #

这是核心影响因素：

• text/plain无明确分块协议，浏览器无强制即时交付规则，容易触发缓冲
• text/event-stream（SSE）是W3C标准的流式传输协议，浏览器会自动识别data:格式的块，收到即交付，无额外缓冲

本地开发服务器行为 #

• FastAPI端：Uvicorn的--reload热重载模式可能会拦截流式响应做中间处理，建议关闭--reload测试
• Next.js端：next dev开发服务器对代理请求有默认缓冲机制，若前端通过Next.js API路由转发FastAPI请求，需手动禁用缓冲

Nginx等反向代理缓冲 #

生产环境中Nginx默认会缓冲完整响应后再转发，必须配置禁用：

proxy_buffering off;
proxy_cache off;
chunked_transfer_encoding on;

Next.js路由/运行时行为 #

• 若通过Next.js API路由转发请求，Node.js Runtime默认会缓冲响应，建议切换为Edge Runtime（原生支持流式）
• 需在API路由中直接返回ReadableStream，避免Next.js内部处理缓冲

缺少换行或SSE格式 #

你的后端当前输出token-i\n，不符合SSE协议要求（需data: 内容\n\n格式），浏览器无法识别独立chunk，导致合并缓冲

二、生产环境安全流式传输方案 #

方案1：采用SSE协议（推荐，适配LLM流式场景） #

后端（FastAPI）代码修改

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

app = FastAPI()

async def sse_generator():
    # 发送初始化信号
    yield "data: [START]\n\n"
    for i in range(10):
        # 严格遵循SSE格式：data: 内容\n\n
        chunk = f"data: token-{i}\n\n"
        print("yielding:", chunk.strip(), flush=True)
        yield chunk
        await asyncio.sleep(0.5)
    # 发送结束标记
    yield "data: [DONE]\n\n"

@app.get("/stream")
async def stream():
    return StreamingResponse(
        sse_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
            "Connection": "keep-alive",
            # 根据实际生产环境配置CORS，避免硬编码*
            "Access-Control-Allow-Origin": "https://your-nextjs-domain.com"
        },
    )

前端（Next.js）代码修改

'use client';
import { useEffect } from 'react';

export default function StreamChat() {
  useEffect(() => {
    const eventSource = new EventSource('https://your-fastapi-domain.com/stream');
    
    eventSource.onmessage = (event) => {
      if (event.data === '[DONE]') {
        eventSource.close();
        return;
      }
      if (event.data === '[START]') {
        // 初始化聊天界面
        return;
      }
      // 实时渲染token到页面
      console.log("received token:", event.data);
    };

    eventSource.onerror = (error) => {
      console.error("SSE连接错误:", error);
      eventSource.close();
    };

    // 组件卸载时关闭连接
    return () => eventSource.close();
  }, []);

  return <div className="chat-container">正在生成回复...</div>;
}

方案2：使用Fetch流式处理（适合自定义场景） #

后端调整

修改媒体类型为application/octet-stream（避免浏览器文本缓冲），并明确分块分隔符：

@app.get("/stream")
async def stream():
    return StreamingResponse(
        token_generator(),
        media_type="application/octet-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
            "Transfer-Encoding": "chunked",
        },
    )

前端优化

手动分割chunk，避免浏览器合并：

export default async function readStream() {
  const response = await fetch("https://your-fastapi-domain.com/stream", {
    method: "GET",
    cache: "no-store",
  });

  if (!response.body) throw new Error("无响应体");

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  while (true) {
    const { value, done } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });
    // 按换行分割独立chunk
    const chunks = buffer.split("\n");
    buffer = chunks.pop(); // 保留未完成的部分
    chunks.forEach(chunk => {
      if (chunk) console.log("收到chunk:", chunk);
    });
  }

  // 处理最后剩余的内容
  if (buffer) console.log("收到最后chunk:", buffer);
}

方案3：反向代理（Nginx）生产配置 #

在FastAPI对应的location块中添加：

location /stream {
    proxy_pass http://localhost:8000;
    proxy_buffering off;
    proxy_cache off;
    chunked_transfer_encoding on;
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    # 传递客户端真实IP（可选）
    proxy_set_header X-Real-IP $remote_addr;
}

方案4：Next.js中间层优化（必须转发请求时） #

使用Edge Runtime直接返回流式响应：

// app/api/stream/route.js
export const runtime = 'edge'; // 启用Edge Runtime，支持流式

export async function GET() {
  const res = await fetch('http://localhost:8000/stream', {
    headers: {
      'Accept': 'text/event-stream',
    },
    cache: 'no-store',
  });

  return new Response(res.body, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive',
      'X-Accel-Buffering': 'no',
    },
  });
}

内容的提问来源于stack exchange，提问作者Darius Daniels