Skip to main content
阿里云百炼 home page阿里云百炼
文本生成

流式输出

在实时聊天或长文本生成应用中，长时间的等待会损害用户体验并可能导致触发服务端超时，导致任务失败。流式输出通过持续返回模型生成的文本片段，解决了这两个核心问题。

工作原理

流式输出基于 Server-Sent Events (SSE) 协议。发起流式请求后，服务端与客户端建立持久化 HTTP 连接。模型每生成一个文本块（称为 chunk），立即通过连接推送。全部内容生成后，服务端发送结束信号。 客户端监听事件流，实时接收并处理文本块，例如逐字渲染界面。这与非流式调用（一次性返回所有内容）形成对比。
以上组件仅供您参考，并未真实发送请求。

计费说明

流式输出计费规则与非流式调用完全相同，根据请求的输入Token数和输出Token数计费。 请求中断时，输出 Token 仅计算服务端收到终止请求前已生成的部分。

如何使用

Qwen3 开源版、QwQ 商业版与开源版、QVQ 、Qwen-Omni等模型仅支持流式输出方式调用。

步骤一：配置 API Key 并选择地域

您需要已获取与配置 API Key。
将API Key配置为环境变量（DASHSCOPE_API_KEY）比在代码中硬编码更安全。

步骤二：发起流式请求

  • OpenAI兼容
  • DashScope
  • 如何开启 设置 stream 为 true 即可。
  • 查看 Token 消耗 OpenAI 协议默认不返回 Token 消耗量，需设置stream_options={"include_usage": true}，使最后一个返回的数据块包含Token消耗信息。
  • Python
  • Node.js
  • curl
import os
from openai import OpenAI

# 1. 准备工作：初始化客户端
client = OpenAI(
    # 建议通过环境变量配置API Key，避免硬编码。
    api_key=os.environ["DASHSCOPE_API_KEY"],
    # API Key与地域强绑定，请确保base_url与API Key的地域一致。
    base_url="https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)

# 2. 发起流式请求
completion = client.chat.completions.create(
    model="qwen-plus",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "请介绍一下自己"}
    ],
    stream=True,
    stream_options={"include_usage": True}
)

# 3. 处理流式响应
# 用列表暂存响应片段，最后 join 比逐次 += 字符串更高效
content_parts = []
print("AI: ", end="", flush=True)

for chunk in completion:
    if chunk.choices:
        content = chunk.choices[0].delta.content or ""
        print(content, end="", flush=True)
        content_parts.append(content)
    elif chunk.usage:
        print("\n--- 请求用量 ---")
        print(f"输入 Tokens: {chunk.usage.prompt_tokens}")
        print(f"输出 Tokens: {chunk.usage.completion_tokens}")
        print(f"总计 Tokens: {chunk.usage.total_tokens}")

full_response = "".join(content_parts)
# print(f"\n--- 完整回复 ---\n{full_response}")

返回结果 

AI: 你好！我是Qwen，是阿里巴巴集团旗下的通义实验室自主研发的超大规模语言模型。我能够回答问题、创作文字，比如写故事、写公文、写邮件、写剧本、逻辑推理、编程等等，还能表达观点，玩游戏等。我支持多种语言，包括但不限于中文、英文、德语、法语、西班牙语等。如果你有任何问题或需要帮助，欢迎随时告诉我！
--- 请求用量 ---
输入 Tokens: 26
输出 Tokens: 87
总计 Tokens: 113

多模态模型的流式输出

多模态模型支持在对话中加入图片、音频等内容，其流式输出的实现方式与文本模型主要有以下不同：
  • 用户消息（user message）的构造方式：多模态模型的输入不仅包括文本，还包含图片、音频等多模态信息。
  • DashScope SDK接口：使用 DashScope Python SDK 时，需调用 MultiModalConversation 接口；使用DashScope Java SDK 时，则调用 MultiModalConversation 类。
多模态模型请参见：图像与视频理解、文字提取、界面交互、音频理解-Qwen3-Omni-Captioner、Kimi等，Qwen-Omni 模型仅支持流式输出，因其输出可包含文本或音频等多模态内容，结果解析方式与其他模型不同，具体请参见全模态。
  • OpenAI兼容
  • DashScope
Python
from openai import OpenAI
import os

client = OpenAI(
    # 若没有配置环境变量，请用百炼API Key将下行替换为：api_key="sk-xxx"
    # 各地域的API Key不同。获取API Key：https://help.aliyun.com/zh/model-studio/get-api-key
    api_key=os.getenv("DASHSCOPE_API_KEY"),

    # 以下为华北2（北京）地域的URL，调用时请将WorkspaceId替换为真实的业务空间ID，各地域的URL不同。
    base_url="https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)

completion = client.chat.completions.create(
    model="qwen3-vl-plus",  # 可按需更换为其它多模态模型，并修改相应的 messages
    messages=[
        {"role": "user",
         "content": [{"type": "image_url",
                    "image_url": {"url": "https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20241022/emyrja/dog_and_girl.jpeg"},},
                    {"type": "text", "text": "图中描绘的是什么景象？"}]}],
    stream=True,
  # stream_options={"include_usage": True}
)
full_content = ""
print("流式输出内容为：")
for chunk in completion:
    # 如果stream_options.include_usage为True，则最后一个chunk的choices字段为空列表，需要跳过（可以通过chunk.usage获取 Token 使用量）
    if chunk.choices and chunk.choices[0].delta.content != "":
        full_content += chunk.choices[0].delta.content
        print(chunk.choices[0].delta.content)
print(f"完整内容为：{full_content}")

思考模型的流式输出

思考模型会先返回reasoning_content（思考过程），再返回content（回复内容）。可根据数据包状态判断当前为思考或是回复阶段。
思考模型详情参见：深度思考、图像与视频理解、视觉推理。
Qwen3-Omni-Flash（思考模式）实现流式输出请参见全模态。
  • OpenAI兼容
  • DashScope
以下是使用 OpenAI Python SDK 以流式方式调用思考模式 qwen-plus 模型时返回的数据格式：
# 思考阶段
...
ChoiceDelta(content=None, function_call=None, refusal=None, role=None, tool_calls=None, reasoning_content='覆盖所有要点，同时')
ChoiceDelta(content=None, function_call=None, refusal=None, role=None, tool_calls=None, reasoning_content='自然流畅。')
# 回复阶段
ChoiceDelta(content='你好！我是**通', function_call=None, refusal=None, role=None, tool_calls=None, reasoning_content=None)
ChoiceDelta(content='义千问**（', function_call=None, refusal=None, role=None, tool_calls=None, reasoning_content=None)
...
  • 若reasoning_content不为 None，content 为 None，则当前处于思考阶段；
  • 若reasoning_content为 None，content 不为 None，则当前处于回复阶段；
  • 若两者均为 None，则阶段与前一包一致。
  • Python
  • Node.js
  • HTTP

示例代码 

from openai import OpenAI
import os

# 初始化OpenAI客户端
client = OpenAI(
    # 如果没有配置环境变量，请用阿里云百炼API Key替换：api_key="sk-xxx"
    # 各地域的API Key不同。获取API Key：https://help.aliyun.com/zh/model-studio/get-api-key
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # 以下为华北2（北京）地域的URL，调用时请将WorkspaceId替换为真实的业务空间ID，各地域的URL不同。
    base_url="https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)

messages = [{"role": "user", "content": "你是谁"}]

completion = client.chat.completions.create(
    model="qwen-plus",  # 您可以按需更换为其它深度思考模型
    messages=messages,
    # enable_thinking 参数开启思考过程，qwen3-30b-a3b-thinking-2507、qwen3-235b-a22b-thinking-2507、QwQ 与 DeepSeek-R1 模型总会进行思考，不支持该参数
    extra_body={"enable_thinking": True},
    stream=True,
    # stream_options={
    #     "include_usage": True
    # },
)

reasoning_content = ""  # 完整思考过程
answer_content = ""  # 完整回复
is_answering = False  # 是否进入回复阶段
print("\n" + "=" * 20 + "思考过程" + "=" * 20 + "\n")

for chunk in completion:
    if not chunk.choices:
        print("\nUsage:")
        print(chunk.usage)
        continue

    delta = chunk.choices[0].delta

    # 只收集思考内容
    if hasattr(delta, "reasoning_content") and delta.reasoning_content is not None:
        if not is_answering:
            print(delta.reasoning_content, end="", flush=True)
        reasoning_content += delta.reasoning_content

    # 收到content，开始进行回复
    if hasattr(delta, "content") and delta.content:
        if not is_answering:
            print("\n" + "=" * 20 + "完整回复" + "=" * 20 + "\n")
            is_answering = True
        print(delta.content, end="", flush=True)
        answer_content += delta.content

返回结果 

====================思考过程====================

好的，用户问“你是谁”，我需要给出一个准确且友好的回答。首先，我要确认自己的身份，即通义千问，由阿里巴巴集团旗下的通义实验室研发。接下来，应该说明我的主要功能，比如回答问题、创作文字、逻辑推理等。同时，要保持语气亲切，避免过于技术化，让用户感觉轻松。还要注意不要使用复杂术语，确保回答简洁明了。另外，可能需要加入一些互动元素，邀请用户提问，促进进一步交流。最后，检查是否有遗漏的重要信息，比如我的中文名称“通义千问”和英文名称“Qwen”，以及所属公司和实验室。确保回答全面且符合用户期望。
====================完整回复====================

你好！我是通义千问，是阿里巴巴集团旗下的通义实验室自主研发的超大规模语言模型。我可以回答问题、创作文字、进行逻辑推理、编程等，旨在为用户提供高质量的信息和服务。你可以叫我Qwen，或者直接叫我通义千问。有什么我可以帮你的吗？

应用于生产环境

  • 性能与资源管理：在后端服务中，为每个流式请求维持一个HTTP长连接会消耗资源。确保您的服务配置了合理的连接池大小和超时时间。在高并发场景下，监控服务的文件描述符（file descriptors）使用情况，防止耗尽。
  • 客户端渲染：在Web前端，使用 ReadableStream 和 TextDecoderStream API 可以平滑地处理和渲染SSE事件流，提供最佳的用户体验。
  • 模型监控：
    • 关键指标：监控首Token延迟（Time to First Token, TTFT），该指标是衡量流式体验的核心。同时监控请求错误率和平均响应时长。
    • 告警设置：为API错误率（特别是4xx和5xx错误）的异常设置告警。
  • Nginx代理配置：若使用 Nginx 作为反向代理，其默认的输出缓冲（proxy_buffering）会破坏流式响应的实时性。为确保数据能被即时推送到客户端，务必在Nginx配置文件中设置proxy_buffering off以关闭此功能。

错误码

如果模型调用失败并返回报错信息，请参见错误码进行解决。

常见问题

Q：为什么返回数据中没有 usage 信息？

A：OpenAI 协议默认不返回 usage 信息，设置stream_options参数使得最后返回的包中包含 usage 信息。

Q：开启流式输出对模型的回复效果是否有影响？

A：无影响，但部分模型仅支持流式输出，且非流式输出可能引发超时错误。建议优先使用流式输出。

Q：非流式调用和流式调用有什么区别？

A：主要区别如下：
  • 超时限制：非流式调用的最大超时时间不少于300秒，实际时长因部署区域与选用模型存在差异。
  • 输出结构：非流式调用一次性返回完整的响应结果（单个JSON对象）。流式调用通过SSE协议逐步返回数据块（chunk），每个chunk包含部分生成内容，需要客户端拼接。
  • 功能兼容：两者均支持JSON Mode、Function Call等功能特性，功能上没有差异。
建议优先使用流式输出，可以避免超时问题并获得更好的用户体验。

Q：流式输出是否支持JSON Mode（结构化输出）？

A：支持。在请求中同时设置stream为true和response_format为{"type": "json_object"}即可。模型会以流式方式逐步返回JSON格式的内容片段，最终拼接后的完整输出为合法的JSON。
阿里云百炼 home page阿里云百炼
开始使用
产品计费
Token Plan
模型体验
接入客户端/开发工具
模型推理
模型调优
模型部署
模型评测
模型压缩目录节点
用量统计与性能监控
资产中心
模型数据
安全合规
实践教程
服务支持
产品动态