⬡ API & SDK/2026-03-14上級

Claude API ストリーミング・ツール使用本番運用 — 並列呼び出し、エラー処理、リトライ戦略

Claude API の本番グレードストリーミングとツール使用パターンをマスターします。並列ツール呼び出し、インテリジェントエラー処理、耐久性の高いリトライ戦略を習得。

Claude API¹¹⁵ ストリーミング⁹ ツール使用² 本番パターンエラー処理

✦ プレミアム記事

プロトタイプと本番システムを分けるもの

Claude API を使用して信頼性の高い本番グレードのシステムを構築するには、ストリーミングレスポンスとツール使用をマスターする必要があります。複数の順序的および並列的なツール呼び出しに依存するエージェントワークフローを構築する場合は特にそうです。プロトタイプと本番システムの違いは、思慮深いエラー処理、インテリジェントなリトライ戦略、リソース最適化にあります。

このガイドでは、数百から数百万の API 呼び出しをスケーリングするためのストリーミング、並列ツール呼び出し、洗練されたエラー処理、および運用上の耐久性に関する本番テスト済みパターンについて説明します。

💡

Claude API に関する最も一般的な本番障害は Claude のコアパフォーマンスには起因せず、不適切なエラー処理、不十分なリトライ戦略、非効率的なリソース管理に起因します。このガイドは、プロトタイプと防弾本番システムの違いを生み出す運用パターンに焦点を当てています。

本番向けストリーミング基礎

ストリームライフサイクルの理解

本番でのストリーミングには、単なるハッピーパスではなく、ストリーム全体のライフサイクルを理解する必要があります:

import anthropic
import json
from typing import Generator
 
client = anthropic.Anthropic()
 
def stream_with_lifecycle_tracking():
    """
    以下を含むストリーミング ライフサイクル全体:
    - 接続確立
    - トークン ストリーミング
    - エラー状態
    - クリーンアップ
    """
    stream_metadata = {
        "tokens_received": 0,
        "errors": [],
        "time_to_first_token": None,
        "total_duration": None
    }
 
    try:
        with client.messages.stream(
            model="claude-opus-4-6",
            max_tokens=1024,
            messages=[
                {
                    "role": "user",
                    "content": "次のデータを分析し、洞察を提供します"
                }
            ]
        ) as stream:
            # 最初のトークンまでの時間を追跡 (UX に重要)
            first_token_received = False
 
            for text in stream.text_stream:
                if not first_token_received:
                    stream_metadata["time_to_first_token"] = stream.get_final_message().id
                    first_token_received = True
 
                stream_metadata["tokens_received"] += len(text.split())
                # トークンが到着したらすぐに処理/生成
                yield text
 
            # 最終メッセージ メタデータをキャプチャ
            final_message = stream.get_final_message()
            stream_metadata["usage"] = {
                "input_tokens": final_message.usage.input_tokens,
                "output_tokens": final_message.usage.output_tokens
            }
 
    except anthropic.APIError as e:
        stream_metadata["errors"].append({
            "type": str(type(e).__name__),
            "message": str(e)
        })
        raise
    finally:
        # 常にストリーム メタデータをログして監視
        log_stream_metrics(stream_metadata)

リアルタイムシステムのバッファリング戦略

異なるシステムは異なるバッファリングアプローチを必要とします:

import asyncio
from collections import deque
from typing import AsyncGenerator
 
class StreamBuffer:
    """ストリーミング レスポンスのインテリジェント バッファリング。"""
 
    def __init__(self, buffer_size: int = 5, flush_timeout: float = 0.1):
        self.buffer = deque(maxlen=buffer_size)
        self.flush_timeout = flush_timeout
 
    async def stream_with_buffer(
        self,
        messages: list
    ) -> AsyncGenerator[str, None]:
        """
        適応的なバッファリング付きストリーム:
        - 低遅延アプリケーション用の小さいバッファ
        - スループット最適化用の大きいバッファ
        - 一貫性のためのタイムベース フラッシング
        """
        client = anthropic.AsyncAnthropic()
        accumulated_text = ""
        last_flush = asyncio.get_event_loop().time()
 
        async with client.messages.stream(
            model="claude-opus-4-6",
            max_tokens=2048,
            messages=messages
        ) as stream:
            async for text in stream:
                self.buffer.append(text)
                accumulated_text += text
 
                current_time = asyncio.get_event_loop().time()
                should_flush = (
                    len(self.buffer) == self.buffer.maxlen or
                    (current_time - last_flush) > self.flush_timeout
                )
 
                if should_flush:
                    # バッファ コンテンツを出力
                    yield accumulated_text
                    accumulated_text = ""
                    last_flush = current_time
 
            # 残りのコンテンツをフラッシュ
            if accumulated_text:
                yield accumulated_text

⚠️

無期限にバッファリングしないでください。新しいトークンが到着しない場合でも、メモリの成長を防ぎ、応答性を確保するために、常に時間ベースのフラッシュ制限を実装します。

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦529 overloaded_error を ±50% ジッター付きバックオフで最終失敗率 2.1%→0.18% に下げた実測パターン

✦並列ツール呼び出しを結果サイズ別に振り分け、平均レイテンシ 8.4秒→5.1秒・タイムアウト率 1.9%→0.4% にした判断基準

✦会話チェックポイントで再実行コストを約60%削減する、最適化より先に入れるべき設計順序

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

並列ツール呼び出しパターン

複数のツール呼び出しの調整

Claude は単一の応答で複数のツール呼び出しを提案できます。本番システムは並列実行を適切に処理する必要があります:

import asyncio
from concurrent.futures import ThreadPoolExecutor
from typing import Any
 
class ToolExecutor:
    """ツール呼び出しを安全に実行します (適切な同時実行管理)。"""
 
    def __init__(self, max_concurrent: int = 5):
        self.max_concurrent = max_concurrent
        self.executor = ThreadPoolExecutor(max_workers=max_concurrent)
 
    async def execute_parallel_tools(
        self,
        tool_calls: list[dict]
    ) -> list[dict]:
        """
        複数のツール呼び出しを並列に実行します:
        - 同時実行の制限
        - エラーの分離
        - タイムアウト保護
        - 結果追跡
        """
        results = []
        semaphore = asyncio.Semaphore(self.max_concurrent)
 
        async def execute_single_tool(tool_call: dict) -> dict:
            async with semaphore:
                try:
                    # タイムアウト付きで実行
                    result = await asyncio.wait_for(
                        self._execute_tool(tool_call),
                        timeout=30.0
                    )
                    return {
                        "tool_use_id": tool_call["id"],
                        "content": result,
                        "error": None
                    }
                except asyncio.TimeoutError:
                    return {
                        "tool_use_id": tool_call["id"],
                        "content": None,
                        "error": "ツール実行タイムアウト (30秒)"
                    }
                except Exception as e:
                    return {
                        "tool_use_id": tool_call["id"],
                        "content": None,
                        "error": f"ツール実行失敗: {str(e)}"
                    }
 
        # すべてのツールを同時実行
        tasks = [execute_single_tool(tc) for tc in tool_calls]
        results = await asyncio.gather(*tasks)
 
        return results
 
    async def _execute_tool(self, tool_call: dict) -> Any:
        """リトライ ロジック付きでツール 1 個を実行。"""
        tool_name = tool_call["name"]
        tool_args = tool_call["input"]
 
        # 適切なツール ハンドラーにルート
        handler = self.get_tool_handler(tool_name)
        return await handler(**tool_args)

ツール使用との思考の連鎖

複雑な問題は、Claude がツールを呼び出す前にソリューションを考え抜くことから恩恵を受けます:

def solve_with_chain_of_thought(
    problem: str,
    available_tools: list[dict]
) -> str:
    """
    Claude の推論を活用:
    1. 問題を考える (ツールなし)
    2. ツール使用を計画
    3. 計画に従ってツールを実行
    4. 結果を合成
    """
 
    # ステップ 1: 問題を考える
    thinking_response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2000,
        system="問題解決者です。提案する前に、これを段階的に考え抜いてください。",
        messages=[
            {
                "role": "user",
                "content": f"問題: {problem}\n\nこの問題をどのように解決するかを考え抜いてください。"
            }
        ]
    )
 
    thinking_process = thinking_response.content[0].text
 
    # ステップ 2: 思考に基づいてツールを使用
    tool_response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=4000,
        tools=available_tools,
        system=f"""このプロブレムを既に考え抜きました:
{thinking_process}
 
現在は利用可能なツールを使用して計画を実行してください。""",
        messages=[
            {
                "role": "user",
                "content": f"問題: {problem}"
            }
        ]
    )
 
    # ステップ 3: ツールを実行して合成
    tool_calls = [
        block for block in tool_response.content
        if block.type == "tool_use"
    ]
 
    tool_results = execute_tools(tool_calls)
 
    # ステップ 4: 最終合成を取得
    final_response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2000,
        messages=[
            {
                "role": "user",
                "content": problem
            },
            {
                "role": "assistant",
                "content": tool_response.content
            },
            {
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": result["tool_use_id"],
                        "content": json.dumps(result["content"])
                    }
                    for result in tool_results
                ]
            }
        ]
    )
 
    return final_response.content[0].text

洗練されたエラー処理

復旧に適切なエラー分類

すべてのエラーが同じとは限りません。本番システムは回復可能なエラーと回復不可能なエラーを区別する必要があります:

import time
from enum import Enum
from typing import Callable, Any
 
class ErrorCategory(Enum):
    TRANSIENT = "transient"        # すぐにリトライ
    RATE_LIMITED = "rate_limited"  # バックオフ付きリトライ
    AUTH = "auth"                  # 人的介入が必要
    INVALID_REQUEST = "invalid"    # リトライしないでください
    SERVER_ERROR = "server"        # バックオフ付きリトライ
 
def categorize_error(error: Exception) -> ErrorCategory:
    """API エラーを分類して適切に処理。"""
    if isinstance(error, anthropic.RateLimitError):
        return ErrorCategory.RATE_LIMITED
    elif isinstance(error, anthropic.APIConnectionError):
        return ErrorCategory.TRANSIENT
    elif isinstance(error, anthropic.AuthenticationError):
        return ErrorCategory.AUTH
    elif isinstance(error, anthropic.BadRequestError):
        return ErrorCategory.INVALID_REQUEST
    elif isinstance(error, anthropic.InternalServerError):
        return ErrorCategory.SERVER_ERROR
    else:
        return ErrorCategory.TRANSIENT
 
class ResilientAPIClient:
    """洗練されたエラー処理を備えた API クライアント。"""
 
    def __init__(self, api_key: str, max_retries: int = 5):
        self.client = anthropic.Anthropic(api_key=api_key)
        self.max_retries = max_retries
 
    def call_with_smart_retry(
        self,
        api_call: Callable,
        *args,
        **kwargs
    ) -> Any:
        """
        カテゴリ対応リトライ ロジック付きで API 呼び出しを実行。
        """
        retry_count = 0
        base_wait_time = 1.0  # 秒
 
        while retry_count < self.max_retries:
            try:
                return api_call(*args, **kwargs)
 
            except Exception as e:
                category = categorize_error(e)
 
                if category == ErrorCategory.AUTH:
                    # 認証エラーはリトライしない
                    raise AuthenticationError(f"認証失敗: {e}")
 
                elif category == ErrorCategory.INVALID_REQUEST:
                    # 不正なリクエストはリトライしない
                    raise ValueError(f"不正なリクエスト: {e}")
 
                elif category == ErrorCategory.TRANSIENT:
                    # 指数バックオフ付きでリトライ
                    wait_time = base_wait_time * (2 ** retry_count)
                    time.sleep(wait_time)
                    retry_count += 1
                    continue
 
                elif category == ErrorCategory.RATE_LIMITED:
                    # レート制限用の指数バックオフ (ジッター付き)
                    wait_time = base_wait_time * (2 ** retry_count)
                    jitter = wait_time * 0.1 * (2 * (hash(time.time()) % 2) - 1)
                    time.sleep(wait_time + jitter)
                    retry_count += 1
                    continue
 
                elif category == ErrorCategory.SERVER_ERROR:
                    # サーバー エラーの長いバックオフ
                    wait_time = base_wait_time * (2 ** retry_count) * 2
                    time.sleep(wait_time)
                    retry_count += 1
                    continue
 
        raise Exception(f"最大リトライ回数 ({self.max_retries}) を超過")

💡

レート制限エラーの場合は、ジッター付きの指数バックオフを実装してください。単純な線形バックオフは、複数のクライアントが同時にリトライするときに「 thundering herd」効果を引き起こすことがあります。

エラー復旧のためのコンテキスト管理

障害からグレースフルに回復するために、会話コンテキストを保持します:

class ConversationManager:
    """エラー復旧が堅牢に行われるように会話状態を管理。"""
 
    def __init__(self, user_id: str, session_id: str):
        self.user_id = user_id
        self.session_id = session_id
        self.messages = []
        self.checkpoints = []  # 定期的に状態を保存
 
    def add_message(self, role: str, content: Any):
        """メッセージを追加してチェックポイントを保存。"""
        self.messages.append({
            "role": role,
            "content": content
        })
 
    def save_checkpoint(self):
        """復旧用に会話状態を保存。"""
        checkpoint = {
            "timestamp": time.time(),
            "message_count": len(self.messages),
            "messages": self.messages.copy(),
            "checksum": self._compute_checksum()
        }
        self.checkpoints.append(checkpoint)
        # データベースに保持
        persist_checkpoint(self.user_id, self.session_id, checkpoint)
 
    def recover_from_failure(self):
        """会話を最後の既知の良い状態に復元。"""
        if not self.checkpoints:
            return False
 
        last_checkpoint = self.checkpoints[-1]
        self.messages = last_checkpoint["messages"].copy()
        return True
 
    def get_next_message_count(self) -> int:
        """ページング/継続のメッセージ数を取得。"""
        return len(self.messages)
 
    def _compute_checksum(self) -> str:
        """整合性検証用チェックサム。"""
        content = json.dumps(self.messages, sort_keys=True)
        import hashlib
        return hashlib.sha256(content.encode()).hexdigest()

スケールでのリトライ戦略

適応型リトライパターン

単純な指数バックオフはすべてのシナリオに最適ではありません。適応型リトライは成功率を向上させます:

import random
from dataclasses import dataclass
 
@dataclass
class RetryMetrics:
    total_attempts: int
    success_count: int
    failure_count: int
    last_error: str
    average_latency: float
 
class AdaptiveRetryStrategy:
    """リトライ パターンから学習して戦略を動的に適応。"""
 
    def __init__(self):
        self.metrics: dict[str, RetryMetrics] = {}
 
    def get_retry_delay(
        self,
        error_type: str,
        attempt_number: int,
        recent_success_rate: float
    ) -> float:
        """
        以下に基づいて最適なリトライ遅延を計算:
        - エラーのタイプ
        - 試行回数
        - 最近の成功率
        """
 
        base_delay = 1.0
 
        # 成功率に基づいてスケール遅延
        # 高い成功率: より積極的なリトライ
        # 低い成功率: より保守的なリトライ
        success_factor = 2.0 if recent_success_rate > 0.8 else 0.5
 
        # 指数バックオフ
        exponential_delay = base_delay * (2 ** attempt_number) * success_factor
 
        # thundering herd を防ぐためにジッターを追加
        jitter = exponential_delay * random.uniform(0, 0.1)
 
        # 最大に上限 (例: 5 分)
        max_delay = 300.0
        return min(exponential_delay + jitter, max_delay)
 
    def should_retry(
        self,
        error_type: str,
        attempt_number: int,
        success_rate: float
    ) -> bool:
        """条件に基づいてリトライするかどうかを判断。"""
 
        # 回復不可能なエラーはリトライしない
        if error_type in ["AUTH", "INVALID_REQUEST"]:
            return False
 
        # 常にレート制限と一時的なエラーをリトライ
        if error_type in ["RATE_LIMITED", "TRANSIENT"]:
            return attempt_number < 10
 
        # サーバー エラーの場合、成功率が低い場合は保守的に
        if error_type == "SERVER_ERROR":
            return attempt_number < (5 if success_rate > 0.7 else 3)
 
        return attempt_number < 5

効率のためのバッチ処理

複数のリクエストを処理する場合、インテリジェントにバッチします:

class BatchProcessor:
    """効率のためにリクエストをバッチで処理。"""
 
    def __init__(self, batch_size: int = 10, batch_timeout: float = 5.0):
        self.batch_size = batch_size
        self.batch_timeout = batch_timeout
        self.batch = []
        self.batch_start_time = None
 
    async def add_request(self, request: dict) -> str:
        """リクエストをバッチに追加。バッチがいっぱいか、タイムアウトしたときにフラッシュ。"""
        self.batch.append(request)
 
        if self.batch_start_time is None:
            self.batch_start_time = time.time()
 
        # バッチがいっぱいの場合はフラッシュ
        if len(self.batch) >= self.batch_size:
            return await self.flush_batch()
 
        # タイムアウトを超えた場合はフラッシュ
        elapsed = time.time() - self.batch_start_time
        if elapsed > self.batch_timeout and self.batch:
            return await self.flush_batch()
 
        return "queued"
 
    async def flush_batch(self) -> str:
        """現在のバッチ内のすべてのリクエストを処理。"""
        if not self.batch:
            return "empty"
 
        batch_to_process = self.batch.copy()
        self.batch = []
        self.batch_start_time = None
 
        # 可能な場合は単一の API 呼び出しでバッチを処理
        results = await self._process_batch(batch_to_process)
        return f"processed_{len(results)}"
 
    async def _process_batch(self, requests: list) -> list:
        """リクエストのバッチを効率的に処理。"""
        # 適切な場合は単一のプロンプトに結合
        combined_prompt = self._combine_requests(requests)
 
        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=4000,
            messages=[
                {
                    "role": "user",
                    "content": combined_prompt
                }
            ]
        )
 
        return self._parse_batch_response(response)

リソース最適化

トークン予算

予期しないコストを防ぐためにトークン消費を管理します:

class TokenBudget:
    """ユーザー/セッションごとのトークン予算を追跡して適用。"""
 
    def __init__(self, monthly_budget: int = 1_000_000):
        self.monthly_budget = monthly_budget
        self.usage_this_month: dict[str, int] = {}
 
    def check_budget(self, user_id: str, estimated_tokens: int) -> bool:
        """このリクエストのバジェットがユーザーにあるかどうかを確認。"""
        current_usage = self.usage_this_month.get(user_id, 0)
 
        if current_usage + estimated_tokens > self.monthly_budget:
            return False
        return True
 
    def record_usage(self, user_id: str, tokens_used: int):
        """予算のためにトークン使用量を記録。"""
        self.usage_this_month[user_id] = (
            self.usage_this_month.get(user_id, 0) + tokens_used
        )
 
    def estimate_tokens(self, messages: list[dict]) -> int:
        """トークンの大まかな推定 (1 トークン ≈ 4 文字)。"""
        total_chars = sum(
            len(msg.get("content", ""))
            for msg in messages
        )
        return total_chars // 4

キャッシング頻繁に繰り返される呼び出し

プロンプトキャッシングを使用して繰り返されるリクエストを最適化します:

from functools import lru_cache
import hashlib
 
class CachedAPIClient:
    """API レスポンスをキャッシュしてトークン使用量とレイテンシを削減。"""
 
    def __init__(self):
        self.cache = {}
        self.cache_stats = {
            "hits": 0,
            "misses": 0,
            "tokens_saved": 0
        }
 
    def _compute_cache_key(self, messages: list, model: str) -> str:
        """メッセージからキャッシュ キーを計算。"""
        content = json.dumps(messages, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()
 
    def get_cached_response(
        self,
        messages: list[dict],
        model: str = "claude-opus-4-6"
    ) -> dict | None:
        """利用可能な場合はキャッシュされたレスポンスを取得。"""
        cache_key = self._compute_cache_key(messages, model)
 
        if cache_key in self.cache:
            self.cache_stats["hits"] += 1
            return self.cache[cache_key]
 
        self.cache_stats["misses"] += 1
        return None
 
    def cache_response(
        self,
        messages: list[dict],
        response: dict,
        model: str = "claude-opus-4-6"
    ):
        """API レスポンスをキャッシュ。"""
        cache_key = self._compute_cache_key(messages, model)
        self.cache[cache_key] = response
 
    def call_with_cache(
        self,
        messages: list[dict],
        **kwargs
    ) -> dict:
        """キャッシング付きで API を呼び出し。"""
        # まずキャッシュをチェック
        cached = self.get_cached_response(messages)
        if cached:
            return cached
 
        # API を呼び出し
        response = client.messages.create(
            messages=messages,
            **kwargs
        )
 
        # キャッシュして戻す
        self.cache_response(messages, response, kwargs.get("model", "claude-opus-4-6"))
        return response

監視と観測性

デバッグ用の包括的なロギング

問題を素早く表面化させるためにログを構造化します:

import logging
from dataclasses import dataclass
from datetime import datetime
 
@dataclass
class APICallLog:
    timestamp: datetime
    request_id: str
    model: str
    input_tokens: int
    output_tokens: int
    latency_ms: float
    error: str | None
    retry_count: int
    cache_hit: bool
 
class APILogger:
    """API 呼び出しの構造化ロギング。"""
 
    def __init__(self, name: str = "claude_api"):
        self.logger = logging.getLogger(name)
 
    def log_api_call(self, log: APICallLog):
        """構造化データで API 呼び出しをログ。"""
        self.logger.info(
            "API call completed",
            extra={
                "request_id": log.request_id,
                "model": log.model,
                "input_tokens": log.input_tokens,
                "output_tokens": log.output_tokens,
                "latency_ms": log.latency_ms,
                "error": log.error,
                "retry_count": log.retry_count,
                "cache_hit": log.cache_hit,
                "total_tokens": log.input_tokens + log.output_tokens
            }
        )
 
    def log_error_pattern(self, error_type: str, count: int, rate: float):
        """エラー パターンで警告。"""
        if rate > 0.1:  # 10% 以上のエラー率
            self.logger.warning(
                f"高いエラー率が検出: {error_type}",
                extra={
                    "error_type": error_type,
                    "error_count": count,
                    "error_rate": rate
                }
            )

公式ドキュメントには書かれていない、運用で気づいた点

ここからは、私が iOS/Android アプリ 6 本のバックエンド処理で Claude API を回しながら気づいた、ドキュメントには明記されていない実運用上のポイントを共有します。数百万回規模の呼び出しを通して見えてきた、地味だけれど効くところです。

`overloaded_error` は「待てば必ず回復する」前提で設計する

公式ドキュメントは 529（overloaded_error）を「リトライ可能」とだけ説明しますが、実運用では 失敗そのものより、失敗後の挙動でシステムの安定度が決まります。私の壁紙アプリの説明文自動生成バッチ（1 日あたり約 4,000 件）では、ピーク帯に 529 が瞬間的に 3〜5% まで上がる時間帯がありました。ここで固定間隔リトライをかけると、同じ秒に全リクエストが再突入して二次的な過負荷を生みます。

ジッター付き指数バックオフに切り替えてから、529 起因の最終失敗率は 2.1% → 0.18% に下がりました。コードは本文の「スケールでのリトライ戦略」のものをそのまま使っていますが、効いたのは base_delay を上げることではなく ジッターの幅を ±50% まで広げたことでした。

import random
 
def backoff_with_jitter(attempt: int, base: float = 1.0, cap: float = 30.0) -> float:
    # 公式例の固定バックオフではなく、±50% のジッターを必ず入れる
    raw = min(cap, base * (2 ** attempt))
    return raw * random.uniform(0.5, 1.5)

並列ツール呼び出しは「同時実行数」より「合計トークン予算」で律速される

並列ツール呼び出しを増やすと速くなる、と思いがちですが、実際に律速するのは同時実行数ではなくレスポンスの合計トークン量でした。並列度を 4 → 8 に上げても、各ツール結果が大きいケースではスループットがほとんど伸びず、むしろタイムアウト率が上がりました。

私が現在採用している判断基準はシンプルです。

ツール結果の想定サイズを 小（<500 トークン）/中/大（>2,000 トークン） で分類する
小さい結果のツールだけ並列度を上げる（最大 6）
大きい結果を返すツール（DB 全文検索・ファイル読み込み等）は並列度 2 に抑える
1 リクエストの合計入力を概ね 30,000 トークン以内に収める

この振り分けにした後、平均レイテンシは 8.4 秒 → 5.1 秒、タイムアウト率は 1.9% → 0.4% に改善しました。

コスト監視は「総額」ではなく「呼び出しあたり中央値」で見る

総コストだけを見ていると、たまに混じる巨大プロンプトに気づけません。呼び出しあたりトークン数の 中央値と p95 を並べて監視するようにしたところ、特定のツールが想定の 6 倍のコンテキストを送っていたバグを早期に発見できました。AdMob を 2014 年から運用してきて「異常値は平均ではなく分布の裾に出る」と痛感してきたので、API コストにも同じ見方を持ち込んでいます。

私の推奨：最初に作るべきは「会話チェックポイント」

新しくこの種のシステムを組むなら、最適化より先に 会話状態のチェックポイント保存を入れることを強くおすすめします。マルチステップのツール呼び出しは途中で落ちると、それまでの数回分の API コストが丸ごと無駄になります。各ステップ後に messages 配列を永続化しておけば、失敗時に最後の成功地点から再開でき、再実行コストを実測で 約 60% 削減できました。速度より先に「やり直しが安いこと」を担保する——これが本番運用での私の優先順位です。