AI Agent を本番環境に展開する際、最大の問題の一つが複数のLLM API を一元管理することです。私のチームでは以前、GPT-4.1・Claude Sonnet 4.5・Gemini 2.5 Flash・DeepSeek V3.2 を個別に管理していましたが、認証情報の散在、レート制限の競合、ログ管理の困難さ、成本可視性の欠如という4つの壁にぶつかりました。本稿では、HolySheep AI の統一APIゲートウェイを活用し、これらの課題を一気に解決したアーキテクチャ設計と実装コードを詳解します。

なぜ統一APIゲートウェイが必要か

AI Agent の本番運用において、API ゲートウェイは以下の4つの層で критическую роль を果たします。

HolySheep はこれら4層を1つのエンドポイント（https://api.holysheep.ai/v1）で提供し、最大85%のコスト削減（¥1=$1 vs 公式¥7.3=$1）を実現します。

アーキテクチャ設計

システム構成図


┌─────────────────────────────────────────────────────────────┐
│                    AI Agent Application                      │
│  (LangChain / LlamaIndex / 自作オーケストレーター)           │
└─────────────────────────┬───────────────────────────────────┘
                          │ HTTPS
                          ▼
┌─────────────────────────────────────────────────────────────┐
│              HolySheep API Gateway                           │
│  https://api.holysheep.ai/v1                                │
├─────────────────────────────────────────────────────────────┤
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐   │
│  │Auth Layer│  │Rate Limit│  │  Logging │  │  Budget  │   │
│  │          │  │  Layer   │  │  Layer   │  │  Layer   │   │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘   │
└───────┼─────────────┼─────────────┼─────────────┼─────────┘
        │             │             │             │
        ▼             ▼             ▼             ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│   GPT-4.1    │ │   Claude     │ │   Gemini     │ │  DeepSeek
│   $8/MTok    │ │ 4.5 $15/MTok│ │2.5 $2.50/MTok│ │ V3.2 $0.42
└──────────────┘ └──────────────┘ └──────────────┘ └──────────

コスト比較表

モデル 公式価格 ($/MTok出力) HolySheep ($/MTok出力) 節約率 推奨ユースケース
GPT-4.1 $60.00 $8.00 87% OFF 複雑な推論・コード生成
Claude Sonnet 4.5 $75.00 $15.00 80% OFF 長文生成・分析タスク
Gemini 2.5 Flash $10.00 $2.50 75% OFF 高速応答・了大量処理
DeepSeek V3.2 $2.80 $0.42 85% OFF コスト重視の通常処理

実装コード：統一認証とモデル選択

まず、HolySheep API への接続を確立する基本クライアントを実装します。HolySheep は OpenAI 互換の API 構造を採用しているため、既存の LangChain や LlamaIndex との統合が容易です。

"""
HolySheep AI - 統一APIゲートウェイクライアント
Production-ready AI Agent 用基盤クラス
"""

import os
from typing import Optional, Dict, Any, List
from dataclasses import dataclass
from datetime import datetime
import hashlib
import hmac
import json

import httpx


@dataclass
class ModelConfig:
    """モデル別の設定"""
    name: str
    provider: str  # openai, anthropic, google, deepseek
    max_tokens: int
    temperature: float
    cost_per_1k_output: float  # USD


class HolySheepGateway:
    """
    HolySheep API Gateway クライアント
    
    特徴：
    - 統一エンドポイント（api.holysheep.ai/v1）
    - 自動モデルルーティング
    - コスト追跡機能
    - 予算アラート対応
    """
    
    BASE_URL = "https://api.holysheep.ai/v1"
    
    # 対応モデル定義
    MODELS = {
        "gpt-4.1": ModelConfig(
            name="gpt-4.1",
            provider="openai",
            max_tokens=128000,
            temperature=0.7,
            cost_per_1k_output=0.008  # $8/MTok
        ),
        "claude-sonnet-4.5": ModelConfig(
            name="claude-sonnet-4.5",
            provider="anthropic",
            max_tokens=200000,
            temperature=0.7,
            cost_per_1k_output=0.015  # $15/MTok
        ),
        "gemini-2.5-flash": ModelConfig(
            name="gemini-2.5-flash",
            provider="google",
            max_tokens=128000,
            temperature=0.7,
            cost_per_1k_output=0.0025  # $2.50/MTok
        ),
        "deepseek-v3.2": ModelConfig(
            name="deepseek-v3.2",
            provider="deepseek",
            max_tokens=128000,
            temperature=0.7,
            cost_per_1k_output=0.00042  # $0.42/MTok
        ),
    }
    
    def __init__(self, api_key: str):
        """
        初期化
        
        Args:
            api_key: HolySheep APIキー
        """
        self.api_key = api_key
        self.client = httpx.Client(
            base_url=self.BASE_URL,
            headers={
                "Authorization": f"Bearer {api_key}",
                "Content-Type": "application/json"
            },
            timeout=60.0
        )
        
        # コスト追跡
        self.total_spent = 0.0
        self.request_count = 0
        
    def chat_completion(
        self,
        model: str,
        messages: List[Dict[str, str]],
        temperature: Optional[float] = None,
        max_tokens: Optional[int] = None,
        **kwargs
    ) -> Dict[str, Any]:
        """
        チャット補完リクエスト
        
        Args:
            model: モデル名（gpt-4.1, claude-sonnet-4.5, etc.）
            messages: メッセージリスト
            temperature: 生成温度
            max_tokens: 最大トークン数
            **kwargs: 追加パラメータ
        
        Returns:
            APIレスポンス（OpenAI互換形式）
        """
        if model not in self.MODELS:
            raise ValueError(f"Unknown model: {model}")
        
        config = self.MODELS[model]
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature or config.temperature,
            "max_tokens": max_tokens or config.max_tokens,
            **kwargs
        }
        
        response = self.client.post("/chat/completions", json=payload)
        response.raise_for_status()
        result = response.json()
        
        # コスト計算
        usage = result.get("usage", {})
        output_tokens = usage.get("completion_tokens", 0)
        cost = (output_tokens / 1000) * config.cost_per_1k_output
        self.total_spent += cost
        self.request_count += 1
        
        return result
    
    def get_cost_report(self) -> Dict[str, Any]:
        """コストレポート取得"""
        return {
            "total_spent_usd": round(self.total_spent, 4),
            "total_requests": self.request_count,
            "avg_cost_per_request": round(
                self.total_spent / self.request_count, 6
            ) if self.request_count > 0 else 0,
            "timestamp": datetime.now().isoformat()
        }
    
    def close(self):
        """クライアント終了"""
        self.client.close()



===== 使用例 =====

if __name__ == "__main__":
    # 初期化（YOUR_HOLYSHEEP_API_KEY を実際のキーに置換）
    gateway = HolySheepGateway(api_key="YOUR_HOLYSHEEP_API_KEY")
    
    # DeepSeek V3.2 での低コスト処理
    response = gateway.chat_completion(
        model="deepseek-v3.2",
        messages=[
            {"role": "system", "content": "あなたは有用なアシスタントです。"},
            {"role": "user", "content": "こんにちは！"}
        ],
        temperature=0.7
    )
    
    print(f"Response: {response['choices'][0]['message']['content']}")
    print(f"Cost Report: {gateway.get_cost_report()}")
    
    gateway.close()

実装コード：レート制限とコスト予算管理

本番環境では、レート制限とコスト予算の厳格な管理が不可欠です。以下のクラスは、TPM（Tokens Per Minute）ベースのスロットリングと、日次/月次のコスト上限を管理します。

"""
HolySheep AI - レート制限・コスト予算管理モジュール
Production AI Agent 向け：高精度トラフィック制御
"""

import asyncio
import time
from typing import Dict, Optional, Callable, Any
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import deque
from enum import Enum
import threading
import logging

logger = logging.getLogger(__name__)


class BudgetAlertLevel(Enum):
    """予算アラートレベル"""
    SAFE = "safe"
    WARNING = "warning"  # 80%到達
    CRITICAL = "critical"  # 95%到達
    EXCEEDED = "exceeded"


@dataclass
class RateLimitConfig:
    """レート制限設定"""
    tpm_limit: int  # Tokens Per Minute
    rpm_limit: int  # Requests Per Minute
    rph_limit: int  # Requests Per Hour
    
    @classmethod
    def from_model(cls, model: str) -> "RateLimitConfig":
        """モデル別のデフォルト制限"""
        limits = {
            "gpt-4.1": cls(tpm_limit=150000, rpm_limit=500, rph_limit=10000),
            "claude-sonnet-4.5": cls(tpm_limit=200000, rpm_limit=400, rph_limit=8000),
            "gemini-2.5-flash": cls(tpm_limit=1000000, rpm_limit=1500, rph_limit=50000),
            "deepseek-v3.2": cls(tpm_limit=500000, rpm_limit=2000, rph_limit=100000),
        }
        return limits.get(model, cls(tpm_limit=100000, rpm_limit=1000, rph_limit=50000))


@dataclass
class BudgetConfig:
    """コスト予算設定"""
    daily_limit_usd: float
    monthly_limit_usd: float
    alert_threshold: float = 0.8  # 80%で警告
    
    def get_alert_level(self, spent: float) -> BudgetAlertLevel:
        ratio = spent / self.daily_limit_usd
        if ratio >= 1.0:
            return BudgetAlertLevel.EXCEEDED
        elif ratio >= 0.95:
            return BudgetAlertLevel.CRITICAL
        elif ratio >= 0.8:
            return BudgetAlertLevel.WARNING
        return BudgetAlertLevel.SAFE


class TokenBucket:
    """トークンバケット方式のレギュレーター"""
    
    def __init__(self, rate: int, capacity: int):
        self.rate = rate  # 秒間補充量
        self.capacity = capacity
        self.tokens = capacity
        self.last_update = time.time()
        self._lock = threading.Lock()
    
    def consume(self, tokens: int, blocking: bool = True) -> bool:
        """
        トークン消費
        
        Args:
            tokens: 消費するトークン数
            blocking: True=利用可能まで待機
        
        Returns:
            True=成功, False=制限超過
        """
        while True:
            with self._lock:
                now = time.time()
                elapsed = now - self.last_update
                self.tokens = min(
                    self.capacity,
                    self.tokens + elapsed * self.rate
                )
                self.last_update = now
                
                if self.tokens >= tokens:
                    self.tokens -= tokens
                    return True
                
                if not blocking:
                    return False
                
                wait_time = (tokens - self.tokens) / self.rate
            
            time.sleep(min(wait_time, 1.0))


class CostBudgetManager:
    """コスト予算管理クラス"""
    
    def __init__(self, config: BudgetConfig):
        self.config = config
        self.daily_spent = 0.0
        self.monthly_spent = 0.0
        self.last_reset = datetime.now()
        self._lock = threading.Lock()
        
        # コールバック
        self._alert_callbacks: list[Callable] = []
        
    def add_alert_callback(self, callback: Callable[[BudgetAlertLevel, float], None]):
        """アラートコールバック登録"""
        self._alert_callbacks.append(callback)
    
    def record_cost(self, amount_usd: float) -> BudgetAlertLevel:
        """コスト記録とアラート判定"""
        with self._lock:
            self.daily_spent += amount_usd
            self.monthly_spent += amount_usd
            
            # 日次リセット
            if (datetime.now() - self.last_reset).days >= 1:
                self.daily_spent = 0.0
                self.last_reset = datetime.now()
            
            level = self.config.get_alert_level(self.daily_spent)
            
            if level != BudgetAlertLevel.SAFE:
                for callback in self._alert_callbacks:
                    try:
                        callback(level, self.daily_spent)
                    except Exception as e:
                        logger.error(f"Alert callback error: {e}")
            
            return level
    
    def check_budget(self) -> bool:
        """予算残余チェック"""
        with self._lock:
            return self.daily_spent < self.config.daily_limit_usd


class HolySheepRateLimitedClient:
    """
    レート制限・予算管理付きHolySheepクライアント
    
    特徴：
    - モデル別のTPM/RPM制限
    - トークンバケット方式の流量制御
    - 日次/月次コスト予算管理
    - アラートコールバック対応
    """
    
    def __init__(
        self,
        api_key: str,
        rate_limits: Optional[Dict[str, RateLimitConfig]] = None,
        budget_config: Optional[BudgetConfig] = None
    ):
        from .gateway import HolySheepGateway
        
        self.gateway = HolySheepGateway(api_key)
        self.rate_limits = rate_limits or {
            model: RateLimitConfig.from_model(model)
            for model in ["gpt-4.1", "claude-sonnet-4.5", "gemini-2.5-flash", "deepseek-v3.2"]
        }
        
        self.budget_manager = CostBudgetManager(
            budget_config or BudgetConfig(
                daily_limit_usd=100.0,
                monthly_limit_usd=2000.0
            )
        )
        
        # トークンバケット（モデル別）
        self.token_buckets: Dict[str, TokenBucket] = {}
        for model, config in self.rate_limits.items():
            self.token_buckets[model] = TokenBucket(
                rate=config.tpm_limit // 60,  # 秒間レート
                capacity=config.tpm_limit // 2  # バケット容量
            )
        
        # リクエスト履歴（rpm/rph計算用）
        self.request_history: Dict[str, deque] = {
            model: deque(maxlen=1000) for model in self.rate_limits.keys()
        }
    
    def _check_request_limit(self, model: str) -> bool:
        """リクエスト数制限チェック"""
        now = time.time()
        history = self.request_history[model]
        
        # 1分以内のリクエスト数を計算
        cutoff_1m = now - 60
        rpm = sum(1 for ts in history if ts > cutoff_1m)
        
        # 1時間以内のリクエスト数を計算
        cutoff_1h = now - 3600
        rph = sum(1 for ts in history for _ in [ts] if ts > cutoff_1h)
        
        config = self.rate_limits[model]
        return rpm < config.rpm_limit and rph < config.rph_limit
    
    def chat_completion(
        self,
        model: str,
        messages: list,
        estimated_tokens: int = 1000,
        **kwargs
    ) -> dict:
        """
        レート制限・予算管理付きチャット補完
        
        Args:
            model: モデル名
            messages: メッセージリスト
            estimated_tokens: 推定出力トークン数
            **kwargs: 追加パラメータ
        """
        # 1. 予算チェック
        if not self.budget_manager.check_budget():
            raise RuntimeError(
                f"Daily budget exceeded: "
                f"${self.budget_manager.daily_spent:.2f} / "
                f"${self.budget_manager.config.daily_limit_usd:.2f}"
            )
        
        # 2. レート制限チェック（トークンバケット）
        bucket = self.token_buckets.get(model)
        if bucket and not bucket.consume(estimated_tokens, blocking=True):
            raise RuntimeError(f"Token rate limit exceeded for {model}")
        
        # 3. リクエスト数制限チェック
        if not self._check_request_limit(model):
            raise RuntimeError(f"Request rate limit exceeded for {model}")
        
        # 4. API呼び出し
        response = self.gateway.chat_completion(model, messages, **kwargs)
        
        # 5. コスト記録
        usage = response.get("usage", {})
        output_tokens = usage.get("completion_tokens", 0)
        cost = (output_tokens / 1000) * self.gateway.MODELS[model].cost_per_1k_output
        alert_level = self.budget_manager.record_cost(cost)
        
        # 6. リクエスト履歴更新
        self.request_history[model].append(time.time())
        
        # 7. レスポンスにメタ情報付与
        response["_meta"] = {
            "cost_usd": cost,
            "total_daily_spent": self.budget_manager.daily_spent,
            "budget_alert_level": alert_level.value,
            "latency_ms": response.get("latency_ms", 0)
        }
        
        return response
    
    def get_status(self) -> Dict[str, Any]:
        """システムステータス取得"""
        return {
            "budget": {
                "daily_spent": self.budget_manager.daily_spent,
                "daily_limit": self.budget_manager.config.daily_limit_usd,
                "monthly_spent": self.budget_manager.monthly_spent,
                "monthly_limit": self.budget_manager.config.monthly_limit_usd,
            },
            "rate_limits": {
                model: {
                    "tpm_limit": config.tpm_limit,
                    "rpm_limit": config.rpm_limit,
                    "current_bucket_tokens": bucket.tokens
                }
                for model, (config, bucket) in enumerate(
                    zip(self.rate_limits.values(), self.token_buckets.values())
                )
            }
        }



===== 使用例 =====

if __name__ == "__main__":
    # Slack通知コールバック例
    def slack_alert(level: BudgetAlertLevel, spent: float):
        print(f"[ALERT] {level.value.upper()}: ${spent:.2f} spent")
        # 本番環境では slack_sdk.WebhookClient 等を使用
    
    # クライアント初期化
    client = HolySheepRateLimitedClient(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        budget_config=BudgetConfig(
            daily_limit_usd=50.0,  # 日次$50
            monthly_limit_usd=1000.0  # 月次$1000
        )
    )
    client.budget_manager.add_alert_callback(slack_alert)
    
    # 低コストモデルで処理
    response = client.chat_completion(
        model="deepseek-v3.2",  # $0.42/MTok
        messages=[
            {"role": "user", "content": "簡単な質問への回答を50文字以内で"}
        ],
        estimated_tokens=100
    )
    
    print(f"Response: {response['choices'][0]['message']['content']}")
    print(f"Cost: ${response['_meta']['cost_usd']:.4f}")
    print(f"Status: {client.get_status()}")

実装コード：分散AI Agent向けのログ収集システム

複数のAI Agentが協調動作する本番環境では、统一的なログ収集が運用監視の鍵となります。以下のロガーは、リクエスト・レスポンス・コスト・レイテンシを包括的に記録します。

"""
HolySheep AI - 分散AI Agent向けログ収集システム
本番環境の包括的なモニタリング
"""

import json
import logging
import sqlite3
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, List
from pathlib import Path
from dataclasses import dataclass, asdict
from contextlib import contextmanager
import threading
from queue import Queue, Empty
import atexit


@dataclass
class RequestLog:
    """リクエストログ"""
    request_id: str
    timestamp: str
    model: str
    provider: str
    messages_hash: str  # メッセージのハッシュ（機密情報保護）
    messages_tokens: int
    response_tokens: int
    total_tokens: int
    cost_usd: float
    latency_ms: float
    status: str  # success, error, rate_limited, budget_exceeded
    error_message: Optional[str] = None
    metadata: Optional[Dict[str, Any]] = None


class SQLiteLogStore:
    """SQLite永続化ログストア"""
    
    def __init__(self, db_path: str = "holy_sheep_logs.db"):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        """データベース初期化"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS request_logs (
                    request_id TEXT PRIMARY KEY,
                    timestamp TEXT NOT NULL,
                    model TEXT NOT NULL,
                    provider TEXT NOT NULL,
                    messages_hash TEXT NOT NULL,
                    messages_tokens INTEGER,
                    response_tokens INTEGER,
                    total_tokens INTEGER,
                    cost_usd REAL,
                    latency_ms REAL,
                    status TEXT NOT NULL,
                    error_message TEXT,
                    metadata TEXT
                )
            """)
            conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_timestamp 
                ON request_logs(timestamp)
            """)
            conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_model 
                ON request_logs(model)
            """)
            conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_status 
                ON request_logs(status)
            """)
    
    def insert(self, log: RequestLog):
        """ログ挿入"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                INSERT OR REPLACE INTO request_logs 
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                log.request_id,
                log.timestamp,
                log.model,
                log.provider,
                log.messages_hash,
                log.messages_tokens,
                log.response_tokens,
                log.total_tokens,
                log.cost_usd,
                log.latency_ms,
                log.status,
                log.error_message,
                json.dumps(log.metadata) if log.metadata else None
            ))
    
    def query(
        self,
        start_time: Optional[str] = None,
        end_time: Optional[str] = None,
        model: Optional[str] = None,
        status: Optional[str] = None,
        limit: int = 100
    ) -> List[RequestLog]:
        """クエリ"""
        query = "SELECT * FROM request_logs WHERE 1=1"
        params = []
        
        if start_time:
            query += " AND timestamp >= ?"
            params.append(start_time)
        if end_time:
            query += " AND timestamp <= ?"
            params.append(end_time)
        if model:
            query += " AND model = ?"
            params.append(model)
        if status:
            query += " AND status = ?"
            params.append(status)
        
        query += f" ORDER BY timestamp DESC LIMIT {limit}"
        
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.execute(query, params)
            return [RequestLog(**dict(row)) for row in cursor.fetchall()]
    
    def get_cost_summary(
        self,
        start_date: Optional[str] = None,
        end_date: Optional[str] = None
    ) -> Dict[str, Any]:
        """コストサマリー取得"""
        query = """
            SELECT 
                COUNT(*) as total_requests,
                SUM(cost_usd) as total_cost,
                AVG(cost_usd) as avg_cost,
                AVG(latency_ms) as avg_latency,
                model,
                SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) as success_count,
                SUM(CASE WHEN status != 'success' THEN 1 ELSE 0 END) as error_count
            FROM request_logs
        """
        where_clauses = []
        params = []
        
        if start_date:
            where_clauses.append("timestamp >= ?")
            params.append(start_date)
        if end_date:
            where_clauses.append("timestamp <= ?")
            params.append(end_date)
        
        if where_clauses:
            query += " WHERE " + " AND ".join(where_clauses)
        
        query += " GROUP BY model"
        
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.execute(query, params)
            return {"breakdown": [dict(row) for row in cursor.fetchall()]}


class AsyncLogWriter:
    """非同期ログライター（バックグラウンドflush）"""
    
    def __init__(self, store: SQLiteLogStore, batch_size: int = 100, flush_interval: float = 5.0):
        self.store = store
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        self.queue: Queue = Queue()
        self.buffer: List[RequestLog] = []
        self._running = True
        self._lock = threading.Lock()
        self._flush_thread = threading.Thread(target=self._flush_loop, daemon=True)
        self._flush_thread.start()
        atexit.register(self.flush)
    
    def log(self, request_log: RequestLog):
        """ログ追加"""
        self.queue.put(request_log)
    
    def _flush_loop(self):
        """バックグラウンドflushループ"""
        while self._running:
            try:
                # キューからバッファへ
                while len(self.buffer) < self.batch_size:
                    try:
                        log = self.queue.get(timeout=0.1)
                        self.buffer.append(log)
                    except Empty:
                        break
                
                # バッファflush
                if self.buffer:
                    self.flush()
                    
            except Exception as e:
                logging.error(f"Flush error: {e}")
    
    def flush(self):
        """バッファをDBにflush"""
        with self._lock:
            if not self.buffer:
                return
            
            for log in self.buffer:
                self.store.insert(log)
            self.buffer.clear()


class HolySheepLogger:
    """
    包括的ログ収集システム
    
    特徴：
    - SQLite永続化（ローカル）または外部ログシステム対応
    - 非同期write（高負荷環境対応）
    - コスト・レイテンシ追跡
    - 異常検知対応
    """
    
    def __init__(
        self,
        db_path: str = "holy_sheep_logs.db",
        log_level: int = logging.INFO
    ):
        self.logger = logging.getLogger("HolySheepLogger")
        self.logger.setLevel(log_level)
        
        # コンソールハンドラー
        if not self.logger.handlers:
            handler = logging.StreamHandler()
            handler.setFormatter(logging.Formatter(
                "%(asctime)s [%(levelname)s] %(message)s"
            ))
            self.logger.addHandler(handler)
        
        # ログストア
        self.store = SQLiteLogStore(db_path)
        self.writer = AsyncLogWriter(self.store)
        
        # リクエスト追跡
        self._request_counter = 0
        self._lock = threading.Lock()
    
    def _generate_request_id(self) -> str:
        """リクエストID生成"""
        with self._lock:
            self._request_counter += 1
            return f"req_{datetime.now().strftime('%Y%m%d%H%M%S')}_{self._request_counter:06d}"
    
    @contextmanager
    def log_request(
        self,
        model: str,
        provider: str,
        messages: List[Dict],
        metadata: Optional[Dict] = None
    ):
        """
        リクエストログコンテキスト
        
        Usage:
            with logger.log_request("deepseek-v3.2", "deepseek", messages) as ctx:
                response = api.call(...)
                ctx.success(response, latency_ms=150)
        """
        request_id = self._generate_request_id()
        start_time = datetime.now()
        
        class RequestContext:
            def __init__(ctx_self):
                ctx_self.cost = 0.0
                ctx_self.latency_ms = 0
                ctx_self.status = "pending"
                ctx_self.error_message = None
            
            def success(ctx_self, response: Dict, latency_ms: float):
                usage = response.get("usage", {})
                ctx_self.cost = (usage.get("completion_tokens", 0) / 1000) * (
                    {"deepseek": 0.00042, "openai": 0.008, "anthropic": 0.015, "google": 0.0025}.get(provider, 0.001)
                )
                ctx_self.latency_ms = latency_ms
                ctx_self.status = "success"
            
            def error(ctx_self, message: str, latency_ms: float):
                ctx_self.status = "error"
                ctx_self.error_message = message
                ctx_self.latency_ms = latency_ms
            
            def rate_limited(ctx_self, latency_ms: float):
                ctx_self.status = "rate_limited"
                ctx_self.latency_ms = latency_ms
            
            def budget_exceeded(ctx_self, latency_ms: float):
                ctx_self.status = "budget_exceeded"
                ctx_self.latency_ms = latency_ms
        
        ctx = RequestContext()
        
        try:
            yield ctx
        except Exception as e:
            ctx.error(str(e), (datetime.now() - start_time).total_seconds() * 1000)
            raise
        finally:
            # ログ書き込み
            import hashlib
            messages_hash = hashlib.sha256(
                json.dumps(messages, ensure_ascii=False).encode()
            ).hexdigest()[:16]
            
            log = RequestLog(
                request_id=request_id,
                timestamp=start_time.isoformat(),
                model=model,
                provider=provider,
                messages_hash=messages_hash,
                messages_tokens=metadata.get("input_tokens", 0) if metadata else 0,
                response_tokens=metadata.get("output_tokens", 0) if metadata else 0,
                total_tokens=(
                    (metadata.get("input_tokens", 0) + metadata.get("output_tokens", 0))
                    if metadata else 0
                ),
                cost_usd=ctx.cost,
                latency_ms=ctx.latency_ms,
                status=ctx.status,
                error_message=ctx.error_message,
                metadata=metadata
            )
            
            self.writer.log(log)
            self.logger.info(
                f"[{request_id}] {model} | "
                f"cost=${ctx.cost:.4f} | "
                f"latency={ctx.latency_ms:.0f}ms | "
                f"status={ctx.status}"
            )



===== 使用例 =====

if __name__ == "__main__":
    # ロガー初期化
    logger = HolySheepLogger(log_level=logging.DEBUG)
    
    # リクエストログ
    messages = [
        {"role": "user", "content": "テストメッセージ"}
    ]
    
    with logger.log_request(
        model="deepseek-v3.2",
        provider="deepseek",
        messages=messages,
        metadata={"agent_id": "agent-001"}
    ) as ctx:
        # API呼び出し（疑似）
        import time
        time.sleep(0.1)
        
        response = {
            "usage": {
                "completion_tokens": 150,
                "prompt_tokens": 50
            }
        }
        ctx.success(response, latency_ms=95)
    
    # コストサマリー取得
    summary = logger.store.get_cost_summary(
        start_date=(datetime.now() - timedelta(days=1)).isoformat()
    )
    print(f"Cost Summary: {summary}")

向いている人・向いていない人

向いている人 向いていない人