Skip to main content

ストリーミング出力

リアルタイムチャットや長文生成アプリケーションでは、長い待ち時間がユーザーエクスペリエンスを低下させ、サーバー側のタイムアウトを引き起こしてタスクが失敗する可能性があります。ストリーミング出力は、モデルがテキストを生成する際に断片を継続的に返すことで、これらの問題に対処します。

仕組み

ストリーミング出力は、Server-Sent Events (SSE) プロトコルを使用します。ストリーミングリクエストが開始されると、サーバーはクライアントとの HTTP 持続的接続を確立します。モデルがテキストブロック（チャンクと呼ばれます）を生成するたびに、そのチャンクをこの接続を通じて即座にプッシュします。すべてのコンテンツが生成されると、サーバーは終了シグナルを送信します。 クライアントはイベントストリームをリッスンし、テキストチャンクをリアルタイムで受信・処理します。たとえば、インターフェイス上で文字を 1 文字ずつレンダリングします。これは、すべてのコンテンツを一度に返す非ストリーミング呼び出しとは対照的です。

課金

ストリーミング出力は、非ストリーミング呼び出しと同じ課金ルールを使用し、リクエスト内の入力トークン数と出力トークン数に基づいて課金されます。 リクエストが中断された場合、出力トークンは、サーバーが終了リクエストを受信する前に生成された部分のみがカウントされます。

使用方法

Qwen3 オープンソース版、QwQ コマーシャルおよびオープンソース版、QVQ、および Qwen-Omni は、ストリーミング出力のみをサポートしています。

ステップ 1：API キーを構成し、リージョンを選択する

API キーを取得済みで、かつ環境変数として構成済みである必要があります。
コード内にハードコーディングするよりも、API キーを環境変数 (DASHSCOPE_API_KEY) として構成する方が安全です。

ステップ 2：ストリーミングリクエストを行う

  • OpenAI 互換
  • DashScope
  • 有効化方法 stream を true に設定します。
  • トークン使用量の確認 OpenAI プロトコルはデフォルトでトークン使用量を返しません。stream_options={"include_usage": true} を設定すると、最後に返されるデータチャンクにトークン使用量情報が含まれます。
  • Python
  • Node.js
  • curl
import os
from openai import OpenAI

# 1. 準備：クライアントを初期化
client = OpenAI(
    # ハードコーディングを避けるため、環境変数を使用して API キーを構成。
    api_key=os.environ["DASHSCOPE_API_KEY"],
    # API キーはリージョンに厳密にバインドされています。base_url が API キーのリージョンと一致することを確認してください。
    # シンガポールリージョンの 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": "Please introduce yourself"}
    ],
    stream=True,
    stream_options={"include_usage": True}
)

# 3. ストリーミングレスポンスを処理
# レスポンスフラグメントをリストに格納。繰り返し文字列連結を行うよりも、最後に結合する方が効率的です。
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--- Request usage ---")
        print(f"Input Tokens: {chunk.usage.prompt_tokens}")
        print(f"Output Tokens: {chunk.usage.completion_tokens}")
        print(f"Total Tokens: {chunk.usage.total_tokens}")

full_response = "".join(content_parts)
# print(f"\n--- Full response ---\n{full_response}")

レスポンス 

AI: Hello! I am Qwen, a large-scale language model independently developed by Tongyi Lab under Alibaba Group. I can answer questions, create content such as stories, official documents, emails, scripts, perform logical reasoning, programming, express opinions, play games, and more. I support multiple languages, including but not limited to Chinese, English, German, French, and Spanish. If you have any questions or need help, feel free to ask me anytime!
--- Request usage ---
Input Tokens: 26
Output Tokens: 87
Total Tokens: 113

マルチモーダルモデルのストリーミング出力

マルチモーダルモデルは、会話に画像や音声などのコンテンツを追加できます。そのストリーミング出力の実装は、テキスト専用モデルとは以下の点で異なります。
  • ユーザーのメッセージ構築：マルチモーダルモデルの入力には、テキストだけでなく、画像や音声などのマルチモーダル情報も含まれます。
  • 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 キーはリージョンによって異なります。API キーの取得先： https://www.alibabacloud.com/help/zh/model-studio/get-api-key
    # 環境変数を構成していない場合は、次の行を Model Studio API キーに置き換えてください： 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=[
        {"role": "user",
        "content": [{"type": "image_url",
                    "image_url": {"url": "https://dashscope.oss-cn-beijing.aliyuncs.com/images/dog_and_girl.jpeg"},},
                    {"type": "text", "text": "What scene is depicted in the image?"}]}],
    stream=True,
  # stream_options={"include_usage": True}
)
full_content = ""
print("Streaming output content:")
for chunk in completion:
    # stream_options.include_usage が True の場合、最後のチャンクの choices フィールドは空のリストになるためスキップします（トークン使用量は chunk.usage から取得できます）
    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: {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='Cover all key points while')
ChoiceDelta(content=None, function_call=None, refusal=None, role=None, tool_calls=None, reasoning_content='remaining natural and fluent.')
# 応答段階
ChoiceDelta(content='Hello! I am **Qwen', 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(
    # 環境変数を構成していない場合は、Alibaba Cloud Model Studio API キーに置き換えてください： 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": "Who are you"}]

completion = client.chat.completions.create(
    model="qwen-plus",  # 必要に応じて他のディープシンキングモデルに置き換えてください
    messages=messages,
    # enable_thinking パラメーターにより思考プロセスを有効化。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

    # コンテンツを受信し、応答を開始
    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」という英語名、および所属企業と研究室といった重要な情報が欠けていないか確認し、ユーザーの期待に応える包括的な回答となるようにします。
====================完全な回答====================

こんにちは！私はアリババグループ傘下の通義実験室が独自に開発した大規模言語モデル Qwen です。質問への回答、テキスト作成、論理的推論、プログラミングなどを行うことができ、高品質な情報とサービスを提供することを目指しています。私を Qwen または単に通義千問と呼んでください。何かお手伝いできることはありますか？

公開

  • パフォーマンスとリソース管理：バックエンドサービスでは、各ストリーミングリクエストに対して HTTP 持続的接続を維持することでリソースを消費します。適切な接続プールサイズとタイムアウト値を設定してください。高い同時実行性のシナリオでは、ファイルディスクリプタの使用量を監視し、枯渇を防いでください。
  • クライアント側のレンダリング：ウェブフロントエンドでは、ReadableStream および TextDecoderStream API を使用して、SSE イベントストリームをスムーズに処理・レンダリングし、最高のユーザーエクスペリエンスを提供してください。
  • モデルモニタリング：
    • 主要メトリック：ストリーミングエクスペリエンスの中核となるメトリックである 初回トークン到達時間 (TTFT) を監視してください。また、API エラー率および平均応答時間も監視します。
    • アラート：特に 4xx および 5xx エラーなど、異常な API エラー率に対してアラートを設定してください。
  • Nginx プロキシ構成：Nginx をリバースプロキシとして使用する場合、デフォルトの出力バッファリング (proxy_buffering) によりストリーミングレスポンスのリアルタイム性が損なわれます。データがクライアントに即座にプッシュされるようにするには、Nginx 構成ファイルで proxy_buffering off を設定してこの機能を無効にしてください。

エラーコード

モデル呼び出しが失敗してエラーメッセージが返された場合は、「エラーコード」をご参照のうえ、解決策を確認してください。

よくある質問

Q：レスポンスに使用量情報が含まれないのはなぜですか？

A：OpenAI プロトコルはデフォルトで使用量情報を返しません。最終的に返されるパケットに使用量情報を含めるには、stream_options パラメーターを設定してください。

Q：ストリーミング出力を有効にしても、モデルの応答品質に影響しますか？

A：いいえ。ただし、一部のモデルはストリーミング出力のみをサポートしており、非ストリーミング呼び出しではタイムアウトエラーが発生する可能性があります。ストリーミング出力の使用を推奨します。

Q：非ストリーミング呼び出しとストリーミング呼び出しの違いは何ですか？

A：主な違いは以下のとおりです。
  • タイムアウト制限：非ストリーミング呼び出しには 300 秒の固定最大タイムアウトがあります。モデルが 300 秒以内に生成を完了しない場合、リクエストはタイムアウトして失敗します。
  • 出力構造：非ストリーミング呼び出しは、一度に完全なレスポンス（単一の JSON オブジェクト）を返します。ストリーミング呼び出しは、SSE プロトコルを介してデータチャンクを段階的に返し、各チャンクには生成されたコンテンツの一部が含まれます。クライアントはこれらのチャンクをアセンブルする必要があります。
  • 機能互換性：JSON モードや関数呼び出しなどの機能は、どちらもサポートされており、機能的な違いはありません。
タイムアウトを回避し、ユーザーエクスペリエンスを向上させるために、ストリーミング出力の使用を推奨します。

Q：ストリーミング出力は JSON モード（構造化出力）をサポートしていますか？

A：はい。リクエストで stream を true に、response_format を {"type": "json_object"} に設定してください。モデルは JSON 形式のコンテンツフラグメントを段階的に返します。最終的にアセンブルされた出力は、有効な JSON になります。