AI APIの監視と費用管理

AIエージェントを24時間稼働させると、APIクレジットは驚くほど速く消費される。クレジット枯渇でBot全停止という事態を二度と起こさないために、監視・アラート・自動フォールバックの三層防御を構築する。

実体験: 36時間の全停止

APIクレジット不足でBot・note・visual_inspector全てが停止した。復旧までの36時間、APIクレジット不要の改善に集中して50本の記事を制作したが、自動化パイプラインの単一障害点がAPI残高であることを痛感した。この経験から、残高監視システムの構築が最優先事項になった。

API使用量の取得方法

Anthropic APIの使用量取得

Anthropic APIではレスポンスヘッダーからトークン消費量を取得できる。

import anthropic
import json
from datetime import datetime

client = anthropic.Anthropic()

def track_usage(prompt: str, model: str = "claude-sonnet-4-20250514") -> dict:
    """API呼び出しとトークン消費量の記録"""
    response = client.messages.create(
        model=model,
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    usage = {
        "timestamp": datetime.now().isoformat(),
        "model": model,
        "input_tokens": response.usage.input_tokens,
        "output_tokens": response.usage.output_tokens,
        "cost_usd": estimate_cost(
            model, response.usage.input_tokens, response.usage.output_tokens
        )
    }
    save_usage_log(usage)
    return usage

def estimate_cost(model: str, input_tok: int, output_tok: int) -> float:
    """モデル別のコスト概算"""
    rates = {
        "claude-sonnet-4-20250514": {"input": 3.0, "output": 15.0},
        "claude-haiku-35-20241022": {"input": 0.25, "output": 1.25},
        "claude-opus-4-20250514": {"input": 15.0, "output": 75.0},
    }
    r = rates.get(model, rates["claude-sonnet-4-20250514"])
    return (input_tok * r["input"] + output_tok * r["output"]) / 1_000_000

OpenAI APIの使用量取得

from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}]
)
print(f"Input: {response.usage.prompt_tokens}")
print(f"Output: {response.usage.completion_tokens}")
print(f"Total: {response.usage.total_tokens}")

残高監視とアラートシステム

Discord Webhookで残高アラート

残高が閾値を下回ったらDiscordに即通知する。

import urllib.request
import json

WEBHOOK_URL = "https://discord.com/api/webhooks/YOUR_WEBHOOK_ID/YOUR_TOKEN"
DAILY_BUDGET_USD = 5.0
WARNING_THRESHOLD = 0.8  # 80%で警告

def check_daily_budget():
    """日次予算の消費チェック"""
    today = datetime.now().date().isoformat()
    daily_cost = sum(
        entry["cost_usd"]
        for entry in load_usage_entries()
        if entry["timestamp"].startswith(today)
    )
    ratio = daily_cost / DAILY_BUDGET_USD

    if ratio >= 1.0:
        send_discord_alert(
            f"日次予算超過: ${daily_cost:.2f} / ${DAILY_BUDGET_USD:.2f}",
            color=0xFF0000
        )
        return "OVER_BUDGET"
    elif ratio >= WARNING_THRESHOLD:
        send_discord_alert(
            f"予算警告: ${daily_cost:.2f} / ${DAILY_BUDGET_USD:.2f} ({ratio*100:.0f}%)",
            color=0xFFAA00
        )
        return "WARNING"
    return "OK"

def send_discord_alert(message: str, color: int = 0x00FF00):
    """Discord Webhookにアラート送信"""
    payload = json.dumps({
        "embeds": [{
            "title": "API費用モニター",
            "description": message,
            "color": color,
            "timestamp": datetime.now().isoformat()
        }]
    }).encode()
    req = urllib.request.Request(
        WEBHOOK_URL, data=payload,
        headers={"Content-Type": "application/json"}
    )
    urllib.request.urlopen(req)

自動フォールバック戦略

高コストモデルで予算を超過しそうなとき、自動的に安価なモデルに切り替える。

段階的モデルフォールバック

class ModelRouter:
    """予算状況に応じたモデル自動選択"""

    FALLBACK_CHAIN = [
        {"model": "claude-sonnet-4-20250514", "max_daily_usd": 5.0},
        {"model": "claude-haiku-35-20241022", "max_daily_usd": 1.0},
    ]

    def __init__(self):
        self.daily_cost = 0.0

    def select_model(self) -> str:
        """現在のコスト状況に応じてモデルを選択"""
        for tier in self.FALLBACK_CHAIN:
            if self.daily_cost < tier["max_daily_usd"]:
                return tier["model"]
        return self.FALLBACK_CHAIN[-1]["model"]

    def call_with_fallback(self, prompt: str) -> str:
        """フォールバック付きAPI呼び出し"""
        model = self.select_model()
        try:
            result = track_usage(prompt, model=model)
            self.daily_cost += result["cost_usd"]
            return result
        except anthropic.RateLimitError:
            fallback = self.FALLBACK_CHAIN[-1]["model"]
            return track_usage(prompt, model=fallback)

APIなしの代替フロー

APIクレジットが完全に枯渇した場合でも業務を継続するための代替フローを常備する。

ローカルLLMへの切り替え

// Node.js: Ollama経由でローカルモデルにフォールバック
const fetch = require('node-fetch');

async function localFallback(prompt) {
  const response = await fetch('http://localhost:11434/api/generate', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      model: 'gemma:7b',
      prompt: prompt,
      stream: false
    })
  });
  const data = await response.json();
  return data.response;
}

API不要作業の自動切り替え

クレジット枯渇検知時に、API不要タスクに自動切り替えるスケジューラ。

def get_next_task(api_available: bool) -> dict:
    """API可否に応じたタスク選択"""
    if api_available:
        return pick_from_queue("all_tasks")
    else:
        # API不要タスク: リファクタリング、テスト追加、ドキュメント整理
        return pick_from_queue("no_api_tasks")

日次コストレポートの自動生成

毎日のAPI消費状況をサマリーして記録する。

def generate_daily_report() -> str:
    """日次コストレポート生成"""
    today = datetime.now().date().isoformat()
    entries = load_usage_entries(today)

    by_model = {}
    for e in entries:
        m = e["model"]
        if m not in by_model:
            by_model[m] = {"calls": 0, "cost": 0.0, "tokens": 0}
        by_model[m]["calls"] += 1
        by_model[m]["cost"] += e["cost_usd"]
        by_model[m]["tokens"] += e["input_tokens"] + e["output_tokens"]

    report_lines = [f"# API Cost Report: {today}\n"]
    total_cost = 0.0
    for model, stats in by_model.items():
        report_lines.append(
            f"- **{model}**: {stats['calls']}回, "
            f"{stats['tokens']:,}トークン, ${stats['cost']:.3f}"
        )
        total_cost += stats["cost"]
    report_lines.append(f"\n**合計: ${total_cost:.3f}**")
    return "\n".join(report_lines)

まとめ: 三層防御で枯渇を防ぐ

層	対策	効果
監視	トークン消費の記録と日次レポート	異常検知
アラート	Discord/Slack通知と閾値設定	即時対応
自動フォールバック	モデル切替とローカルLLM	無停止運用

この三層防御により、APIクレジット枯渇による全停止を防ぎ、24時間の自動運用を維持できる。

AI APIの監視と費用管理 — クレジット枯渇を二度と起こさない