Skip to main content
文本產生

流式輸出

在即時聊天或長文本產生應用中，長時間的等待會損害使用者體驗並可能導致觸發服務端逾時，導致任務失敗。流式輸出通過持續返回模型產生的文本片段，解決了這兩個核心問題。

工作原理

流式輸出基於 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的地區一致。
    # 以下為新加坡地區URL，調用時請將WorkspaceId替換為真實的業務空間ID，各地區的URL不同。
    base_url="https://{WorkspaceId}.ap-southeast-1.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：https://www.alibabacloud.com/help/zh/model-studio/get-api-key
    # 若沒有配置環境變數，請用百鍊API Key將下行替換為：api_key="sk-xxx"
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    # 以下為新加坡地區URL，調用時請將WorkspaceId替換為真實的業務空間ID，各地區的URL不同。
    base_url="https://{WorkspaceId}.ap-southeast-1.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://dashscope.oss-cn-beijing.aliyuncs.com/images/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=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://{WorkspaceId}.ap-southeast-1.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 模型無效
    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，其預設的輸出緩衝（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。
開始使用
Token Plan
模型體驗
接入用戶端/開發工具
模型推理
模型調優
模型部署
模型評測
用量統計與效能監控
資產中心
安全合規
實踐教程
服務支援
產品動態