約 5 分で読めます

ClaudeストリーミングAPI設計——TTFT最適化と段階応答

by Kuu株式会社 エンジニアリング
ストリーミングClaude APITTFTeager_input_streaming
テーマ全体像：エージェントガバナンス完全ガイド →

エージェントを本番で動かすと、最初の応答が3秒以上かかるだけで利用者は「壊れた」と感じる。LLMの生成速度は変えられないが、ストリーミング設計で知覚レイテンシは劇的に下げられる。Anthropicが提供するSSE（Server-Sent Events）ベースのストリーミングAPIを正しく実装すれば、TTFT（Time to First Token）を300ms台に抑え、ツール使用中も部分的な引数を段階配信できる。

ストリーミングAPIとはどう動くか

Claude APIはSSEプロトコルでトークンを逐次配信する。stream: trueを設定すると、生成が完了する前からcontent_block_deltaイベントでトークンが流れ始め、体感レイテンシを大幅に短縮できる。

リクエストに "stream": true を渡すだけで、レスポンスがSSEストリームに切り替わる。イベントの種類は6つある。

イベント内容
message_startメッセージオブジェクトの初期化（使用トークン数など）
content_block_startコンテンツブロック開始（textまたはtool_use）
content_block_deltaテキスト差分（text_delta）またはツール引数差分（input_json_delta）
content_block_stopコンテンツブロック終了
message_deltaメッセージレベルの更新（stop_reason, usage など）
message_stopストリーム終了

Python SDKでは messages.stream() コンテキストマネージャが内部のSSE処理を隠蔽し、stream.text_stream でテキストのみを取得できる。

``python
with client.messages.stream(
model="claude-opus-4-8",
max_tokens=1024,
messages=[{"role": "user", "content": "Hello"}],
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
``

TypeScript SDKでは .on("text", handler) で同様のパターンが使える。低レベルのSSEイベントを直接扱いたい場合は stream_raw() / createStreaming() を使う。

TTFTはなぜエージェントUXを左右するか

TTFT（Time to First Token）は「リクエスト送信から最初のトークン受信まで」の時間で、ユーザーが体感する応答速度を決定する指標だ。200ms以下なら即座、1秒以上は遅延として知覚される。

TTFTは2つの要因に分解できる。

  1. ネットワーク往復時間: クライアントからAnthropicのエンドポイントまでの距離。東京リージョン経由なら50〜100ms程度。
  2. モデル処理時間: プロンプト長とモデルサイズに依存。Claude Haiku 4.5は最速クラス（公開ベンチマークで0.81秒前後）、Sonnet 4.6は1.14秒前後。

ノンストリーミングでは全トークンが生成完了するまでユーザーは何も見えない。1,000トークンの応答をOpus 4.8（約62 t/s）で生成すると16秒超かかる。ストリーミングにすれば最初のトークンは1秒以内に届き始め、体感速度が大幅に改善する。

エージェントが複数回ツールを呼ぶ場合、各ターンのTTFTが積み上がる。メインのオーケストレーターはSonnet 4.6、サブエージェントは高速なHaiku 4.5に分けると、全体TTFTを抑えつつ精度を確保できる（エージェント設計のClaudeモデル選択も参照）。

SSEイベント処理パターンはどう設計するか

ストリームハンドラは「蓄積」と「リアクション」を分離して設計する。SDKのアキュムレーターで最終メッセージを組み立て、リアクション層でUIに部分出力を即時反映する。

ストリーミング設計の基本原則は「蓄積」と「リアクション」の分離だ。

  • 蓄積: SDKのアキュムレーター（Python: stream.get_final_message()）に最終的な完全オブジェクトを組み立てさせる。自前で content_block_delta を連結するのは、SDKが既にそれを行うためDRY違反になる。
  • リアクション: on("text", handler) や stream.text_stream のイベントリスナーで、到着した差分をUIや下流処理へ即時反映する。

```python
with client.messages.stream(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": "Q3の売上サマリを作成して"}],
) as stream:
# リアクション: 到着したテキストを即時出力
for text in stream.text_stream:
print(text, end="", flush=True)

蓄積: 完全なメッセージオブジェクトを取得

final = stream.get_final_message() stop_reason = final.stop_reason # "end_turn" | "tool_use" | "max_tokens" ```

stop_reason の確認は必須だ。tool_use なら次のAPIコールでツール結果を返すターンに入り、max_tokens なら切断された可能性があるため、max_tokens を増やして再送するかエラーを返すかを判断する。

ツール使用のストリーミング——eager_input_streamingの設計

eager_input_streaming: true を設定すると、サーバー側のJSONバリデーション前にツール引数の断片が流れ始める。大きなドキュメント生成ツールでは最初の断片までのレイテンシを大幅に短縮できる。

通常のツール使用ストリーミングはサーバーがJSONを検証してから配信するため、引数が大きいと最初の断片が届くまで待たされる。eager_input_streaming を true にするとこの検証をスキップし、Claudeが生成した瞬間に断片が到着する。

``python
with client.messages.stream(
model="claude-opus-4-8",
max_tokens=4096,
tools=[{
"name": "generate_report",
"description": "長文レポートを生成する",
"eager_input_streaming": True,
"input_schema": {
"type": "object",
"properties": {
"content": {"type": "string"}
},
"required": ["content"]
}
}],
messages=[{"role": "user", "content": "Q3レポートを書いて"}],
) as stream:
for event in stream:
if event.type == "input_json":
# 断片が到着次第、プログレスバーを更新するなどのリアクションが可能
print(event.partial_json, end="", flush=True)
final = stream.get_final_message()
``

ツール引数の蓄積パターンは、コンテンツブロックインデックスをキーにした辞書で管理する。

  1. content_block_start（type=tool_use）: tool_inputs[index] = ""
  2. content_block_delta（type=input_json_delta）: tool_inputs[index] += event.delta.partial_json
  3. content_block_stop: json.loads(tool_inputs[index]) で解析（例外をキャッチ）

eager_input_streaming 使用時は、蓄積したJSONが不完全な場合がある。解析に失敗したら is_error: true とともに {"INVALID_JSON": "<生の文字列>"} をツール結果として返す。これによりClaudeが再試行の文脈を理解できる。

本番での3つの落とし穴と対策

マークダウンの途中レンダリング・ストリームの途中終了・接続タイムアウトの3点が本番で最もよく起きる問題だ。それぞれ実装レベルで確実に対処できる。

① マークダウンの途中レンダリング

text_delta が届くたびにマークダウンをHTMLに変換すると、壊れたHTMLが生成される。bold の最初の ** だけが届いた瞬間に変換しようとするためだ。対策は2つある。

  • テキストをプレーンテキストで表示し、message_stop 後にHTMLに変換する
  • 完結したブロック（段落・見出し）単位まで蓄積してから変換する

② ストリームの途中終了

with stream コンテキストを抜けると残りのチャンクが破棄され、get_final_message() が不完全なデータを返す。text_stream のループが途中で break した場合も同様だ。途中終了が必要なケースはタイムアウト処理と切り離し、CancelledError を適切にハンドルする。

③ 接続タイムアウト

長い生成中にリバースプロキシやロードバランサーが接続を切ることがある。Anthropic SDKは接続タイムアウトを timeout パラメータで設定できる（デフォルトは600秒）。ストリームが中断されたら stop_reason が end_turn でないことで検出し、会話履歴を引き継いで再開する。

規模別の留意点（SMB / エンタープライズ）

SMB向け: SDKのデフォルト設定（Python: anthropic.Anthropic(), TypeScript: new Anthropic()）で始められる。まずテキストストリーミングのみ実装し、ツール使用のストリーミングは後から追加するのが安全だ。KuuのAIエージェント運用管理サービスでは、ストリーミング対応の設計相談も受け付けている。

エンタープライズ向け: 複数チームが同一ストリームエンドポイントを利用する場合、LLMゲートウェイ側でSSEのプロキシが必要になる。ゲートウェイが保持する接続数の上限設計と、バックプレッシャー時のエラーハンドリングが重要だ。また、大規模な eager_input_streaming はJSON解析の失敗率をモニタリングし、閾値超過時にアラートを出す仕組みを組み込む。エンタープライズ規模のストリーミング基盤設計はRDEサービスで支援している。

参考

まとめ

Claude APIのストリーミング設計は、stream: true の一行から始まり、ツール使用での eager_input_streaming、本番でのタイムアウト・マークダウン・JSON蓄積の3点を押さえることで完結する。TTFTを300ms台に抑えることはモデル能力の問題ではなく実装設計の問題だ。

エージェントのストリーミング設計や本番運用の相談はKuu株式会社のAIエージェント運用管理サービスへ。エンタープライズ規模の基盤設計はRDEサービスで対応している。

関連記事

Claude APIで始める業務自動化——中小企業がビジネス活用を実現する3つの切り口Claude Sonnet 5エンタープライズ設計——ノード配置とコスト最適化computer useとマルチモーダルエージェントの設計判断エージェント設計のClaudeモデル選択——ツール使用性能比較
← ブログ一覧へ戻る