Python 最小流式示例
先完成Python 环境配置。这里以 Chat Completions 为例；Responses 和其他协议的事件结构不同。
import os
from openai import OpenAI
with OpenAI(api_key=os.environ["QINGYUN_TOP_API_KEY"],
base_url=os.environ["QINGYUN_TOP_BASE_URL"],
timeout=60.0, max_retries=0) as client:
stream = client.chat.completions.create(
model=os.environ["QINGYUN_TOP_MODEL"],
messages=[{"role": "user", "content": "用三句话介绍流式输出"}],
stream=True,
)
finished = False
try:
for chunk in stream:
if not chunk.choices:
continue
choice = chunk.choices[0]
print(choice.delta.content or "", end="", flush=True)
if choice.finish_reason is not None:
finished = True
finally:
stream.close()
print("\n已收到完成原因" if finished else "\n未确认完成，请检查连接和日志")业务接入要处理什么
- 首字到来前显示等待状态，设置连接、读取与业务总超时。
- 按协议解析事件。不要假设一次网络分块就是一个完整 JSON，也不要把所有 SSE 事件都当正文。
- 区分文本增量、工具参数增量、完成原因和错误。用量字段可能只在特定配置下返回，不能保证每块都存在。
- 用户取消时停止读取并关闭上游连接，避免无意义生成；停止后是否已消耗额度看日志。
- 断开后保留已接收内容并显示“未完成”，不要自动从头重复一个可能已被执行的任务。
后端收到分块，浏览器却最后一起显示
检查你的代理、应用服务器和前端读取是否缓冲整份响应，核对 SSE 的 Content-Type、压缩及超时配置。这里只调整你自己的应用链路，不需要在客户端反复换密钥。