Claude API × Webhook・非同期処理のエラーとリカバリー戦略

import anthropic
import time
import random
import logging
from typing import Optional
 
logger = logging.getLogger(__name__)
 
def call_claude_with_retry(
    payload: dict,
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0
) -> Optional[str]:
    """エクスポネンシャルバックオフ付きリトライ"""
    client = anthropic.Anthropic()
    PERMANENT_ERROR_CODES = {400, 401, 403, 404}
 
    for attempt in range(max_retries + 1):
        try:
            response = client.messages.create(
                model=payload.get('model', 'claude-sonnet-4-6'),
                max_tokens=payload.get('max_tokens', 2048),
                messages=payload['messages']
            )
            return response.content[0].text
 
        except anthropic.RateLimitError as e:
            if attempt >= max_retries:
                raise
            retry_after = e.response.headers.get('retry-after')
            wait_time = float(retry_after) if retry_after else min(
                base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay
            )
            logger.warning(f"レート制限: {wait_time:.1f}秒後にリトライ (試行 {attempt + 1}/{max_retries})")
            time.sleep(wait_time)
 
        except anthropic.APIStatusError as e:
            if e.status_code in PERMANENT_ERROR_CODES:
                logger.error(f"恒久的エラー {e.status_code}: リトライ不可")
                raise
            if attempt >= max_retries:
                raise
            wait_time = min(base_delay * (2 ** attempt) + random.uniform(0, 1), max_delay)
            logger.warning(f"サーバーエラー {e.status_code}: {wait_time:.1f}秒後にリトライ")
            time.sleep(wait_time)
 
        except anthropic.APIConnectionError:
            if attempt >= max_retries:
                raise
            wait_time = min(base_delay * (2 ** attempt), max_delay)
            logger.warning(f"接続エラー: {wait_time:.1f}秒後にリトライ")
            time.sleep(wait_time)
 
    return None

import time
 
def handle_job_failure(job_data: dict, error: Exception, max_retries: int = 3):
    """失敗ジョブのハンドリングとDLQ移動"""
    retry_count = job_data.get('retry_count', 0)
 
    if retry_count < max_retries:
        delay = 2 ** retry_count  # 1, 2, 4秒
        retry_job = {
            **job_data,
            'retry_count': retry_count + 1,
            'last_error': str(error),
            'retry_after': time.time() + delay
        }
        redis_client.zadd('claude_jobs_delayed',
                          {json.dumps(retry_job): retry_job['retry_after']})
        logger.info(f"リトライキューに追加: {job_data['job_id']} ({retry_count + 1}回目)")
    else:
        dead_job = {
            **job_data,
            'final_error': str(error),
            'failed_at': time.time(),
            'total_attempts': retry_count + 1
        }
        redis_client.rpush('claude_jobs_dead', json.dumps(dead_job))
        send_alert(
            f"[Claude API] ジョブが完全に失敗: {job_data['job_id']}\n"
            f"エラー: {error}\n試行回数: {retry_count + 1}"
        )
        logger.error(f"DLQに移動: {job_data['job_id']}")

from enum import Enum
import threading
 
class CircuitState(Enum):
    CLOSED = "closed"        # 正常：リクエストを通す
    OPEN = "open"            # 異常：リクエストをブロック
    HALF_OPEN = "half_open"  # 回復確認中
 
class ClaudeAPICircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60.0, success_threshold=2):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
        self.lock = threading.Lock()
 
    def call(self, func, *args, **kwargs):
        with self.lock:
            if self.state == CircuitState.OPEN:
                if time.time() - self.last_failure_time > self.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.success_count = 0
                else:
                    raise Exception("サーキットブレーカーがOPEN状態: Claude APIへのアクセスをブロック中")
 
        try:
            result = func(*args, **kwargs)
            with self.lock:
                if self.state == CircuitState.HALF_OPEN:
                    self.success_count += 1
                    if self.success_count >= self.success_threshold:
                        self.state = CircuitState.CLOSED
                        self.failure_count = 0
                        logger.info("サーキットブレーカー: CLOSEDに回復")
                elif self.state == CircuitState.CLOSED:
                    self.failure_count = 0
            return result
 
        except Exception as e:
            with self.lock:
                self.failure_count += 1
                self.last_failure_time = time.time()
                if (self.state == CircuitState.CLOSED and
                        self.failure_count >= self.failure_threshold):
                    self.state = CircuitState.OPEN
                    logger.error(f"サーキットブレーカーがOPENに: {self.failure_count}回連続失敗")
                elif self.state == CircuitState.HALF_OPEN:
                    self.state = CircuitState.OPEN
            raise
 
circuit_breaker = ClaudeAPICircuitBreaker(failure_threshold=5, recovery_timeout=120.0)

import hashlib
 
class GracefulClaudeHandler:
    """Claude API 障害時のグレースフルデグラデーション"""
 
    def __init__(self):
        self.circuit_breaker = ClaudeAPICircuitBreaker()
        self.fallback_responses = {
            'summarize': 'サービスが一時的に利用できないため、要約を提供できません。後ほどお試しください。',
            'analyze': '分析機能は現在メンテナンス中です。',
            'default': 'ご不便をおかけして申し訳ありません。現在サービスが混み合っています。'
        }
 
    def process(self, task_type: str, payload: dict) -> dict:
        # 1. キャッシュを確認
        cache_key = hashlib.md5(json.dumps(payload, sort_keys=True).encode()).hexdigest()
        cached = redis_client.get(f'response_cache:{cache_key}')
        if cached:
            return {'source': 'cache', 'result': json.loads(cached)}
 
        # 2. Claude API を試みる（サーキットブレーカー経由）
        try:
            result = self.circuit_breaker.call(call_claude_with_retry, payload)
            redis_client.setex(f'response_cache:{cache_key}', 3600, json.dumps(result))
            return {'source': 'claude_api', 'result': result}
 
        except Exception as e:
            logger.error(f"Claude API 失敗: {e}")
            # 3. フォールバックレスポンスを返す
            fallback = self.fallback_responses.get(task_type, self.fallback_responses['default'])
            return {'source': 'fallback', 'result': fallback, 'degraded': True}

def check_and_alert(metrics: dict):
    if metrics['error_rate'] > 0.1:
        send_alert(f"⚠️ Claude API エラー率が高い: {metrics['error_rate']:.1%}")
    if metrics['dlq_count'] > 10:
        send_alert(f"🚨 DLQ に {metrics['dlq_count']} 件の失敗ジョブが蓄積")
    if metrics['avg_processing_time'] > 30:
        send_alert(f"⏱️ Claude API の平均応答時間が遅い: {metrics['avg_processing_time']:.1f}秒")