第3章 · Claude API Basics

Streaming で逐次表示する

Streaming Responses

→ で次のスライド · F でフルスクリーン · N で講師ノート · Esc で終了

重要キーワード

Streaming

ストリーミング

応答を逐次受け取る方式 (体感レイテンシ改善)

SSE

Server-Sent Events

HTTP 上のテキストストリーミング規格

TTFT

Time To First Token

最初のトークンが届くまでの時間

Backpressure

バックプレッシャ

受信側が処理しきれないときの流量制御

Streaming (ストリーミング)

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

基本

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---")

イベント単位で扱う

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 字程度で教えてください。物語のように。

Hands-on Exercise

演習: ストリーミング CLI

前のレッスンで作った CLI チャットを ストリーミング対応 に書き換えてください。 - 各文字が逐次表示される - Ctrl+C で途中停止できる (try/except KeyboardInterrupt)

▶ Playground を開いて実行

理解度チェック

4 問のクイズで理解度を確認しましょう。

クイズを開く

🎉

まとめ

お疲れ様でした!