FastAPI StreamingResponse无法实时输出问题求助

要实现实时流式响应，需要对代码做两处关键调整：

1. 将路由函数改为异步函数
   原代码中路由函数是同步的def root()，但你的生成器是异步类型，这会导致FastAPI等待整个异步生成器执行完毕后才一次性返回所有结果。把路由函数改为async def root()，让框架能逐步处理生成器的每一次输出。

2. 添加禁用缓冲的响应头部（可选但推荐）
   部分HTTP服务器（如Uvicorn）默认会对响应做缓冲处理，添加Cache-Control: no-cache和Transfer-Encoding: chunked头部可以强制服务器实时发送每个数据块。

修改后的完整代码：

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

app = FastAPI()

async def generator():
    for i in range(10):
        yield json.dumps({"eid":i,"data":f"{i+1}data chunk","is_final_event":i ==9}) + '\n'
        await asyncio.sleep(1)

@app.get("/")
async def root():  # 改为异步函数
    headers = {
        "Cache-Control": "no-cache",
        "Transfer-Encoding": "chunked"
    }
    return StreamingResponse(generator(), media_type="application/x-ndjson", headers=headers)

验证方式 #

启动服务后，用curl测试能更直观看到实时输出：

curl http://127.0.0.1:8000/

你会看到每秒输出一条JSON数据，而非等待10秒一次性返回所有内容。

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