← レッスンに戻る
第3章 · Claude API入門

Streaming で逐次表示する

Streaming Responses · 約 12 分

重要キーワード

English日本語説明
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.pyapi_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 を開き、同じプロンプトを 2 回 送ります。

  1. 1 回目: 「ストリーミング表示」に チェックを入れたまま 送信 → 最初の文字が数百 ms で出はじめる
  2. 2 回目: 「ストリーミング表示」の チェックを外して 送信 → 受信し終わるまで画面は空のまま

応答パネルの下に 最初の文字まで 0.85s (TTFT) ・ 受信 24 回 ・ 412 字 ・ 合計 6.1s のような実測値が出ます。 合計時間はほとんど変わらないのに、最初の文字までの時間だけが数倍違う — これが streaming の効き目です。 (「なめらか表示」は、届いたかたまりを 1 文字ずつ流し込んで描画するオプションです。OFF にすると デルタが届いた瞬間にドサッと出るので、API が実際に返している単位が見えます。)

▶ ストリーミング体感 (Playground は SSE 対応)
Python の歴史を 500 字程度で教えてください。物語のように。

演習問題

演習 1: ストリーミング CLI

ここでいう 前のレッスンとは ch3-l2「Messages API の基本」 の演習 です。 見当たらない人のために解答コードを再掲します。これを ストリーミング対応 に書き換えてください。

# 書き換える前 (ch3-l2 の解答): 生成が終わるまで 1 文字も表示されない
from anthropic import Anthropic

client = Anthropic()
history = []

while True:
    u = input('> ').strip()
    if u.lower() == 'quit': break
    history.append({'role': 'user', 'content': u})
    msg = client.messages.create(
        model='claude-haiku-4-5',
        max_tokens=512,
        messages=history,
    )
    text = msg.content[0].text
    history.append({'role': 'assistant', 'content': text})
    print(text)
    print(f'  [tokens] in={msg.usage.input_tokens} out={msg.usage.output_tokens}')

要件

  1. messages.create(...)messages.stream(...) に変え、届いたテキストを その場で 出す (print(t, end="", flush=True))
  2. Ctrl+C で生成だけを打ち切れる (try / except KeyboardInterrupt)。打ち切ってもプログラムは終了せず、次の入力に戻る
  3. 会話履歴の保持と usage 表示は元のまま維持する (usage は stream.get_final_message() から取る)

確認ポイント

  • 最初の文字が出るまでの時間 (TTFT) が create 版より明らかに短い
  • Ctrl+C した直後、「途中まで」の応答が手元に残る → それを履歴に入れるかは自分で決める

⚠️ 「各文字が逐次表示される」の実際: text_stream が返すのは 1 文字ずつではなく 数文字〜十数文字のかたまり (delta) です。文字単位で流したいときは delta をさらに 1 文字ずつに分解して time.sleep(0.02) を挟みます (サンプル解答の「発展」を参照)。

スタータープロンプト:
次の Python CLI チャットを Streaming 対応に書き換えてください。

```python
from anthropic import Anthropic
client = Anthropic()
history = []
while True:
    u = input('> ').strip()
    if u.lower() == 'quit': break
    history.append({'role': 'user', 'content': u})
    msg = client.messages.create(model='claude-haiku-4-5', max_tokens=512, messages=history)
    text = msg.content[0].text
    history.append({'role': 'assistant', 'content': text})
    print(text)
```

条件: client.messages.stream(...) を with で使い text_stream を print(end='', flush=True) で逐次表示 / Ctrl+C で生成だけ中断して次の入力に戻る (try/except KeyboardInterrupt) / 中断時に途中まで受け取ったテキストの扱いも書く / usage は stream.get_final_message() から表示。コードと短い解説のみ。
ヒントを見る
  • 形は with client.messages.stream(...) as s: の中で for t in s.text_stream: です。
  • print(t, end="", flush=True)flush=True を忘れると改行まで画面に出ません (行バッファリング)。
  • try/except KeyboardInterruptwhile ループの外ではなく with ブロックを囲む位置 に置きます。外側に置くと Ctrl+C でチャットごと終了してしまい「生成だけ止める」になりません。
  • 中断で 1 文字も受け取れていないのに assistant を履歴に足すと、次のリクエストで content が空になり API エラーになります。空なら直前の user 発言ごと取り消すのが安全です。
サンプル解答を見る
# chat_stream.py — ch3-l2 の CLI チャットを streaming 化したもの
from anthropic import Anthropic

client = Anthropic()
history = []

print("quit で終了 / 生成中の Ctrl+C はその応答だけ中断")
while True:
    try:
        u = input("\n> ").strip()
    except (EOFError, KeyboardInterrupt):    # 入力待ちの Ctrl+C は終了
        print()
        break
    if not u:
        continue
    if u.lower() in ("quit", "exit"):
        break

    history.append({"role": "user", "content": u})
    parts, interrupted = [], False
    try:
        with client.messages.stream(
            model="claude-haiku-4-5",
            max_tokens=512,
            messages=history,
        ) as stream:
            for t in stream.text_stream:     # 届いた分だけ即表示
                parts.append(t)
                print(t, end="", flush=True)
            usage = stream.get_final_message().usage
            print(f"\n  [tokens] in={usage.input_tokens} out={usage.output_tokens}")
    except KeyboardInterrupt:                # 生成中の Ctrl+C = この応答だけ中断
        interrupted = True
        print("\n  [中断] Ctrl+C で停止しました")

    text = "".join(parts)
    if text:
        history.append({"role": "assistant",
                        "content": text + ("…(中断)" if interrupted else "")})
    else:
        history.pop()   # 1 文字も来ていなければ user 発言ごと取り消す (空 content は API エラー)

発展: 本当に 1 文字ずつ流したいとき

import time
for t in stream.text_stream:
    for ch in t:                 # delta をさらに 1 文字に割る
        print(ch, end="", flush=True)
        time.sleep(0.02)

見た目はタイプライターになりますが、表示は必ず受信より遅れます。Playground の「なめらか表示」も同じ考え方で、受信済みの文字だけを一定速度で描いています。

理解度チェック

  1. ストリーミングを使う主な利点は?
    1. 出力品質が上がる
    2. 体感レイテンシの改善
    3. API コストが半額になる
    4. max_tokens 制限がなくなる
  2. Web アプリでブラウザにリアルタイム転送する代表的な方式は?
    1. FTP
    2. Server-Sent Events (SSE)
    3. SMTP
    4. telnet
  3. Python SDK で streaming を使うには?
    1. messages.create(stream=True)
    2. messages.stream(...) を with 文で
    3. messages.async_create(...)
    4. websocket.connect(...)
  4. TTFT (Time To First Token) は何を表す?
    1. 全応答完了までの時間
    2. 最初のトークンが届くまでの時間
    3. API キー有効期限
    4. DB への接続時間
解答と解説を見る
  1. B — 最初のトークンが早く届くので UX が大幅に改善します。
  2. B — SSE はテキストストリーミングに適した HTTP 上のプロトコルです。
  3. B — `messages.stream(...)` を context manager で使うのが推奨方法です。
  4. B — 最初のトークンが出るまでの待機時間で、ストリーミング UX の鍵指標です。