跳转到主要内容
千问云 home page
开放平台文档
运行与扩展

流式输出

逐 token 实时接收模型生成的文本。

启用流式输出

  • OpenAI Chat Completions
  • OpenAI Responses API
  • DashScope
默认不返回用量信息。设置 stream_options 可在最后一个 chunk 中获取 token 用量。
import os
from openai import OpenAI
client = OpenAI(api_key=os.getenv("DASHSCOPE_API_KEY"), base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")

completion = client.chat.completions.create(
  model="qwen3.6-plus",
  messages=[{"role": "user", "content": "Hi"}],
  stream=True,                              # ← 启用流式输出
  stream_options={"include_usage": True},    # ← 仅在最后一个 chunk 返回用量
)
for chunk in completion:
  if chunk.choices:
    print(chunk.choices[0].delta.content or "", end="", flush=True)
  elif chunk.usage:                         # ← 最后一个 chunk：仅含用量信息
    print(f"\nTokens: {chunk.usage.total_tokens}")

事件格式

每个 SSE 事件是一行 data:，内容为 JSON 格式的 chunk。最后一条 data: [DONE] 表示流结束。 
data: {"choices":[{"delta":{"content":"I am"},...,"finish_reason":null}],...}
data: {"choices":[{"delta":{"content":" Qwen"},...,"finish_reason":null}],...}
data: {"choices":[{"delta":{"content":""},...,"finish_reason":"stop"}],...}
data: [DONE]

思考模式下的流式输出

两阶段流式输出：先输出思考过程，再输出回答。
  • OpenAI Chat Completions
  • OpenAI Responses API
  • DashScope
for chunk in completion:
  delta = chunk.choices[0].delta
  if hasattr(delta, "reasoning_content") and delta.reasoning_content:
    print(delta.reasoning_content, end="", flush=True)  # ← 阶段 1：思考
  if hasattr(delta, "content") and delta.content:
    print(delta.content, end="", flush=True)             # ← 阶段 2：回答
-> 完整配置：推理模式 | Qwen3-Omni：音视频

工具调用的流式输出

流式输出 function calling 响应时，工具调用的参数以增量 delta 形式到达，需要拼接完成后再解析 JSON。
  • Chat Completions
  • Responses API
每个 chunk 的 delta 可能包含 tool_calls[i].function.arguments——一个 JSON 片段字符串。按 tool call 索引累积所有片段，流结束后再 JSON.parse() 完整字符串。
tool_args = {}
for chunk in completion:
  delta = chunk.choices[0].delta
  if delta.tool_calls:
    for tc in delta.tool_calls:
      tool_args.setdefault(tc.index, "")
      tool_args[tc.index] += tc.function.arguments or ""
# 流结束后：
for idx, args_str in tool_args.items():
  parsed = json.loads(args_str)
启用思考模式后，流式输出分为三个阶段：思考 token、工具调用 delta，以及（发送工具返回结果后的）最终回答。

注意事项

  • Nginx 代理：设置 proxy_buffering off，否则 SSE 事件会被缓冲
  • 高并发场景：合理配置连接池大小，监控文件描述符数量
  • Web 前端：使用 ReadableStream + TextDecoderStream 处理流
  • 输出质量：流式输出不影响响应质量
  • 仅支持流式输出的模型：QwQ 和 QVQ 仅支持流式输出，非流式调用会失败或返回空内容。