Streaming で逐次表示する
重要キーワード
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 を開き、同じプロンプトを 2 回 送ります。
- 1 回目: 「ストリーミング表示」に チェックを入れたまま 送信 → 最初の文字が数百 ms で出はじめる
- 2 回目: 「ストリーミング表示」の チェックを外して 送信 → 受信し終わるまで画面は空のまま
応答パネルの下に 最初の文字まで 0.85s (TTFT) ・ 受信 24 回 ・ 412 字 ・ 合計 6.1s のような実測値が出ます。
合計時間はほとんど変わらないのに、最初の文字までの時間だけが数倍違う — これが streaming の効き目です。
(「なめらか表示」は、届いたかたまりを 1 文字ずつ流し込んで描画するオプションです。OFF にすると
デルタが届いた瞬間にドサッと出るので、API が実際に返している単位が見えます。)
Python の歴史を 500 字程度で教えてください。物語のように。演習: ストリーミング 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}')
要件
messages.create(...)をmessages.stream(...)に変え、届いたテキストを その場で 出す (print(t, end="", flush=True))- Ctrl+C で生成だけを打ち切れる (
try/except KeyboardInterrupt)。打ち切ってもプログラムは終了せず、次の入力に戻る - 会話履歴の保持と usage 表示は元のまま維持する (usage は
stream.get_final_message()から取る)
確認ポイント
- 最初の文字が出るまでの時間 (TTFT) が create 版より明らかに短い
- Ctrl+C した直後、「途中まで」の応答が手元に残る → それを履歴に入れるかは自分で決める
⚠️ 「各文字が逐次表示される」の実際:
text_streamが返すのは 1 文字ずつではなく 数文字〜十数文字のかたまり (delta) です。文字単位で流したいときは delta をさらに 1 文字ずつに分解してtime.sleep(0.02)を挟みます (サンプル解答の「発展」を参照)。
まとめ
お疲れ様でした!