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 を開き、同じプロンプトを 2 回 送ります。
- 1 回目: 「ストリーミング表示」に チェックを入れたまま 送信 → 最初の文字が数百 ms で出はじめる
- 2 回目: 「ストリーミング表示」の チェックを外して 送信 → 受信し終わるまで画面は空のまま
応答パネルの下に 最初の文字まで 0.85s (TTFT) ・ 受信 24 回 ・ 412 字 ・ 合計 6.1s のような実測値が出ます。
合計時間はほとんど変わらないのに、最初の文字までの時間だけが数倍違う — これが streaming の効き目です。
(「なめらか表示」は、届いたかたまりを 1 文字ずつ流し込んで描画するオプションです。OFF にすると
デルタが届いた瞬間にドサッと出るので、API が実際に返している単位が見えます。)
▶ ストリーミング体感 (Playground は SSE 対応)
Python の歴史を 500 字程度で教えてください。物語のように。