Claude本番運用ガイド｜レート制限・エラー処理・モニタリングの実践

importanthropic
importtime
importrandom

defcall_with_backoff(
    client: anthropic.Anthropic,
    max_retries: int = 5,
    base_delay: float = 1.0,
    **kwargs
):
    for attempt in range(max_retries):
        try:
            return client.messages.create(**kwargs)
        except anthropic.RateLimitError:
            if attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            print(f"レート制限。{delay:.1f}秒後にリトライ ({attempt+1}/{max_retries})")
            time.sleep(delay)
        except anthropic.APIStatusError as e:
            if e.status_code == 529:  # overloaded
                if attempt == max_retries - 1:
                    raise
                delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                time.sleep(delay)
            else:
                raise

importanthropic
importlogging

logger = logging.getLogger(__name__)

defsafe_call(client: anthropic.Anthropic, **kwargs) -> str | None:
    try:
        response = call_with_backoff(client, **kwargs)
        return response.content[0].text
    except anthropic.AuthenticationError:
        logger.error("APIキーが無効です。ANTHROPIC_API_KEYを確認してください。")
        return None
    except anthropic.BadRequestError as e:
        logger.error(f"リクエストエラー:{e.message}")
        return None
    except anthropic.RateLimitError:
        logger.error("レート制限を超過しました。リトライ上限に達しました。")
        return None
    except anthropic.APIConnectionError:
        logger.error("API接続エラー。ネットワークを確認してください。")
        return None
    except anthropic.APIStatusError as e:
        logger.error(f"APIエラー:{e.status_code}{e.message}")
        return None

response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "こんにちは"}]
)

print(f"入力トークン:{response.usage.input_tokens}")
print(f"出力トークン:{response.usage.output_tokens}")

# キャッシュ使用時
if response.usage.cache_creation_input_tokens:
    print(f"キャッシュ作成:{response.usage.cache_creation_input_tokens}")
if response.usage.cache_read_input_tokens:
    print(f"キャッシュ読み取り:{response.usage.cache_read_input_tokens}")

fromdataclassesimport dataclass, field

PRICING = {
    "claude-opus-4-6": {"input": 5.0, "output": 25.0},
    "claude-sonnet-4-5": {"input": 3.0, "output": 15.0},
    "claude-haiku-4-5": {"input": 1.0, "output": 5.0},
}

@dataclass
classCostTracker:
    total_input_tokens: int = 0
    total_output_tokens: int = 0
    total_cost_usd: float = 0.0
    request_count: int = 0

    deftrack(self, model: str, usage) -> float:
        pricing = PRICING.get(model, PRICING["claude-sonnet-4-5"])
        input_cost = (usage.input_tokens / 1_000_000) * pricing["input"]
        output_cost = (usage.output_tokens / 1_000_000) * pricing["output"]
        cost = input_cost + output_cost

        self.total_input_tokens += usage.input_tokens
        self.total_output_tokens += usage.output_tokens
        self.total_cost_usd += cost
        self.request_count += 1

        return cost

    defreport(self) -> str:
        return (
            f"リクエスト数:{self.request_count}\n"
            f"入力トークン:{self.total_input_tokens:,}\n"
            f"出力トークン:{self.total_output_tokens:,}\n"
            f"合計コスト: ${self.total_cost_usd:.4f}"
        )

importlogging

logger = logging.getLogger(__name__)

classBudgetGuard:
    def__init__(self, daily_limit_usd: float = 10.0):
        self.daily_limit = daily_limit_usd
        self.tracker = CostTracker()

    defcheck(self, model: str, usage) -> bool:
        cost = self.tracker.track(model, usage)
        if self.tracker.total_cost_usd > self.daily_limit:
            logger.warning(
                f"日次予算超過: ${self.tracker.total_cost_usd:.2f} / ${self.daily_limit:.2f}"
            )
            return False  # 予算超過
        return True

response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "あなたは顧客サポートアシスタントです。... (長いシステムプロンプト)",
            "cache_control": {"type": "ephemeral"}
        }
    ],
    messages=[{"role": "user", "content": user_input}]
)

defsanitize_input(text: str) -> str:
    # 極端に長い入力を制限
    if len(text) > 10000:
        text = text[:10000]

    return text

# プロンプト構築時
user_message = sanitize_input(raw_input)
response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    system="あなたはカスタマーサポートです。ユーザーの質問に答えてください。",
    messages=[
        {"role": "user", "content": f"<user_input>{user_message}</user_input>"}
    ]
)

importjson
importtime
importlogging

logger = logging.getLogger(__name__)

deflogged_call(client, **kwargs):
    start = time.time()
    model = kwargs.get("model", "unknown")

    try:
        response = client.messages.create(**kwargs)
        elapsed = time.time() - start

        logger.info(json.dumps({
            "event": "api_call",
            "model": model,
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens,
            "latency_sec": round(elapsed, 2),
            "stop_reason": response.stop_reason,
            "status": "success",
        }))

        return response

    except Exception as e:
        elapsed = time.time() - start
        logger.error(json.dumps({
            "event": "api_call",
            "model": model,
            "latency_sec": round(elapsed, 2),
            "status": "error",
            "error_type": type(e).__name__,
            "error_message": str(e),
        }))
        raise