AI API コスト治理：Token 维度告警・部門分账・超额熔断の完全実装ガイド

Token 维度告警の実装

コスト управления の第一歩は「今どの程度のトークンを消費しているか」を可視化することだ。HolySheep AI の API を使ってリアルタイムでトークン使用量を監視し、しきい値を超えた場合に通知を送るシステムを作る。

import httpx
import json
from datetime import datetime, timedelta
from typing import Optional
import asyncio

class HolySheepTokenMonitor:
    """
    HolySheep AI API のトークン使用量を監視し、
    しきい値超過時にアラートを発生させるクラス
    """
    
    def __init__(self, api_key: str):
        self.base_url = "https://api.holysheep.ai/v1"
        self.api_key = api_key
        self.alert_thresholds = {
            "minute": 100_000,      # 1分あたりのトークン上限
            "hourly": 2_000_000,    # 1時間あたりのトークン上限
            "daily": 10_000_000,    # 1日あたりのトークン上限
        }
        self.department_budgets = {
            "engineering": 5_000_000,
            "marketing": 2_000_000,
            "support": 1_500_000,
            "data_science": 1_500_000,
        }
        self.usage_cache = {"minute": {}, "hourly": {}, "daily": {}}
    
    async def fetch_usage_stats(self) -> dict:
        """現在の使用量統計を取得"""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/usage/stats",
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                timeout=30.0
            )
            
            if response.status_code == 401:
                raise ConnectionError(
                    "401 Unauthorized: APIキーが無効です。"
                    "https://www.holysheep.ai/register で新しいキーを発行してください。"
                )
            elif response.status_code == 429:
                raise ConnectionError(
                    "429 Too Many Requests: レートリミットに達しました。"
                    "30秒後に再試行してください。"
                )
            elif response.status_code != 200:
                raise ConnectionError(
                    f"API Error {response.status_code}: {response.text}"
                )
            
            return response.json()
    
    def calculate_cost(self, tokens: int, model: str) -> float:
        """トークン数とモデルからコストを算出（USD）"""
        prices_per_million = {
            "gpt-4.1": 8.00,
            "claude-sonnet-4.5": 15.00,
            "gemini-2.5-flash": 2.50,
            "deepseek-v3.2": 0.42,
        }
        price = prices_per_million.get(model, 10.00)
        return (tokens / 1_000_000) * price
    
    async def check_token_alerts(self) -> list[dict]:
        """しきい値を超過したトークン使用があるかチェック"""
        alerts = []
        stats = await self.fetch_usage_stats()
        
        current_minute_usage = stats.get("tokens_this_minute", 0)
        current_hourly_usage = stats.get("tokens_this_hour", 0)
        current_daily_usage = stats.get("tokens_today", 0)
        
        if current_minute_usage > self.alert_thresholds["minute"]:
            alerts.append({
                "severity": "CRITICAL",
                "level": "minute",
                "usage": current_minute_usage,
                "threshold": self.alert_thresholds["minute"],
                "message": f"1分あたりトークン使用량이閾値を超過: "
                          f"{current_minute_usage:,} > {self.alert_thresholds['minute']:,}"
            })
        
        if current_hourly_usage > self.alert_thresholds["hourly"]:
            alerts.append({
                "severity": "HIGH",
                "level": "hourly",
                "usage": current_hourly_usage,
                "threshold": self.alert_thresholds["hourly"],
                "message": f"時間あたりトークン使用量が閾値を超過: "
                          f"{current_hourly_usage:,} > {self.alert_thresholds['hourly']:,}"
            })
        
        if current_daily_usage > self.alert_thresholds["daily"]:
            alerts.append({
                "severity": "WARNING",
                "level": "daily",
                "usage": current_daily_usage,
                "threshold": self.alert_thresholds["daily"],
                "message": f"日次トークン使用量が閾値を超過: "
                          f"{current_daily_usage:,} > {self.alert_thresholds['daily']:,}"
            })
        
        return alerts
    
    async def monitor_loop(self, interval: int = 60):
        """継続的な監視ループ"""
        print(f"[{datetime.now()}] HolySheep AI トークン監視開始")
        
        while True:
            try:
                alerts = await self.check_token_alerts()
                
                for alert in alerts:
                    print(f"[ALERT] {alert['severity']}: {alert['message']}")
                    await self.send_alert_notification(alert)
                
                stats = await self.fetch_usage_stats()
                estimated_cost = self.calculate_cost(
                    stats.get("tokens_today", 0),
                    stats.get("primary_model", "deepseek-v3.2")
                )
                print(f"[INFO] 本日使用: {stats.get('tokens_today', 0):,} tokens "
                      f"(推定コスト: ${estimated_cost:.2f})")
                
            except ConnectionError as e:
                print(f"[ERROR] {str(e)}")
                await asyncio.sleep(60)
            
            await asyncio.sleep(interval)
    
    async def send_alert_notification(self, alert: dict):
        """アラート通知の送信（Slack / Email / WeChat）"""
        # 実際の実装ではSlack webhookやメール送信処理をここに記述
        print(f"[NOTIFICATION] Alert sent: {alert}")


使用例
monitor = HolySheepTokenMonitor(api_key="YOUR_HOLYSHEEP_API_KEY")
asyncio.run(monitor.monitor_loop(interval=60))

部門別コスト分账システム

企业中ではEngineering、Marketing、Support など多个部门がAI APIを共有利用することが多い。部门ごとにコストを正確に分配し、各部门的API使用状況を見える化することは、コスト治理の核心だ。

import httpx
import json
from datetime import datetime
from collections import defaultdict
from typing import Dict, List
from dataclasses import dataclass

@dataclass
class DepartmentCost:
    department: str
    total_tokens: int
    request_count: int
    estimated_cost_usd: float
    budget_limit: int
    utilization_rate: float
    top_models: List[tuple]

class HolySheepDepartmentBilling:
    """
    部門別のコスト分配账を行うクラス
    HolySheep AI の API を使用して部門ごとの使用量を追跡
    """
    
    def __init__(self, api_key: str):
        self.base_url = "https://api.holysheep.ai/v1"
        self.api_key = api_key
        self.departments = {
            "engineering": {"budget": 5_000_000, "cost_center": "CC-ENG-001"},
            "marketing": {"budget": 2_000_000, "cost_center": "CC-MKT-002"},
            "support": {"budget": 1_500_000, "cost_center": "CC-SUP-003"},
            "data_science": {"budget": 1_500_000, "cost_center": "CC-DS-004"},
        }
    
    async def fetch_detailed_usage(self, start_date: str, end_date: str) -> dict:
        """期間中の詳細使用量を取得"""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/usage/detailed",
                params={
                    "start_date": start_date,
                    "end_date": end_date,
                },
                headers={
                    "Authorization": f"Bearer {self.api_key}",
                    "Content-Type": "application/json"
                },
                timeout=30.0
            )
            
            if response.status_code == 401:
                raise ConnectionError(
                    "401 Unauthorized: 部門別請求のため有効なAPIキーが必要です。"
                    "https://www.holysheep.ai/register で確認してください。"
                )
            elif response.status_code == 403:
                raise ConnectionError(
                    "403 Forbidden: このAPIキーには使用量詳細へのアクセス権がありません。"
                )
            
            return response.json()
    
    def assign_department(self, request_metadata: dict) -> str:
        """リクエストメタデータから部門を特定"""
        project_tags = request_metadata.get("tags", [])
        
        tag_to_dept = {
            "project:chatbot": "engineering",
            "project:content-gen": "marketing",
            "project:customer-support": "support",
            "project:analytics": "data_science",
            "team:backend": "engineering",
            "team:frontend": "engineering",
            "team:copy": "marketing",
        }
        
        for tag in project_tags:
            if tag in tag_to_dept:
                return tag_to_dept[tag]
        
        return request_metadata.get("department", "unassigned")
    
    async def generate_department_report(
        self,
        start_date: str,
        end_date: str
    ) -> List[DepartmentCost]:
        """部門別のコストレポートを生成"""
        usage_data = await self.fetch_detailed_usage(start_date, end_date)
        
        department_usage = defaultdict(lambda: {
            "tokens": 0,
            "requests": 0,
            "cost": 0.0,
            "models": defaultdict(int),
        })
        
        model_prices = {
            "gpt-4.1": 8.00,
            "claude-sonnet-4.5": 15.00,
            "gemini-2.5-flash": 2.50,
            "deepseek-v3.2": 0.42,
        }
        
        for request in usage_data.get("requests", []):
            dept = self.assign_department(request.get("metadata", {}))
            
            tokens = request.get("tokens_used", 0)
            model = request.get("model", "deepseek-v3.2")
            price = model_prices.get(model, 10.00)
            cost = (tokens / 1_000_000) * price
            
            department_usage[dept]["tokens"] += tokens
            department_usage[dept]["requests"] += 1
            department_usage[dept]["cost"] += cost
            department_usage[dept]["models"][model] += tokens
        
        reports = []
        for dept_name, usage in department_usage.items():
            budget = self.departments.get(dept_name, {}).get("budget", 0)
            utilization = (usage["tokens"] / budget * 100) if budget > 0 else 0
            
            top_models = sorted(
                usage["models"].items(),
                key=lambda x: x[1],
                reverse=True
            )[:3]
            
            report = DepartmentCost(
                department=dept_name,
                total_tokens=usage["tokens"],
                request_count=usage["requests"],
                estimated_cost_usd=usage["cost"],
                budget_limit=budget,
                utilization_rate=utilization,
                top_models=top_models,
            )
            reports.append(report)
        
        return reports
    
    def format_budget_alert(self, report: DepartmentCost) -> str:
        """部門別の予算アラートメッセージを作成"""
        if report.utilization_rate >= 100:
            status = "🔴 超過"
            action = "至急：当月のAPI利用を停止または削減してください"
        elif report.utilization_rate >= 80:
            status = "🟡 警告"
            action = "注意：予算の80%に達しました。月末まで利用を継続しますか？"
        elif report.utilization_rate >= 50:
            status = "🟢 正常"
            action = "順調：予算の半分以上がまだ利用可能です"
        else:
            status = "✅ 余裕"
            action = "快適：予算に十分な余裕があります"
        
        model_info = "\n  ".join([
            f"- {model}: {tokens:,} tokens"
            for model, tokens in report.top_models
        ])
        
        return f"""
部门: {report.department.upper()}
{status} 予算消化率: {report.utilization_rate:.1f}%

📊 今月の使用状況:
  - 総トークン数: {report.total_tokens:,}
  - リクエスト数: {report.request_count:,}
  - 推定コスト: ${report.estimated_cost_usd:.2f}
  - モデル内訳:
  {model_info}

💡 推奨アクション:
{action}
"""

    async def export_to_json(self, reports: List[DepartmentCost]) -> str:
        """レポートをJSON形式にエクスポート"""
        return json.dumps(
            [asdict(r) for r in reports],
            indent=2,
            ensure_ascii=False,
            default=str
        )


使用例
async def main():
    billing = HolySheepDepartmentBilling(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    today = datetime.now().strftime("%Y-%m-%d")
    month_start = datetime.now().replace(day=1).strftime("%Y-%m-%d")
    
    reports = await billing.generate_department_report(
        start_date=month_start,
        end_date=today
    )
    
    for report in reports:
        print(billing.format_budget_alert(report))
        
        if report.utilization_rate >= 100:
            print("⚠️ 予算超過アラートをSlackに送信しました")
    
    # JSON エクスポート
    json_report = await billing.export_to_json(reports)
    print("\nJSON レポート:")
    print(json_report[:500] + "..." if len(json_report) > 500 else json_report)

asyncio.run(main())

超额熔断机制の実装

冒頭のストーリーのように、無限ループや误ったbatch处理 导致 超過が発生した場合、自动的にAPIコールを停止する熔断机制は不可欠だ。以下に、部门별・プロジェクト別の熔断実装を示す。

import asyncio
import time
from datetime import datetime, timedelta
from enum import Enum
from typing import Dict, Optional
from dataclasses import dataclass, field
import httpx

class CircuitState(Enum):
    CLOSED = "closed"      # 正常状態：リクエスト許可
    OPEN = "open"          # 開放状態：リクエスト 차단
    HALF_OPEN = "half_open"  # 半開状態：試験的にリクエスト許可

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5        # OPENにする失敗回数
    success_threshold: int = 3         # CLOSEDに戻す成功回数
    timeout: int = 300                 # OPEN状态的継続時間（秒）
    rate_limit_per_minute: int = 1000  # 1分あたりのリクエスト上限
    budget_monthly_usd: float = 100.0  # 月額予算上限

@dataclass
class CircuitBreaker:
    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    success_count: int = 0
    last_failure_time: Optional[datetime] = None
    opened_at: Optional[datetime] = None
    request_history: list = field(default_factory=list)
    
    def record_request(self, success: bool, tokens_used: int = 0):
        """リクエスト結果を記録"""
        now = datetime.now()
        self.request_history.append({
            "timestamp": now,
            "success": success,
            "tokens": tokens_used,
        })
        
        # 1分以内の履歴のみ保持
        cutoff = now - timedelta(minutes=1)
        self.request_history = [
            r for r in self.request_history if r["timestamp"] > cutoff
        ]
        
        if success:
            self.success_count += 1
            if self.state == CircuitHalfOpenCircuitBreaker.HALF_OPEN:
                if self.success_count >= self.success_threshold:
                    self.state = CircuitState.CLOSED
                    self.success_count = 0
                    print(f"[CircuitBreaker] 恢复 CLOSED 状态")
        else:
            self.failure_count += 1
            self.last_failure_time = now
            if self.state == CircuitState.CLOSED:
                if self.failure_count >= failure_threshold:
                    self.state = CircuitState.OPEN
                    self.opened_at = now
                    print(f"[CircuitBreaker] 开放 OPEN 状态 - 失败次数过多")
    
    def can_request(self) -> bool:
        """リクエスト可能かチェック"""
        if self.state == CircuitState.OPEN:
            if self.opened_at:
                elapsed = (datetime.now() - self.opened_at).total_seconds()
                if elapsed >= self.timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.failure_count = 0
                    print(f"[CircuitBreaker] 转为 HALF_OPEN 状态 - 试验请求")
                    return True
            return False
        
        # レートリミットチェック
        recent_requests = len([
            r for r in self.request_history
            if (datetime.now() - r["timestamp"]).total_seconds() < 60
        ])
        return recent_requests < self.rate_limit_per_minute
    
    def get_stats(self) -> dict:
        return {
            "state": self.state.value,
            "failure_count": self.failure_count,
            "success_count": self.success_count,
            "requests_last_minute": len(self.request_history),
            "total_tokens_today": sum(r["tokens"] for r in self.request_history),
        }


class CircuitHalfOpenCircuitBreaker:
    """メインビジネスロジックと統合された熔断机制"""
    
    def __init__(self, api_key: str, config: CircuitBreakerConfig):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.config = config
        self.circuit_breakers: Dict[str, CircuitBreaker] = {}
        self.monthly_spend = 0.0
        self.month_start = datetime.now().replace(day=1)
        
        # デフォルトの熔断器を作成
        self.circuit_breakers["default"] = CircuitBreaker(
            rate_limit_per_minute=config.rate_limit_per_minute
        )
    
    def get_breaker(self, department: str) -> CircuitBreaker:
        """部门別の熔断器を取得または作成"""
        if department not in self.circuit_breakers:
            self.circuit_breakers[department] = CircuitBreaker(
                rate_limit_per_minute=self.config.rate_limit_per_minute
            )
        return self.circuit_breakers[department]
    
    async def api_request(
        self,
        department: str,
        model: str,
        prompt: str,
        max_tokens: int = 1000,
    ) -> dict:
        """熔断机制を備えたAPIリクエスト"""
        breaker = self.get_breaker(department)
        
        # 月額予算チェック
        if self.monthly_spend >= self.config.budget_monthly_usd:
            print(f"[CircuitBreaker] 月额予算超过 ${self.monthly_spend:.2f}")
            raise ConnectionError(
                f"月度预算已超支 (${self.monthly_spend:.2f} / ${self.config.budget_monthly_usd})"
            )
        
        # 熔断器チェック
        if not breaker.can_request():
            raise ConnectionError(
                f"Circuit breaker OPEN for department '{department}'. "
                f"请稍后重试或在 {self.base_url}/dashboard 确认。"
            )
        
        # コスト试算
        estimated_cost = (max_tokens / 1_000_000) * {
            "gpt-4.1": 8.00,
            "claude-sonnet-4.5": 15.00,
            "gemini-2.5-flash": 2.50,
            "deepseek-v3.2": 0.42,
        }.get(model, 10.00)
        
        try:
            async with httpx.AsyncClient() as client:
                response = await client.post(
                    f"{self.base_url}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {self.api_key}",
                        "Content-Type": "application/json",
                        "X-Department": department,
                        "X-Request-ID": f"{department}-{int(time.time())}",
                    },
                    json={
                        "model": model,
                        "messages": [{"role": "user", "content": prompt}],
                        "max_tokens": max_tokens,
                    },
                    timeout=60.0,
                )
                
                if response.status_code == 200:
                    data = response.json()
                    actual_tokens = data.get("usage", {}).get("total_tokens", max_tokens)
                    breaker.record_request(success=True, tokens_used=actual_tokens)
                    
                    # コスト累積
                    actual_cost = (actual_tokens / 1_000_000) * estimated_cost
                    self.monthly_spend += actual_cost
                    
                    return {"status": "success", "data": data}
                
                elif response.status_code == 401:
                    breaker.record_request(success=False)
                    raise ConnectionError(
                        "401 Unauthorized: API 密钥无效"
                    )
                elif response.status_code == 429:
                    breaker.record_request(success=False)
                    raise ConnectionError(
                        "429 Rate Limited: 请求过于频繁"
                    )
                else:
                    breaker.record_request(success=False)
                    raise ConnectionError(
                        f"API Error {response.status_code}: {response.text}"
                    )
                    
        except httpx.TimeoutException:
            breaker.record_request(success=False)
            raise ConnectionError(
                "Request timeout: API响应超时"
            )


使用例
async def main():
    config = CircuitBreakerConfig(
        failure_threshold=5,
        success_threshold=3,
        timeout=300,
        rate_limit_per_minute=100,
        budget_monthly_usd=100.0,
    )
    
    client = CircuitHalfOpenCircuitBreaker(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        config=config,
    )
    
    # 部门別のリクエスト例
    departments = ["engineering", "marketing", "support"]
    
    for dept in departments:
        try:
            result = await client.api_request(
                department=dept,
                model="deepseek-v3.2",
                prompt="简短的问候语",
                max_tokens=50,
            )
            print(f"[{dept}] Success: {result['status']}")
            
        except ConnectionError as e:
            print(f"[{dept}] Failed: {str(e)}")
    
    # 熔断器状态確認
    print("\n熔断器状态:")
    for dept, breaker in client.circuit_breakers.items():
        stats = breaker.get_stats()
        print(f"  {dept}: {stats}")

asyncio.run(main())

よくあるエラーと対処法

エラー1: 401 Unauthorized — API キーが無効

エラー内容：

ConnectionError: 401 Unauthorized: APIキーが無効です。
https://www.holysheep.ai/register で新しいキーを発行してください。

原因： API キーが期限切れ、有効期限内でもスコープが合わない、または貼り付け時に空白が混入している場合に発生。

解決方法：

# API キーの再確認と再設定
import os

環境変数からAPIキーを安全に設定
api_key = os.environ.get("HOLYSHEEP_API_KEY")

if not api_key or not api_key.startswith("hs_"):
    # 新しいキーを https://www.holysheep.ai/register から取得
    raise ValueError(
        "Invalid API key format. Please obtain a valid key from "
        "https://www.holysheep.ai/register"
    )

キーの先頭を検証（実際の形式に合わせて調整）
print(f"Using API key: {api_key[:8]}...{api_key[-4:]}")

エラー2: 429 Rate Limited — レートリミット超過

エラー内容：

ConnectionError: 429 Too Many Requests: レートリミットに達しました。
30秒後に再試行してください。

原因： 短時間内に大量のリクエストを送信した場合、または月間配额を使い切った場合に発生。前のセクションで実装した熔断器が正しく機能していない可能性がある。

解決方法：

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

class RateLimitHandler:
    def __init__(self, max_retries: int = 3, base_delay: int = 30):
        self.max_retries = max_retries
        self.base_delay = base_delay
    
    async def execute_with_retry(self, func, *args, **kwargs):
        """指数バックオフでリトライを実行"""
        for attempt in range(self.max_retries):
            try:
                return await func(*args, **kwargs)
            except ConnectionError as e:
                if "429" in str(e) and attempt < self.max_retries - 1:
                    delay = self.base_delay * (2 ** attempt)
                    print(f"Rate limit hit. Retrying in {delay} seconds...")
                    await asyncio.sleep(delay)
                else:
                    raise
        
        raise ConnectionError(
            f"Rate limit exceeded after {self.max_retries} attempts. "
            "Consider upgrading your plan at https://www.holysheep.ai/dashboard"
        )

使用例
handler = RateLimitHandler(max_retries=3, base_delay=30)

async def call_api():
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.holysheep.ai/v1/chat/completions",
            headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"},
            json={"model": "deepseek-v3.2", "messages": [{"role": "user", "content": "test"}]},
            timeout=60.0,
        )
        return response.json()

result = await handler.execute_with_retry(call_api)

エラー3: Budget Exceeded — 月額予算超過

エラー内容：

ConnectionError: 月额予算已超支 ($105.50 / $100.00)

原因： 設定した月額予算的上限に達した。前述の部門別レポートで部门ごとに予assanを守らなかった場合に發生。

解決方法：

import asyncio
from datetime import datetime

class BudgetController:
    def __init__(self, monthly_limit_usd: float = 100.0):
        self.monthly_limit = monthly_limit_usd
        self.current_spend = 0.0
        self.reset_date = datetime.now().replace(day=1)
    
    def check_and_update_budget(self, cost: float) -> bool:
        """予算をチェックし、超過の場合はリクエストを 차단"""
        # 月またぎの場合、リセット
        now = datetime.now()
        if now.day == 1 and now.month != self.reset_date.month:
            self.current_spend = 0.0
            self.reset_date = now.replace(day=1)
            print(f"[BudgetController] 月额预算已重置")
        
        if self.current_spend + cost > self.monthly_limit:
            remaining = self.monthly_limit - self.current_spend
            raise ConnectionError(
                f"月度预算超过: 剩余 ${remaining:.2f}，请求费用 ${cost:.2f}"
                f"请升级计划或等待下月重置: https://www.holysheep.ai/dashboard"
            )
        
        self.current_spend += cost
        print(f"[BudgetController] 消费 ${self.current_spend:.2f} / ${self.monthly_limit:.2f}")
        return True
    
    def get_budget_status(self) -> dict:
        remaining = max(0, self.monthly_limit - self.current_spend)
        percentage = (self.current_spend / self.monthly_limit) * 100
        return {
            "spent": self.current_spend,
            "limit": self.monthly_limit,
            "remaining": remaining,
            "percentage": percentage,
            "reset_date": self.reset_date.strftime("%Y-%m-%d"),
        }

使用例
controller = BudgetController(monthly_limit_usd=100.0)

API呼び出し前に必ずチェック
try:
    estimated_cost = 0.42 / 1_000_000 * 1000  # DeepSeek V3.2 で 1000 tokens
    controller.check_and_update_budget(estimated_cost)
    print("リクエスト許可")
except ConnectionError as e:
    print(f"リクエスト拒否: {e}")

print(f"\n当前预算状态: {controller.get_budget_status()}")

ダッシュボードでの確認と管理