Claude API のストリーミングは「全部届く」前提が崩れる — 部分失敗を回復に変える運用メモ

import asyncio, logging
import anthropic
 
log = logging.getLogger(__name__)
 
class ResilientStream:
    def __init__(self, client: anthropic.AsyncAnthropic, max_retries: int = 3):
        self.client = client
        self.max_retries = max_retries
        self.state = StreamState()
 
    async def run(self, **kwargs):
        attempt = 0
        while True:
            try:
                async with self.client.messages.stream(**kwargs) as stream:
                    async for event in stream:
                        self._track(event)
                        yield event
                    return  # message_stop まで到達
            except (anthropic.APIConnectionError, anthropic.APITimeoutError) as e:
                attempt += 1
                if attempt > self.max_retries:
                    log.error("stream gave up after %d attempts: %s", attempt, e)
                    raise
                await asyncio.sleep(min(2 ** attempt, 20))
                kwargs = self._resume_kwargs(kwargs)  # 部分応答を履歴に積む
 
    def _track(self, event) -> None:
        t = event.type
        if t == "message_start":
            self.state.message_id = event.message.id
            self.state.model = event.message.model
        elif t == "content_block_start":
            self.state.current_index = event.index
        elif t == "content_block_delta":
            self.state.apply_delta(event.index, event.delta.model_dump())
        elif t == "message_delta" and event.delta.stop_reason:
            self.state.stop_reason = event.delta.stop_reason
 
    def _resume_kwargs(self, original: dict) -> dict:
        partial = self.state.assistant_text()
        if not partial:
            return original
        resumed = dict(original)
        resumed["messages"] = list(original.get("messages", [])) + [
            {"role": "assistant", "content": partial}
        ]
        return resumed

from hashlib import sha256
 
class DedupWindow:
    """直近 window 件のイベントを覚えて、二度目以降を捨てる。"""
    def __init__(self, window: int = 512):
        self.seen: set[str] = set()
        self.order: list[str] = []
        self.window = window
 
    def is_duplicate(self, message_id: str, index: int, position: int) -> bool:
        key = sha256(f"{message_id}:{index}:{position}".encode()).hexdigest()
        if key in self.seen:
            return True
        self.seen.add(key)
        self.order.append(key)
        if len(self.order) > self.window:
            self.seen.discard(self.order.pop(0))
        return False

import json
from typing import Optional
 
def parse_tool_input(block: dict) -> Optional[dict]:
    """tool_use ブロックの partial_json を安全に読む。途中なら None。"""
    raw = block.get("partial_json", "").strip()
    if not raw:
        return None
    try:
        return json.loads(raw)
    except json.JSONDecodeError as e:
        log.warning("incomplete tool input, skipping: %s (head=%r)", e, raw[:120])
        return None
 
# 実行境界では必ずこの関数を通す
args = parse_tool_input(block)
if args is None:
    # 引数が揃っていない間はツールを起動しない
    return
run_tool(block["name"], args)

import random
 
async def backoff_for(attempt: int, error: Exception) -> None:
    if isinstance(error, anthropic.RateLimitError):
        base = getattr(error, "retry_after", None) or 30
        delay = base + random.uniform(0, 5)         # サーバー指示 + 散らし
    elif isinstance(error, (anthropic.APIConnectionError, anthropic.APITimeoutError)):
        delay = min(2 ** attempt, 20) + random.uniform(0, 1)
    else:
        delay = min(2 ** attempt, 60)
    log.info("backoff %.1fs (attempt=%d, %s)", delay, attempt, type(error).__name__)
    await asyncio.sleep(delay)

if event.type == "error":
    err = event.error
    if err.type == "overloaded_error":
        raise anthropic.APIStatusError(err.message)  # 容量不足は再試行の価値あり
    raise RuntimeError(f"fatal stream error: {err.type}: {err.message}")