普通接口调用通常要等模型完整生成后才返回结果。流式输出可以让内容一段一段显示，适合聊天页面、命令行工具和长文本生成场景。

什么是流式输出？

普通请求的流程通常是：

发送请求 → 等待模型生成 → 一次性返回完整内容

流式请求则是：

发送请求 → 持续接收内容片段 → 实时显示 → 生成结束

用户不需要一直等待空白页面，可以先看到已经生成的内容，整体体验会更自然。

---

一、普通调用和流式调用的区别

普通调用示例：

response = client.chat.completions.create(
    model="your-model-name",
    messages=[
        {"role": "user", "content": "介绍一下 Python"}
    ]
)

print(response.choices[0].message.content)

流式调用只需要增加 stream=True：

stream = client.chat.completions.create(
    model="your-model-name",
    messages=[
        {"role": "user", "content": "介绍一下 Python"}
    ],
    stream=True,
)

返回结果不再是一次完成的文本，而是一组连续的数据片段。

---

二、Python 实现最小流式输出

from openai import OpenAI

client = OpenAI(
    api_key="your-api-key",
    base_url="https://your-api-domain.com/v1",
)

stream = client.chat.completions.create(
    model="your-model-name",
    messages=[
        {"role": "user", "content": "用简单语言介绍 Python"}
    ],
    stream=True,
)

for chunk in stream:
    content = chunk.choices[0].delta.content
    if content:
        print(content, end="", flush=True)

print()

这里有两个关键点：

• stream=True：开启流式响应
• flush=True：让终端立即刷新内容

如果不加 flush=True，某些环境可能会积累一段内容后才显示。

---

三、为什么要判断 content 是否为空？

流式响应中的每个片段不一定都有文本内容。有些片段可能携带角色信息、结束标记或其他元数据。

因此不要直接打印：

print(chunk.choices[0].delta.content)

更稳妥的写法是：

content = chunk.choices[0].delta.content
if content:
    print(content, end="", flush=True)

这样可以避免输出 None，也能兼容不同接口返回的片段结构。

---

四、把流式内容保存成完整文本

有时你既想实时显示，又想在结束后保存完整回答，可以使用列表收集内容：

parts = []

for chunk in stream:
    content = chunk.choices[0].delta.content
    if content:
        print(content, end="", flush=True)
        parts.append(content)

full_text = "".join(parts)
print("\n总长度：", len(full_text))

这种方式适合：

• 保存聊天记录
• 写入数据库
• 生成 Markdown 文件
• 后续进行结构化处理

---

五、流式输出中的异常处理

网络中断或服务异常时，流式循环可能提前结束。因此建议加上异常处理：

try:
    stream = client.chat.completions.create(
        model="your-model-name",
        messages=[
            {"role": "user", "content": "写一段 Python 说明"}
        ],
        stream=True,
    )

    for chunk in stream:
        content = chunk.choices[0].delta.content
        if content:
            print(content, end="", flush=True)

except Exception as exc:
    print(f"\n流式请求失败：{exc}")

在正式项目中，还可以把已经接收到的内容保存下来，方便用户重试或继续处理。

---

六、流式输出适合哪些场景？

1. 聊天页面

用户可以边看边读，不需要等待完整回答。

2. 命令行工具

终端中实时显示生成过程，反馈更及时。

3. 长文本生成

文章、报告或代码生成时间较长时，流式输出更有价值。

4. Agent 工作流

当某一步处理时间较长，可以把过程信息逐步展示出来。

---

七、使用流式输出时的注意事项

1. 前端需要持续接收数据

后端开启流式后，前端不能再按普通 JSON 一次性读取。

2. 需要处理连接中断

网络波动可能导致内容只返回一部分，要设计重试或提示机制。

3. 不要把片段当成完整结果

每个 chunk 只是部分内容，完整文本需要拼接。

4. 保存时要等待流结束

如果需要写文件或入库，应在循环结束后处理完整文本。

5. 先确认接口支持流式

不同模型和接口对流式响应的支持方式可能不同，接入前应查看实际文档。

---

八、结语

流式输出的核心并不复杂：

1. 请求时开启 stream=True
2. 循环读取返回片段
3. 过滤空内容
4. 及时刷新输出
5. 必要时拼接成完整文本

对于 Python AI 项目来说，流式输出可以明显改善长文本和聊天场景的交互体验。建议先用命令行完成最小验证，再接入自己的网页或工具界面。

免责声明

本文内容仅用于技术交流与经验分享，具体实现请结合项目实际情况调整。