流式输出Stream实现

入门·10 分钟·通过可运行代码讲解「流式输出Stream实现」，覆盖环境准备、核心代码、参数说明、异常处理、工程化改造和练习任务，适合 AI 编程入门。

流式输出Stream实现

流式输出的效果是：模型一边生成，页面或命令行一边显示，而不是等完整回答生成后一次性返回。ChatGPT 打字机一样的显示效果，本质上就是流式输出。

为什么需要流式输出

如果模型回答需要 10 秒，非流式接口会让用户盯着空白页面等待 10 秒。流式输出可以在第一批 token 生成后马上展示，用户会感觉响应更快。

流式输出特别适合：聊天机器人、长文生成、代码生成、语音播报前的文本生成、实时客服等场景。

基本思路

普通调用是：请求一次，等待完整响应。流式调用是：请求一次，服务端不断返回小片段，客户端每收到一段就展示一段。

伪代码如下：

for chunk in stream:
    text = parse(chunk)
    print(text, end='', flush=True)

关键点是 flush=True，否则命令行可能不会立刻显示。

Python命令行示例

下面是一个简化结构，具体 SDK 名称可能随版本变化：

from openai import OpenAI
client = OpenAI()
 
stream = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=[{'role': 'user', 'content': '写一段关于RAG的解释'}],
    stream=True,
)
 
for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end='', flush=True)
print()

这段代码每收到一个增量片段就输出。

Web里怎么做

Web 场景常用 Server-Sent Events 或 WebSocket。后端接收模型流，再把片段转发给浏览器。浏览器不断把新文本追加到页面上。

流程是：

前端提交用户问题；
后端调用模型流式接口；
后端把 chunk 转成 SSE 消息；
前端监听事件并追加文本；
完成后关闭连接。

流式输出的错误处理

流式输出不是只有成功路径。你要考虑：

连接中途断开；
模型返回错误；
用户点击停止生成；
输出中包含敏感内容；
前端重复追加导致文本错乱。

真实项目中要为每次生成分配请求 ID，并在日志里记录开始、结束、错误和耗时。

和非流式的区别

流式输出提升体验，但代码更复杂。非流式适合后台批处理、短文本分类、结构化 JSON 返回。流式适合用户正在等待长回答的交互场景。

练习任务

做一个命令行聊天程序：用户输入问题后，模型回答逐字显示。增加一个计时器，比较首字输出时间和完整响应时间。

小结

流式输出解决的是体验问题。它不会让模型本身更快完成，但能让用户更早看到结果。实现时重点关注 chunk 解析、前端追加、异常中断和日志记录。

← 上一篇：Function Calling使用下一篇：用API实现多轮对话 →

反馈问题