做 AI 应用的人几乎都踩过这个坑:请求里 stream=True 开着,后端返回的也是 StreamingResponse,可页面还是等了好几秒,然后整段回答啪地一下全出来。要理解这件事,得先承认一个前提:开了流式,只说明链路里某一段支持增量传输,不代表内容真的被逐段送到了用户眼前。这篇顺着正文从模型到页面的路径,把每一层可能把内容攒住的地方拆开。
首字延迟与总耗时
判断流式有没有生效,先得把两个时间分开。首字延迟(TTFT)是请求发出后,客户端多久收到第一段有效正文;总耗时要等到整个回答生成并传输完。流式输出改善的是前者——它让用户提前看到开头,并不减少模型生成完整答案所需的时间。很多时候「开了流式还要等」的错觉,就是把这俩混在了一起:模型其实已经在逐步生成,但某一层把内容攒住了,首字延迟照样拉满。反过来,只要某一段开始出现的时刻提前了,即便最后一段仍花几秒,体验也已经改善。
一条回答的四层交接
一条流式回答要经过四次交接:模型生成正文、后端读取上游、后端转发给浏览器、浏览器解析并更新页面。每一层都可能把内容攒住。模型侧,流式接口在生成过程中持续返回增量,但未必每生成一个 token 就发一条消息;后端如果先把分片收进列表、等上游结束再返回,用户仍然看不到增量;中间的反向代理或网关可能缓冲响应;前端如果等整个响应读完才处理,前面的流式也白流了。所以「模型支持流式」「后端返回 SSE」「用户看到逐段输出」是三件独立的事,谁也不能替代谁。
SSE 只规定格式
SSE(Server-Sent Events)是 HTTP 上持续发送事件的格式,一个事件通常带事件名和数据,以空行收尾:
event: token
data: {"text":"新增的正文"}
客户端按空行切分事件,再取出正文显示。但 SSE 只规定事件怎么组织,不规定后端什么时候发。同样用 text/event-stream,把一段正文立刻发出去、和把全文拼完最后只发一条事件,两种实现都合法,只有前者让用户提前看到。所以光看响应头里有没有 text/event-stream,或者看框架是不是 StreamingResponse,判断不了流式有没有真正生效——还得看数据在后端什么时候被交给响应对象。
后端攒着不发
最常见的坑就在后端:代码名义上返回了流,实际上在生成器里把整段拼完才 yield。这段是反面教材,先看错的写法:
# 错误:把整段回答拼完才一次性返回,流式名存实亡
import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.post("/chat")
async def chat():
async def gen():
chunks = []
async for piece in upstream_stream(): # 上游已经在流式吐字
chunks.append(piece)
yield "".join(chunks) # 却在这里一次性吐出
return StreamingResponse(gen(), media_type="text/event-stream")
上游其实已经在逐段吐字,但 gen 把每一段都收进列表,等上游结束才 yield 一次,用户看到的还是「整段一起出来」。正确的做法是拿到一段就立刻按 SSE 格式转发一段:
# 正确:收到一段就包装成一条 SSE 事件立即转发
import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.post("/chat")
async def chat():
async def gen():
async for piece in upstream_stream():
if not piece:
continue
yield f"event: token
data: {json.dumps({'text': piece})}
"
return StreamingResponse(gen(), media_type="text/event-stream")
差别就在 yield 的位置:在循环里逐段 yield,而不是循环外一次 yield。这一处改对,首字延迟立刻从「整段生成完」降到「第一段生成完」。
代理缓冲
后端改对了,经过 Nginx、网关或 CDN 之后可能又被打回原形——代理默认会把小响应缓冲起来,凑够一块再发给客户端。本地直连 Uvicorn 测出来是流式,不代表上了代理还是流式。Nginx 里要显式关掉代理缓冲:
location /chat {
proxy_buffering off;
proxy_cache off;
proxy_pass http://127.0.0.1:8000;
}
应用层还可以给响应加一个 X-Accel-Buffering: no 头,表达「不希望被缓冲」的意图,但最终要看代理实际配置——头是建议,不是命令。
测量口径
流式到底快没快,取决于你在哪一端计时。在前端测「第一次显示文字」,接近用户真实感受;在后端测「第一次收到上游正文」,更适合看模型服务与上游链路。起止点不说清,数字没有比较意义。这里给一段能直接跑的 TTFT 测量脚本,对着 OpenAI 兼容接口测首字延迟:
# 测首字延迟:从发请求到收到第一段正文内容
import time
from openai import OpenAI
client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
t0 = time.time()
stream = client.chat.completions.create(
model="qwen2.5:7b",
messages=[{"role": "user", "content": "用一句话解释什么是流式输出"}],
stream=True,
)
first = None
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
first = time.time()
break
print(f"首字延迟 TTFT: {first - t0:.3f} 秒")
注意循环里要跳过空 delta——上游事件里有些 chunk 只有角色、结束标记或用量信息,没有正文内容,把它们算进首字时间会把数字虚高。
值不值得优化
最后说边界。首字延迟值得优化的前提,是用户真的盯着「第一段字出现」这件事,对话式产品尤其敏感。但如果你的产品用户本来就要等完整结果——批量生成、文件处理这类——把精力花在总耗时上更划算。排查时记住一句话:第一段正文生成了吗?如果生成了,它现在停在哪一层。顺着模型、后端、代理、前端一路查过去,通常比换模型、换框架更接近问题本身。