Streaming Responses

Streaming で逐次表示する

約 12 分 · クイズ 4 問 · 演習 1 問

重要キーワード (4 語)

Streaming (ストリーミング) — 応答を逐次受け取る方式 (体感レイテンシ改善)

SSE (Server-Sent Events) — HTTP 上のテキストストリーミング規格

TTFT (Time To First Token) — 最初のトークンが届くまでの時間

Backpressure (バックプレッシャ) — 受信側が処理しきれないときの流量制御

Streaming (ストリーミング)

長い応答を 逐次表示 したいときは streaming を使います。チャット UI の体感レイテンシ (TTFT = Time To First Token) が大きく改善します。

基本

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Pythonの歴史を教えて"}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)

イベント単位で扱う

text_stream だけでなく、低レベルのイベントも取れます。

with client.messages.stream(...) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            print(event.delta.text, end="")
        elif event.type == "message_stop":
            print("\n---")

主なイベント種別

event.type	意味
`message_start`	応答開始
`content_block_start`	新ブロック開始
`content_block_delta`	テキスト追加
`content_block_stop`	ブロック終了
`message_delta`	部分メタデータ (stop_reason など)
`message_stop`	応答完了

Server-Sent Events (SSE)

REST API レベルでは SSE 形式でデルタが送られます。 Web アプリでブラウザに直接流すなら、サーバー側で受けて text/event-stream のレスポンスとして転送するのが定番。

Flask での例

@app.route("/stream", methods=["POST"])
def stream_chat():
    prompt = request.json["prompt"]
    def generate():
        with client.messages.stream(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}],
        ) as s:
            for text in s.text_stream:
                yield f"data: {text}\n\n"
        yield "data: [DONE]\n\n"
    return Response(generate(), mimetype="text/event-stream")

実は本講座のアプリ (/api/playground ルート) もこの方式で実装されています。ソースコード app.py の api_playground() を見てみましょう。

エラー処理

ストリーム中の例外は stream.__exit__ で再 raise されるので、通常の try/except で捉えられます。

try:
    with client.messages.stream(...) as s:
        for t in s.text_stream:
            ...
except RateLimitError:
    ...

完了後にメタデータを取る

ストリーミングでも最終的な Message オブジェクトを取得できます。

with client.messages.stream(...) as s:
    for t in s.text_stream:
        process(t)
    final = s.get_final_message()
    print("usage:", final.usage)

Streaming を体感する

長めの応答で違いを感じてみましょう。

▶ ストリーミング体感 (Playground は SSE 対応)

Python の歴史を 500 字程度で教えてください。物語のように。