Claude API の Parallel Tool Use を本番で使いこなす — 並列実行の制御・エラー設計・レイテンシ最適化

# 並列を明示的に禁止する例
import anthropic
 
client = anthropic.Anthropic()
 
response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    tools=[
        {"name": "get_inventory", "description": "在庫を取得", "input_schema": {"type": "object", "properties": {"sku": {"type": "string"}}}},
        {"name": "get_price", "description": "価格を取得", "input_schema": {"type": "object", "properties": {"sku": {"type": "string"}}}},
    ],
    tool_choice={
        "type": "auto",
        "disable_parallel_tool_use": True  # 1ターンに1ツールだけ呼ぶ
    },
    messages=[{"role": "user", "content": "SKU-A123 の在庫と価格を教えて"}]
)

import anthropic
import asyncio
import httpx
from typing import Any
 
client = anthropic.AsyncAnthropic()
 
# 各ツールの実装（async で書くのが必須）
async def get_inventory(sku: str) -> dict:
    async with httpx.AsyncClient(timeout=5.0) as http:
        r = await http.get(f"https://api.example.com/inventory/{sku}")
        r.raise_for_status()
        return r.json()
 
async def get_price(sku: str) -> dict:
    async with httpx.AsyncClient(timeout=5.0) as http:
        r = await http.get(f"https://api.example.com/price/{sku}")
        r.raise_for_status()
        return r.json()
 
async def get_shipping(sku: str, zip_code: str) -> dict:
    async with httpx.AsyncClient(timeout=5.0) as http:
        r = await http.get(f"https://api.example.com/shipping", params={"sku": sku, "zip": zip_code})
        r.raise_for_status()
        return r.json()
 
TOOL_MAP = {
    "get_inventory": get_inventory,
    "get_price": get_price,
    "get_shipping": get_shipping,
}
 
# 並列度の上限（外部APIへの同時接続を制限）
SEMAPHORE = asyncio.Semaphore(5)
 
async def run_tool(block: Any) -> dict:
    """1つの tool_use ブロックを実行して tool_result を返す"""
    async with SEMAPHORE:
        try:
            fn = TOOL_MAP[block.name]
            result = await fn(**block.input)
            return {
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": str(result),
                "is_error": False,
            }
        except Exception as e:
            return {
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": f"Error: {type(e).__name__}: {e}",
                "is_error": True,
            }
 
async def run_parallel_tools(response) -> list[dict]:
    """Claude の応答内のすべての tool_use ブロックを並列実行"""
    tool_blocks = [b for b in response.content if b.type == "tool_use"]
    # asyncio.gather で並列実行（Semaphore で同時実行数を制限）
    results = await asyncio.gather(
        *[run_tool(b) for b in tool_blocks],
        return_exceptions=False  # 例外は run_tool 内でキャッチ済み
    )
    return results

async def strict_multi_tool(messages, tools):
    """全ツール成功を必須とする厳格モード"""
    response = await client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=2048,
        tools=tools,
        messages=messages,
    )
    if response.stop_reason != "tool_use":
        return response
 
    results = await run_parallel_tools(response)
    # 1つでも失敗があればリトライへ
    if any(r["is_error"] for r in results):
        raise PartialFailureError(
            failed_tools=[r["tool_use_id"] for r in results if r["is_error"]]
        )
    # 全て成功 → 次のターンへ
    messages.append({"role": "assistant", "content": response.content})
    messages.append({"role": "user", "content": results})
    return await strict_multi_tool(messages, tools)

async def lenient_multi_tool(messages, tools):
    """部分失敗を許容するモード — Claude に判断を委ねる"""
    response = await client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=2048,
        tools=tools,
        messages=messages,
    )
    if response.stop_reason != "tool_use":
        return response
 
    results = await run_parallel_tools(response)
    messages.append({"role": "assistant", "content": response.content})
    messages.append({"role": "user", "content": results})
 
    # エラーがあってもそのまま続行 → Claude が部分情報で回答
    return await lenient_multi_tool(messages, tools)

async def run_tool_with_retry(block, max_retries=3):
    """指数バックオフで個別リトライ"""
    for attempt in range(max_retries):
        async with SEMAPHORE:
            try:
                fn = TOOL_MAP[block.name]
                result = await fn(**block.input)
                return {
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(result),
                    "is_error": False,
                }
            except (httpx.TimeoutException, httpx.ConnectError) as e:
                # 一時的なエラーはリトライ
                if attempt < max_retries - 1:
                    await asyncio.sleep(2 ** attempt)
                    continue
                return {
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": f"Error after {max_retries} retries: {e}",
                    "is_error": True,
                }
            except Exception as e:
                # リトライ不可能なエラーは即失敗
                return {
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": f"Non-retryable error: {e}",
                    "is_error": True,
                }

async def create_order(user_id: str, sku: str, quantity: int, idempotency_key: str) -> dict:
    """注文作成 — 冪等性キーで重複防止"""
    async with httpx.AsyncClient(timeout=10.0) as http:
        r = await http.post(
            "https://api.example.com/orders",
            json={"user_id": user_id, "sku": sku, "quantity": quantity},
            headers={"Idempotency-Key": idempotency_key}
        )
        r.raise_for_status()
        return r.json()

async def run_tool_idempotent(block) -> dict:
    """tool_use_id を冪等性キーに自動注入"""
    if block.name == "create_order":
        # ツールの引数に idempotency_key を自動付与
        input_with_key = {**block.input, "idempotency_key": block.id}
        result = await create_order(**input_with_key)
    else:
        fn = TOOL_MAP[block.name]
        result = await fn(**block.input)
    return {"type": "tool_result", "tool_use_id": block.id, "content": str(result)}

import anthropic
import asyncio
 
client = anthropic.AsyncAnthropic()
 
async def stream_and_execute_parallel(messages, tools):
    tool_blocks = []
    assistant_content = []
 
    async with client.messages.stream(
        model="claude-sonnet-4-5",
        max_tokens=2048,
        tools=tools,
        messages=messages,
    ) as stream:
        async for event in stream:
            # 各イベントを順次処理（テキストブロック・tool_use ブロック両方）
            pass
        # ストリーム完了時点で final_message を取得
        final = await stream.get_final_message()
        tool_blocks = [b for b in final.content if b.type == "tool_use"]
        assistant_content = final.content
 
    # ストリーム完了後にまとめて並列実行
    if tool_blocks:
        results = await asyncio.gather(*[run_tool(b) for b in tool_blocks])
        messages.append({"role": "assistant", "content": assistant_content})
        messages.append({"role": "user", "content": results})
        # 次のターンへ
        return await stream_and_execute_parallel(messages, tools)
 
    return final

import time
from prometheus_client import Histogram, Counter
 
TOOL_LATENCY = Histogram(
    "claude_tool_duration_seconds",
    "Tool execution duration",
    ["tool_name", "status"]
)
TOOL_CALLS = Counter(
    "claude_tool_calls_total",
    "Total tool calls",
    ["tool_name", "status"]
)
PARALLEL_DEGREE = Histogram(
    "claude_parallel_tool_degree",
    "Number of tools called in parallel per turn",
    buckets=[1, 2, 3, 5, 8, 13, 21]
)
 
async def run_tool_observed(block) -> dict:
    start = time.monotonic()
    status = "success"
    try:
        async with SEMAPHORE:
            fn = TOOL_MAP[block.name]
            result = await fn(**block.input)
            return {
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": str(result),
                "is_error": False,
            }
    except Exception as e:
        status = "error"
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": f"Error: {e}",
            "is_error": True,
        }
    finally:
        duration = time.monotonic() - start
        TOOL_LATENCY.labels(tool_name=block.name, status=status).observe(duration)
        TOOL_CALLS.labels(tool_name=block.name, status=status).inc()
 
async def run_parallel_tools_observed(response):
    tool_blocks = [b for b in response.content if b.type == "tool_use"]
    PARALLEL_DEGREE.observe(len(tool_blocks))  # 並列度を記録
    return await asyncio.gather(*[run_tool_observed(b) for b in tool_blocks])