流式输出实战:一个接口给所有大模型加上“打字机“效果
·
用户问完问题,盯着空白等上好几秒,和 0 点几秒就看到字开始往外蹦,体验天差地别。后者靠的就是"流式输出"(streaming)。本文用聚合 API 的 OpenAI 兼容接口,一份代码给所有模型加上流式输出,前端照抄就能用。
一、流式和非流式,差别在哪
默认情况下,chat.completions 是"攒够了再给":模型把整段回答生成完,一次性返回。所以你看到的是——转圈、转圈、突然一整坨文字。
流式输出反过来:模型生成一个 token,就吐一个 token。你看到的是字一个字往外蹦,像打字机。
这个差距用两个指标衡量:
- 首字延迟(TTFT,Time To First Token):从发出请求到第一个字出现。非流式要等整段生成完,流式几乎立刻开始。
- 体感:同样是几秒出完,流式让你觉得"它一直在干活",非流式让你觉得"卡了"。
对聊天、客服、写作、代码补全这类场景,流式基本是标配。
二、环境准备
from openai import OpenAI
client = OpenAI(
base_url="https://api.tokenportal.ai/v1", # 控制台获取的网关地址
api_key="YOUR_TOKENPORTAL_KEY", # 控制台获取的 Key
)
三、最小实现
非流式(你大概率已经会了):
resp = client.chat.completions.create(
model="deepseek-v4-pro",
messages=[{"role": "user", "content": "用三句话介绍你自己"}],
)
print(resp.choices[0].message.content)
流式,只差一个参数 stream=True:
resp = client.chat.completions.create(
model="deepseek-v4-pro",
messages=[{"role": "user", "content": "用三句话介绍你自己"}],
stream=True,
)
for chunk in resp:
delta = chunk.choices[0].delta
if delta.content: # 首个 chunk 可能只有 role 没有正文,要判空
print(delta.content, end="", flush=True)
跑一遍,你会看到文字逐字往外蹦。
四、两个要注意的细节
delta.content可能为空:流式返回的第一个分片,有时只有role没有正文,所以上面加了if delta.content判空,不然会报错。- 结束信号看
finish_reason:最后一个分片的finish_reason会从None变成stop(或length表示被截断)。生产代码靠它判断"是不是说完了"。
五、接到前端(Web)
本地 print 只是演示。真正落到网页,是把流式接到 SSE(Server-Sent Events,服务端推送),服务端每收到一个分片就往浏览器推一段,浏览器边收边渲染。FastAPI 一个最小示例:
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
@app.post("/chat")
def chat():
def gen():
resp = client.chat.completions.create(
model="deepseek-v4-pro",
messages=[{"role": "user", "content": "你好"}],
stream=True,
)
for chunk in resp:
if chunk.choices[0].delta.content:
yield f"data: {chunk.choices[0].delta.content}\n\n"
return StreamingResponse(gen(), media_type="text/event-stream")
前端用 EventSource 接收,就能做出对话产品的打字机效果。
六、为什么用聚合 API 做流式
流式格式各家大体一致,但细节(分片结构、结束标志)有差异,逐个适配费时。聚合 API 把主流模型统一成 OpenAI 兼容格式,stream=True 一套代码全通,切换模型不用改调用逻辑。
更多推荐
所有评论(0)