# 評価のレベルを体系的に分類
from enum import Enum
 
class EvaluationLevel(Enum):
    """評価の粒度レベル"""
 
    # Level 1: 単一ステップ評価
    STEP = "step"
    # 個々のツール呼び出しが適切か
    # 例: 正しいファイルを読んだか、適切なコマンドを実行したか
 
    # Level 2: タスク評価
    TASK = "task"
    # タスク全体が正しく完了したか
    # 例: バグが修正されたか、機能が実装されたか
 
    # Level 3: セッション評価
    SESSION = "session"
    # 複数タスクにわたるセッション全体の品質
    # 例: 一連の開発作業の生産性
 
    # Level 4: システム評価
    SYSTEM = "system"
    # エージェントシステム全体のパフォーマンス
    # 例: 月間の成功率推移、コスト効率
 
# 各レベルで測定すべきメトリクスの対応
LEVEL_METRICS = {
    EvaluationLevel.STEP: [
        "tool_selection_accuracy",   # 正しいツールを選んだか
        "parameter_accuracy",        # 正しいパラメータを渡したか
        "step_relevance",           # そのステップが必要だったか
    ],
    EvaluationLevel.TASK: [
        "task_success_rate",         # タスク成功率
        "partial_completion_rate",   # 部分完了率
        "total_steps",              # ステップ数
        "execution_time",           # 実行時間
        "cost_per_task",            # タスクあたりコスト
    ],
    EvaluationLevel.SESSION: [
        "tasks_completed",           # 完了タスク数
        "session_efficiency",        # セッション効率
        "context_utilization",       # コンテキスト利用効率
        "error_recovery_rate",       # エラー回復率
    ],
    EvaluationLevel.SYSTEM: [
        "daily_success_rate",        # 日次成功率
        "monthly_cost",             # 月間コスト
        "p50_latency",             # レイテンシ中央値
        "p99_latency",             # レイテンシ99パーセンタイル
        "safety_incident_rate",     # 安全性インシデント率
    ],
}

# エージェント評価メトリクスの定義
from dataclasses import dataclass, field
from typing import Optional
import json
 
@dataclass
class AgentMetrics:
    # 正確性
    task_success_rate: float      # タスク成功率 (0-1)
    partial_completion: float     # 部分完了率 (0-1)
 
    # 効率性
    total_steps: int              # 総ステップ数
    tool_calls: int               # ツール呼び出し回数
    total_time_seconds: float     # 実行時間
    redundant_steps: int          # 冗長なステップ数
 
    # コスト
    input_tokens: int             # 入力トークン数
    output_tokens: int            # 出力トークン数
    total_cost_usd: float         # 総コスト（ドル）
 
    # 安全性
    unsafe_actions: int           # 安全でない操作の回数
    guardrail_triggers: int       # ガードレール発動回数
 
    # 堅牢性
    error_recovery_rate: float    # エラー回復率
    graceful_failures: int        # 適切な失敗処理の回数
 
    @property
    def cost_per_task(self) -> float:
        return self.total_cost_usd
 
    @property
    def efficiency_score(self) -> float:
        if self.total_steps == 0:
            return 0
        return 1 - (self.redundant_steps / self.total_steps)
 
    @property
    def safety_score(self) -> float:
        """安全性スコア (0-1)"""
        if self.tool_calls == 0:
            return 1.0
        return 1 - (self.unsafe_actions / self.tool_calls)
 
    @property
    def composite_score(self) -> float:
        """重み付き総合スコア"""
        weights = {
            "accuracy": 0.35,
            "efficiency": 0.20,
            "safety": 0.25,
            "robustness": 0.10,
            "cost": 0.10,
        }
        # コストスコア: $1以下なら1.0、$10以上なら0.0
        cost_score = max(0, 1 - self.total_cost_usd / 10)
 
        return (
            weights["accuracy"] * self.task_success_rate
            + weights["efficiency"] * self.efficiency_score
            + weights["safety"] * self.safety_score
            + weights["robustness"] * self.error_recovery_rate
            + weights["cost"] * cost_score
        )
 
    def to_dict(self) -> dict:
        return {
            "accuracy": {
                "task_success_rate": self.task_success_rate,
                "partial_completion": self.partial_completion,
            },
            "efficiency": {
                "total_steps": self.total_steps,
                "tool_calls": self.tool_calls,
                "total_time_seconds": self.total_time_seconds,
                "redundant_steps": self.redundant_steps,
                "efficiency_score": self.efficiency_score,
            },
            "cost": {
                "input_tokens": self.input_tokens,
                "output_tokens": self.output_tokens,
                "total_cost_usd": self.total_cost_usd,
            },
            "safety": {
                "unsafe_actions": self.unsafe_actions,
                "guardrail_triggers": self.guardrail_triggers,
                "safety_score": self.safety_score,
            },
            "robustness": {
                "error_recovery_rate": self.error_recovery_rate,
                "graceful_failures": self.graceful_failures,
            },
            "composite_score": self.composite_score,
        }

# メトリクス収集を自動化するデコレータとフック
import time
import functools
from typing import Callable, Any
 
class MetricsCollector:
    """エージェント実行中のメトリクスを自動収集"""
 
    def __init__(self):
        self.steps: list[dict] = []
        self.tool_calls: list[dict] = []
        self.errors: list[dict] = []
        self.start_time: float = 0
        self.total_input_tokens: int = 0
        self.total_output_tokens: int = 0
 
    def start(self):
        """計測開始"""
        self.start_time = time.time()
        self.steps = []
        self.tool_calls = []
        self.errors = []
 
    def record_step(self, step_num: int, response):
        """APIレスポンスからステップ情報を記録"""
        self.steps.append({
            "step": step_num,
            "stop_reason": response.stop_reason,
            "input_tokens": response.usage.input_tokens,
            "output_tokens": response.usage.output_tokens,
            "timestamp": time.time(),
        })
        self.total_input_tokens += response.usage.input_tokens
        self.total_output_tokens += response.usage.output_tokens
 
    def record_tool_call(self, name: str, input_data: dict,
                         result: str, duration: float, is_error: bool = False):
        """ツール呼び出しを記録"""
        self.tool_calls.append({
            "name": name,
            "input_keys": list(input_data.keys()),
            "result_length": len(result),
            "duration": duration,
            "is_error": is_error,
            "timestamp": time.time(),
        })
 
    def record_error(self, error: Exception, step: int):
        """エラーを記録"""
        self.errors.append({
            "type": type(error).__name__,
            "message": str(error),
            "step": step,
            "timestamp": time.time(),
        })
 
    def get_summary(self) -> dict:
        """メトリクスサマリーを取得"""
        elapsed = time.time() - self.start_time
        total_tool_calls = len(self.tool_calls)
        error_tool_calls = sum(1 for tc in self.tool_calls if tc["is_error"])
 
        return {
            "total_steps": len(self.steps),
            "total_tool_calls": total_tool_calls,
            "elapsed_seconds": round(elapsed, 2),
            "total_input_tokens": self.total_input_tokens,
            "total_output_tokens": self.total_output_tokens,
            "tool_error_rate": error_tool_calls / total_tool_calls if total_tool_calls > 0 else 0,
            "errors": len(self.errors),
            "avg_step_tokens": (
                (self.total_input_tokens + self.total_output_tokens) / len(self.steps)
                if self.steps else 0
            ),
            "tool_call_breakdown": self._tool_breakdown(),
        }
 
    def _tool_breakdown(self) -> dict:
        """ツール呼び出しの内訳"""
        breakdown = {}
        for tc in self.tool_calls:
            name = tc["name"]
            if name not in breakdown:
                breakdown[name] = {"count": 0, "errors": 0, "total_duration": 0}
            breakdown[name]["count"] += 1
            breakdown[name]["total_duration"] += tc["duration"]
            if tc["is_error"]:
                breakdown[name]["errors"] += 1
        return breakdown

# SWE-bench スタイルの評価パイプライン
import subprocess
from pathlib import Path
import json
import tempfile
import shutil
 
class SWEBenchEvaluator:
    """SWE-benchスタイルのコーディングエージェント評価"""
 
    def __init__(self, workspace_dir: str = "/tmp/swe-eval"):
        self.workspace_dir = Path(workspace_dir)
        self.workspace_dir.mkdir(parents=True, exist_ok=True)
 
    def evaluate_patch(self, repo_path: str, patch: str,
                       test_command: str) -> dict:
        """エージェントが生成したパッチを評価"""
 
        # 1. パッチを適用
        patch_file = Path(repo_path) / "agent.patch"
        patch_file.write_text(patch)
        apply_result = subprocess.run(
            ["git", "apply", "agent.patch"],
            cwd=repo_path, capture_output=True, text=True
        )
 
        if apply_result.returncode != 0:
            return {
                "success": False,
                "reason": "パッチ適用失敗",
                "error": apply_result.stderr
            }
 
        # 2. テスト実行
        test_result = subprocess.run(
            test_command.split(),
            cwd=repo_path, capture_output=True, text=True,
            timeout=300
        )
 
        # 3. 結果判定
        return {
            "success": test_result.returncode == 0,
            "tests_passed": self._count_passed(test_result.stdout),
            "tests_failed": self._count_failed(test_result.stdout),
            "output": test_result.stdout[-2000:]  # 最後の2000文字
        }
 
    def _count_passed(self, output: str) -> int:
        """テスト通過数を抽出"""
        import re
        match = re.search(r"(\d+) passed", output)
        return int(match.group(1)) if match else 0
 
    def _count_failed(self, output: str) -> int:
        """テスト失敗数を抽出"""
        import re
        match = re.search(r"(\d+) failed", output)
        return int(match.group(1)) if match else 0
 
    def evaluate_batch(self, test_cases: list[dict]) -> dict:
        """複数のテストケースをバッチ評価"""
        results = []
        for case in test_cases:
            result = self.evaluate_single_case(case)
            results.append(result)
 
        # 集計
        total = len(results)
        resolved = sum(1 for r in results if r["success"])
        return {
            "total": total,
            "resolved": resolved,
            "resolve_rate": resolved / total if total > 0 else 0,
            "details": results,
        }
 
    def evaluate_single_case(self, case: dict) -> dict:
        """単一のSWE-benchケースを評価"""
        repo_url = case["repo"]
        commit = case["base_commit"]
        instance_id = case["instance_id"]
 
        # リポジトリをクローン
        work_dir = self.workspace_dir / instance_id
        if work_dir.exists():
            shutil.rmtree(work_dir)
 
        subprocess.run(
            ["git", "clone", "--depth", "1", repo_url, str(work_dir)],
            capture_output=True, timeout=60,
        )
        subprocess.run(
            ["git", "checkout", commit],
            cwd=str(work_dir), capture_output=True,
        )
 
        # エージェントにパッチ生成を依頼（実装は別途）
        patch = self._generate_patch(work_dir, case)
 
        # パッチを評価
        result = self.evaluate_patch(
            str(work_dir),
            patch,
            case.get("test_command", "pytest")
        )
        result["instance_id"] = instance_id
        return result
 
    def _generate_patch(self, work_dir: Path, case: dict) -> str:
        """エージェントにパッチ生成を依頼（プレースホルダ）"""
        # 実際にはここでエージェントを呼び出す
        return case.get("agent_patch", "")

# HumanEval評価の実装
import ast
import signal
from typing import Optional
 
class HumanEvalRunner:
    """HumanEvalスタイルのコード生成評価"""
 
    def __init__(self, timeout_seconds: int = 10):
        self.timeout = timeout_seconds
 
    def evaluate_solution(self, problem: dict, generated_code: str) -> dict:
        """生成されたコードをテストケースで評価"""
        entry_point = problem["entry_point"]
        test_code = problem["test"]
 
        # 構文チェック
        try:
            ast.parse(generated_code)
        except SyntaxError as e:
            return {
                "passed": False,
                "reason": f"構文エラー: {e}",
                "tests_run": 0,
                "tests_passed": 0,
            }
 
        # テスト実行
        full_code = f"{generated_code}\n\n{test_code}"
        return self._execute_tests(full_code, entry_point)
 
    def _execute_tests(self, code: str, entry_point: str) -> dict:
        """タイムアウト付きでテストを実行"""
        def timeout_handler(signum, frame):
            raise TimeoutError()
 
        old_handler = signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(self.timeout)
 
        try:
            exec_globals = {}
            exec(code, exec_globals)
 
            return {
                "passed": True,
                "reason": "全テスト通過",
                "tests_run": 1,
                "tests_passed": 1,
            }
 
        except AssertionError as e:
            return {
                "passed": False,
                "reason": f"アサーションエラー: {e}",
                "tests_run": 1,
                "tests_passed": 0,
            }
        except TimeoutError:
            return {
                "passed": False,
                "reason": "タイムアウト",
                "tests_run": 1,
                "tests_passed": 0,
            }
        except Exception as e:
            return {
                "passed": False,
                "reason": f"実行時エラー: {type(e).__name__}: {e}",
                "tests_run": 1,
                "tests_passed": 0,
            }
        finally:
            signal.alarm(0)
            signal.signal(signal.SIGALRM, old_handler)
 
    def evaluate_batch(self, problems: list[dict], solutions: list[str]) -> dict:
        """バッチ評価"""
        results = []
        for problem, solution in zip(problems, solutions):
            result = self.evaluate_solution(problem, solution)
            result["task_id"] = problem.get("task_id", "unknown")
            results.append(result)
 
        passed = sum(1 for r in results if r["passed"])
        return {
            "pass@1": passed / len(results) if results else 0,
            "total": len(results),
            "passed": passed,
            "details": results,
        }

# 実務に即した独自ベンチマークの設計
from dataclasses import dataclass
from typing import Callable, Optional
import json
 
@dataclass
class BenchmarkCase:
    """ベンチマークの1テストケース"""
    id: str
    name: str
    category: str
    difficulty: str  # easy, medium, hard
    input_prompt: str
    expected_behavior: str  # 期待される動作の記述
    validator: Callable[[str, dict], bool]  # (output, context) -> bool
    setup: Optional[Callable[[], dict]] = None  # テスト前のセットアップ
    teardown: Optional[Callable[[dict], None]] = None  # テスト後のクリーンアップ
    timeout_seconds: int = 120
    tags: list[str] = None
 
    def __post_init__(self):
        if self.tags is None:
            self.tags = []
 
class CustomBenchmark:
    """独自ベンチマークの管理と実行"""
 
    def __init__(self, name: str, version: str):
        self.name = name
        self.version = version
        self.cases: list[BenchmarkCase] = []
 
    def add_case(self, case: BenchmarkCase):
        self.cases.append(case)
 
    def filter_cases(self, category: str = None,
                     difficulty: str = None,
                     tags: list[str] = None) -> list[BenchmarkCase]:
        """条件でフィルタリング"""
        filtered = self.cases
        if category:
            filtered = [c for c in filtered if c.category == category]
        if difficulty:
            filtered = [c for c in filtered if c.difficulty == difficulty]
        if tags:
            filtered = [c for c in filtered if any(t in c.tags for t in tags)]
        return filtered
 
    def export_to_json(self, filepath: str):
        """ベンチマークをJSONにエクスポート"""
        data = {
            "name": self.name,
            "version": self.version,
            "total_cases": len(self.cases),
            "cases": [
                {
                    "id": c.id,
                    "name": c.name,
                    "category": c.category,
                    "difficulty": c.difficulty,
                    "input_prompt": c.input_prompt,
                    "expected_behavior": c.expected_behavior,
                    "tags": c.tags,
                }
                for c in self.cases
            ],
        }
        with open(filepath, "w") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
 
# 実務ベンチマークの作成例
benchmark = CustomBenchmark("coding-agent-bench", "1.0.0")
 
# ケース1: ファイル操作
benchmark.add_case(BenchmarkCase(
    id="file-001",
    name="CSVの集計",
    category="file_operations",
    difficulty="easy",
    input_prompt="data.csvを読み込んで、カラムAの合計値を計算して",
    expected_behavior="CSVファイルを読み込み、カラムAの合計値を正しく計算する",
    validator=lambda output, ctx: str(ctx["expected_sum"]) in output,
    setup=lambda: _create_test_csv(),
    teardown=lambda ctx: _cleanup_test_csv(ctx),
    tags=["file", "csv", "basic"],
))
 
# ケース2: バグ修正
benchmark.add_case(BenchmarkCase(
    id="bugfix-001",
    name="off-by-oneエラーの修正",
    category="bug_fix",
    difficulty="medium",
    input_prompt="loop.pyのforループにoff-by-oneエラーがあります。修正してテストを通してください。",
    expected_behavior="ループの範囲を修正し、テストが通ること",
    validator=lambda output, ctx: ctx["test_passed"],
    setup=lambda: _create_buggy_code(),
    tags=["bugfix", "python", "loop"],
))
 
# ケース3: リファクタリング
benchmark.add_case(BenchmarkCase(
    id="refactor-001",
    name="クラスの分割",
    category="refactoring",
    difficulty="hard",
    input_prompt="god_object.pyの大きなクラスを責務ごとに分割してください。全テストが通る状態を維持すること。",
    expected_behavior="クラスが適切に分割され、テストが全て通ること",
    validator=lambda output, ctx: ctx["test_passed"] and ctx["class_count"] >= 3,
    setup=lambda: _create_god_object(),
    tags=["refactoring", "oop", "advanced"],
    timeout_seconds=300,
))

# 汎用的なエージェント評価フレームワーク
import json
import time
from typing import Callable, Optional
from pathlib import Path
from datetime import datetime
 
class AgentEvaluator:
    """エージェントの包括的な評価を実行するフレームワーク"""
 
    def __init__(self, agent_factory: Callable, output_dir: str = "./eval_results"):
        self.agent_factory = agent_factory
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.results = []
 
    def run_evaluation(self, test_cases: list[dict],
                       parallel: bool = False) -> dict:
        """テストケースのバッチ評価"""
        for i, case in enumerate(test_cases):
            print(f"評価 {i+1}/{len(test_cases)}: {case['name']}")
            result = self._evaluate_single(case)
            self.results.append(result)
 
        aggregated = self._aggregate_results()
        self._save_results(aggregated)
        return aggregated
 
    def _evaluate_single(self, case: dict) -> dict:
        agent = self.agent_factory()
        context = {}
        start_time = time.time()
 
        # セットアップ
        if "setup" in case and case["setup"]:
            context = case["setup"]()
 
        try:
            output = agent.run(case["input"])
            elapsed = time.time() - start_time
 
            # 正確性チェック
            if "expected_output" in case:
                is_correct = case"checker"
            elif "validator" in case:
                is_correct = case"validator"
            else:
                is_correct = None
 
            return {
                "name": case["name"],
                "category": case.get("category", "unknown"),
                "difficulty": case.get("difficulty", "unknown"),
                "success": is_correct,
                "output": output[:500],
                "time_seconds": elapsed,
                "steps": getattr(agent, "step_count", None),
                "error": None
            }
 
        except Exception as e:
            return {
                "name": case["name"],
                "category": case.get("category", "unknown"),
                "difficulty": case.get("difficulty", "unknown"),
                "success": False,
                "output": None,
                "time_seconds": time.time() - start_time,
                "steps": None,
                "error": str(e)
            }
 
        finally:
            # クリーンアップ
            if "teardown" in case and case["teardown"]:
                case"teardown"
 
    def _aggregate_results(self) -> dict:
        total = len(self.results)
        successes = sum(1 for r in self.results if r["success"])
        times = [r["time_seconds"] for r in self.results if r["time_seconds"]]
 
        # カテゴリ別の集計
        category_stats = {}
        for r in self.results:
            cat = r.get("category", "unknown")
            if cat not in category_stats:
                category_stats[cat] = {"total": 0, "success": 0}
            category_stats[cat]["total"] += 1
            if r["success"]:
                category_stats[cat]["success"] += 1
 
        for cat in category_stats:
            s = category_stats[cat]
            s["success_rate"] = s["success"] / s["total"] if s["total"] > 0 else 0
 
        # 難易度別の集計
        difficulty_stats = {}
        for r in self.results:
            diff = r.get("difficulty", "unknown")
            if diff not in difficulty_stats:
                difficulty_stats[diff] = {"total": 0, "success": 0}
            difficulty_stats[diff]["total"] += 1
            if r["success"]:
                difficulty_stats[diff]["success"] += 1
 
        for diff in difficulty_stats:
            s = difficulty_stats[diff]
            s["success_rate"] = s["success"] / s["total"] if s["total"] > 0 else 0
 
        return {
            "timestamp": datetime.now().isoformat(),
            "total_cases": total,
            "success_rate": successes / total if total > 0 else 0,
            "avg_time": sum(times) / len(times) if times else 0,
            "max_time": max(times) if times else 0,
            "min_time": min(times) if times else 0,
            "error_rate": sum(1 for r in self.results if r["error"]) / total,
            "by_category": category_stats,
            "by_difficulty": difficulty_stats,
            "details": self.results,
        }
 
    def _save_results(self, results: dict):
        """結果をファイルに保存"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filepath = self.output_dir / f"eval_{timestamp}.json"
        with open(filepath, "w") as f:
            json.dump(results, f, ensure_ascii=False, indent=2)
        print(f"結果を保存: {filepath}")

# LLMを評価者として使う
import anthropic
 
class LLMJudge:
    """LLMを評価者として活用する汎用ジャッジ"""
 
    def __init__(self, model: str = "claude-sonnet-4-20250514"):
        self.client = anthropic.Anthropic()
        self.model = model
 
    def evaluate(self, task: str, output: str,
                 criteria: list[str]) -> dict:
        """LLMで出力を評価"""
        criteria_text = "\n".join(f"- {c}" for c in criteria)
 
        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            messages=[{"role": "user", "content": f"""
以下のタスクとその出力を評価してください。
 
タスク: {task}
出力: {output}
 
評価基準:
{criteria_text}
 
各基準について1-5のスコアと理由をJSON形式で出力してください。
形式: {{"criteria_name": {{"score": N, "reason": "..."}}}}
"""}]
        )
        try:
            return json.loads(response.content[0].text)
        except json.JSONDecodeError:
            return {"error": "JSON解析失敗", "raw": response.content[0].text}
 
    def compare(self, task: str, output_a: str, output_b: str,
                criteria: list[str]) -> dict:
        """2つの出力をペアワイズ比較"""
        criteria_text = "\n".join(f"- {c}" for c in criteria)
 
        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            messages=[{"role": "user", "content": f"""
以下のタスクに対する2つの出力を比較評価してください。
 
タスク: {task}
 
出力A: {output_a}
出力B: {output_b}
 
評価基準:
{criteria_text}
 
各基準について、どちらが優れているかを判定してJSON形式で出力してください。
形式: {{"criteria_name": {{"winner": "A" or "B" or "tie", "reason": "..."}}}}
"""}]
        )
        try:
            return json.loads(response.content[0].text)
        except json.JSONDecodeError:
            return {"error": "JSON解析失敗", "raw": response.content[0].text}
 
    def evaluate_trajectory(self, task: str, steps: list[dict]) -> dict:
        """エージェントの行動軌跡を評価"""
        steps_text = "\n".join(
            f"Step {i+1}: [{s['action']}] {s.get('detail', '')}"
            for i, s in enumerate(steps)
        )
 
        response = self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            messages=[{"role": "user", "content": f"""
以下のタスクに対するエージェントの行動軌跡を評価してください。
 
タスク: {task}
 
行動軌跡:
{steps_text}
 
以下の観点で評価してJSON形式で出力してください:
1. plan_quality: 計画の質 (1-5)
2. step_efficiency: ステップの効率性 (1-5)
3. error_handling: エラー対応 (1-5)
4. tool_selection: ツール選択の適切さ (1-5)
5. overall: 総合評価 (1-5)
6. redundant_steps: 不要だったステップ番号のリスト
7. suggestions: 改善提案
 
形式: {{"plan_quality": {{"score": N, "reason": "..."}}, ...}}
"""}]
        )
        try:
            return json.loads(response.content[0].text)
        except json.JSONDecodeError:
            return {"error": "JSON解析失敗", "raw": response.content[0].text}
 
# 使用例
judge = LLMJudge()
 
# 単一評価
eval_result = judge.evaluate(
    task="Pythonでクイックソートを実装して",
    output=agent_output,
    criteria=["正確性", "コードの可読性", "エラーハンドリング", "効率性"]
)
 
# ペアワイズ比較
comparison = judge.compare(
    task="REST APIの設計",
    output_a=agent_a_output,
    output_b=agent_b_output,
    criteria=["APIデザイン", "エラーレスポンス", "ドキュメント品質"]
)

# LLM-as-Judge の精度向上テクニック
class CalibratedJudge:
    """キャリブレーション済みのLLMジャッジ"""
 
    def __init__(self):
        self.client = anthropic.Anthropic()
        self.calibration_examples: list[dict] = []
 
    def add_calibration_example(self, task: str, output: str,
                                 human_scores: dict):
        """人手評価の例を追加してキャリブレーション"""
        self.calibration_examples.append({
            "task": task,
            "output": output,
            "scores": human_scores,
        })
 
    def evaluate_with_calibration(self, task: str, output: str,
                                   criteria: list[str]) -> dict:
        """キャリブレーション例を含めて評価"""
        # Few-shot例を構築
        examples_text = ""
        for i, ex in enumerate(self.calibration_examples[:3]):
            examples_text += f"""
例{i+1}:
タスク: {ex['task']}
出力: {ex['output'][:300]}
正解スコア: {json.dumps(ex['scores'], ensure_ascii=False)}
---
"""
 
        criteria_text = "\n".join(f"- {c}" for c in criteria)
 
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": f"""
あなたはAIシステムの出力品質を評価する専門家です。
以下の例を参考に、一貫した基準で評価してください。
 
{examples_text}
 
新しい評価対象:
タスク: {task}
出力: {output}
 
評価基準:
{criteria_text}
 
上記の例と同じスケール（1-5）で評価してJSON形式で出力してください。
"""}]
        )
 
        try:
            return json.loads(response.content[0].text)
        except json.JSONDecodeError:
            return {"error": "JSON解析失敗"}
 
    def evaluate_with_multiple_judges(self, task: str, output: str,
                                       criteria: list[str],
                                       num_judges: int = 3) -> dict:
        """複数回評価して合意を取る"""
        all_scores = []
 
        for _ in range(num_judges):
            result = self.evaluate_with_calibration(task, output, criteria)
            if "error" not in result:
                all_scores.append(result)
 
        if not all_scores:
            return {"error": "全ての評価が失敗"}
 
        # 平均スコアと分散を計算
        aggregated = {}
        for criterion in criteria:
            scores = []
            for judge_result in all_scores:
                if criterion in judge_result:
                    score_data = judge_result[criterion]
                    if isinstance(score_data, dict):
                        scores.append(score_data.get("score", 0))
                    elif isinstance(score_data, (int, float)):
                        scores.append(score_data)
 
            if scores:
                avg = sum(scores) / len(scores)
                variance = sum((s - avg) ** 2 for s in scores) / len(scores)
                aggregated[criterion] = {
                    "mean_score": round(avg, 2),
                    "variance": round(variance, 2),
                    "individual_scores": scores,
                    "agreement": variance < 0.5,  # 低分散=高合意
                }
 
        return aggregated

# コスト追跡の実装
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import json
 
class CostTracker:
    """エージェントのコストを詳細に追跡"""
 
    PRICING = {
        "claude-opus-4-20250514": {"input": 15.0, "output": 75.0},
        "claude-sonnet-4-20250514": {"input": 3.0, "output": 15.0},
        "claude-haiku-4-20250514": {"input": 0.25, "output": 1.25},
    }
 
    def __init__(self):
        self.records: list[dict] = []
 
    def track(self, response, model: str = None):
        """APIレスポンスのコストを記録"""
        usage = response.usage
        model_name = model or getattr(response, "model", "unknown")
 
        record = {
            "timestamp": datetime.now().isoformat(),
            "model": model_name,
            "input_tokens": usage.input_tokens,
            "output_tokens": usage.output_tokens,
            "cost_usd": self._calculate_cost(
                model_name, usage.input_tokens, usage.output_tokens
            ),
        }
        self.records.append(record)
        return record
 
    def _calculate_cost(self, model: str, input_tokens: int,
                        output_tokens: int) -> float:
        pricing = self.PRICING.get(model, {"input": 3.0, "output": 15.0})
        input_cost = (input_tokens / 1_000_000) * pricing["input"]
        output_cost = (output_tokens / 1_000_000) * pricing["output"]
        return round(input_cost + output_cost, 6)
 
    def get_total_cost(self) -> float:
        return sum(r["cost_usd"] for r in self.records)
 
    def get_cost_by_model(self) -> dict:
        """モデル別コスト"""
        by_model = {}
        for r in self.records:
            model = r["model"]
            if model not in by_model:
                by_model[model] = {"calls": 0, "cost": 0, "tokens": 0}
            by_model[model]["calls"] += 1
            by_model[model]["cost"] += r["cost_usd"]
            by_model[model]["tokens"] += r["input_tokens"] + r["output_tokens"]
        return by_model
 
    def get_cost_trend(self, window: timedelta = timedelta(hours=1)) -> list:
        """時間窓ごとのコスト推移"""
        if not self.records:
            return []
 
        buckets = {}
        for r in self.records:
            ts = datetime.fromisoformat(r["timestamp"])
            bucket_key = ts.replace(minute=0, second=0, microsecond=0).isoformat()
            if bucket_key not in buckets:
                buckets[bucket_key] = 0
            buckets[bucket_key] += r["cost_usd"]
 
        return [{"time": k, "cost": v} for k, v in sorted(buckets.items())]
 
    def summary(self) -> str:
        total = self.get_total_cost()
        by_model = self.get_cost_by_model()
        total_input = sum(r["input_tokens"] for r in self.records)
        total_output = sum(r["output_tokens"] for r in self.records)
 
        lines = [
            f"=== コストサマリー ===",
            f"API呼び出し: {len(self.records)}回",
            f"入力トークン: {total_input:,}",
            f"出力トークン: {total_output:,}",
            f"合計トークン: {total_input + total_output:,}",
            f"推定コスト: ${total:.4f}",
            f"",
            f"--- モデル別 ---",
        ]
        for model, stats in by_model.items():
            lines.append(
                f"  {model}: {stats['calls']}回, "
                f"${stats['cost']:.4f}, "
                f"{stats['tokens']:,}トークン"
            )
        return "\n".join(lines)
 
    def export_csv(self, filepath: str):
        """CSVにエクスポート"""
        import csv
        with open(filepath, "w", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=[
                "timestamp", "model", "input_tokens",
                "output_tokens", "cost_usd"
            ])
            writer.writeheader()
            writer.writerows(self.records)

# コスト最適化の分析ツール
class CostOptimizer:
    """コスト最適化の提案を生成"""
 
    def __init__(self, tracker: CostTracker):
        self.tracker = tracker
 
    def analyze(self) -> dict:
        """コスト最適化の分析"""
        records = self.tracker.records
        if not records:
            return {"message": "データなし"}
 
        total_cost = self.tracker.get_total_cost()
        by_model = self.tracker.get_cost_by_model()
 
        recommendations = []
 
        # 1. モデルダウングレードの提案
        for model, stats in by_model.items():
            if "opus" in model and stats["calls"] > 10:
                potential_saving = stats["cost"] * 0.8  # Sonnetなら80%削減
                recommendations.append({
                    "type": "model_downgrade",
                    "description": f"{model}の呼び出し{stats['calls']}回をSonnetに変更",
                    "potential_saving_usd": round(potential_saving, 4),
                    "risk": "複雑な推論の品質低下の可能性",
                })
 
        # 2. キャッシュの提案
        total_input_tokens = sum(r["input_tokens"] for r in records)
        if total_input_tokens > 1_000_000:
            cache_saving = total_input_tokens * 0.9 * 3.0 / 1_000_000 * 0.5
            recommendations.append({
                "type": "prompt_caching",
                "description": "プロンプトキャッシュの有効化",
                "potential_saving_usd": round(cache_saving, 4),
                "risk": "キャッシュミスによる追加レイテンシ",
            })
 
        # 3. バッチAPIの提案
        if len(records) > 50:
            batch_saving = total_cost * 0.5
            recommendations.append({
                "type": "batch_api",
                "description": "リアルタイム不要なタスクにBatch APIを使用（50%割引）",
                "potential_saving_usd": round(batch_saving, 4),
                "risk": "結果取得まで最大24時間",
            })
 
        # 4. ステップ数最適化
        avg_steps = len(records) / max(1, len(set(r["timestamp"][:10] for r in records)))
        if avg_steps > 10:
            recommendations.append({
                "type": "step_optimization",
                "description": f"平均ステップ数{avg_steps:.1f}を削減（目標: 5以下）",
                "potential_saving_usd": round(total_cost * 0.3, 4),
                "risk": "タスク成功率の低下の可能性",
            })
 
        total_potential_saving = sum(r["potential_saving_usd"] for r in recommendations)
 
        return {
            "current_total_cost": round(total_cost, 4),
            "recommendations": recommendations,
            "total_potential_saving": round(total_potential_saving, 4),
            "potential_reduction_pct": round(
                total_potential_saving / total_cost * 100, 1
            ) if total_cost > 0 else 0,
        }

# エージェント導入のROI分析
@dataclass
class ROIAnalysis:
    """エージェント導入のROI分析"""
 
    # 人間の作業コスト
    human_hourly_rate_usd: float = 80.0  # エンジニアの時給
    human_hours_per_task: float = 2.0    # タスクあたりの人間の作業時間
    tasks_per_month: int = 100           # 月間タスク数
 
    # エージェントのコスト
    agent_cost_per_task_usd: float = 0.50  # タスクあたりのAPI費用
    agent_success_rate: float = 0.85       # エージェントの成功率
    agent_infra_monthly_usd: float = 100   # インフラ費用（月額）
 
    # 人間のレビューコスト
    review_minutes_per_task: float = 15    # レビューにかかる時間（分）
 
    @property
    def human_cost_per_task(self) -> float:
        """人間が全て手作業で行う場合のコスト"""
        return self.human_hourly_rate_usd * self.human_hours_per_task
 
    @property
    def agent_total_cost_per_task(self) -> float:
        """エージェント使用時の1タスクあたり総コスト"""
        review_cost = self.human_hourly_rate_usd * (self.review_minutes_per_task / 60)
        # 成功時: API + レビュー
        # 失敗時: API + 人間が全てやり直し
        success_cost = self.agent_cost_per_task_usd + review_cost
        failure_cost = self.agent_cost_per_task_usd + self.human_cost_per_task
        return (
            self.agent_success_rate * success_cost
            + (1 - self.agent_success_rate) * failure_cost
        )
 
    @property
    def monthly_saving(self) -> float:
        """月間の節約額"""
        human_monthly = self.human_cost_per_task * self.tasks_per_month
        agent_monthly = (
            self.agent_total_cost_per_task * self.tasks_per_month
            + self.agent_infra_monthly_usd
        )
        return human_monthly - agent_monthly
 
    @property
    def roi_percentage(self) -> float:
        """ROI（投資対効果）"""
        human_monthly = self.human_cost_per_task * self.tasks_per_month
        agent_monthly = (
            self.agent_total_cost_per_task * self.tasks_per_month
            + self.agent_infra_monthly_usd
        )
        investment = agent_monthly
        return ((human_monthly - agent_monthly) / investment) * 100
 
    def generate_report(self) -> str:
        """ROIレポートを生成"""
        human_monthly = self.human_cost_per_task * self.tasks_per_month
        agent_monthly = (
            self.agent_total_cost_per_task * self.tasks_per_month
            + self.agent_infra_monthly_usd
        )
 
        return f"""
=== エージェント導入 ROI分析 ===
 
■ 前提条件
  エンジニア時給: ${self.human_hourly_rate_usd}/h
  タスクあたり作業時間: {self.human_hours_per_task}h
  月間タスク数: {self.tasks_per_month}
  エージェント成功率: {self.agent_success_rate:.0%}
 
■ コスト比較（月間）
  人間のみ: ${human_monthly:,.0f}
  エージェント活用: ${agent_monthly:,.0f}
  差額: ${self.monthly_saving:,.0f}
 
■ タスクあたりコスト
  人間のみ: ${self.human_cost_per_task:.2f}
  エージェント: ${self.agent_total_cost_per_task:.2f}
 
■ ROI
  {self.roi_percentage:.0f}%
  年間節約額: ${self.monthly_saving * 12:,.0f}
"""
 
# 使用例
roi = ROIAnalysis(
    human_hourly_rate_usd=80,
    human_hours_per_task=1.5,
    tasks_per_month=200,
    agent_cost_per_task_usd=0.40,
    agent_success_rate=0.88,
)
print(roi.generate_report())

# エージェントのA/Bテスト
import random
from typing import Callable
from datetime import datetime
 
class AgentABTest:
    """2つのエージェント構成をA/Bテスト"""
 
    def __init__(self, name: str,
                 agent_a_factory: Callable,
                 agent_b_factory: Callable,
                 judge: Optional["LLMJudge"] = None):
        self.name = name
        self.agent_a_factory = agent_a_factory
        self.agent_b_factory = agent_b_factory
        self.judge = judge
        self.results_a: list[dict] = []
        self.results_b: list[dict] = []
 
    def run(self, test_cases: list[dict], randomize: bool = True) -> dict:
        """A/Bテストを実行"""
        cases = test_cases.copy()
        if randomize:
            random.shuffle(cases)
 
        for i, case in enumerate(cases):
            print(f"テスト {i+1}/{len(cases)}: {case.get('name', 'unnamed')}")
 
            # Agent A
            result_a = self._run_single(self.agent_a_factory, case)
            self.results_a.append(result_a)
 
            # Agent B
            result_b = self._run_single(self.agent_b_factory, case)
            self.results_b.append(result_b)
 
            # LLM-as-Judgeによる比較（オプション）
            if self.judge and result_a["output"] and result_b["output"]:
                comparison = self.judge.compare(
                    task=case["input"],
                    output_a=result_a["output"],
                    output_b=result_b["output"],
                    criteria=["正確性", "完全性", "コード品質"],
                )
                result_a["judge_comparison"] = comparison
                result_b["judge_comparison"] = comparison
 
        return self._analyze()
 
    def _run_single(self, factory: Callable, case: dict) -> dict:
        """単一のエージェントでテストを実行"""
        agent = factory()
        start = time.time()
        try:
            output = agent.run(case["input"])
            elapsed = time.time() - start
 
            is_correct = None
            if "validator" in case:
                is_correct = case"validator"
 
            return {
                "success": is_correct,
                "output": output[:1000],
                "time": elapsed,
                "error": None,
            }
        except Exception as e:
            return {
                "success": False,
                "output": None,
                "time": time.time() - start,
                "error": str(e),
            }
 
    def _analyze(self) -> dict:
        """結果を分析"""
        n = len(self.results_a)
 
        # 成功率
        success_a = sum(1 for r in self.results_a if r["success"]) / n if n > 0 else 0
        success_b = sum(1 for r in self.results_b if r["success"]) / n if n > 0 else 0
 
        # 平均時間
        times_a = [r["time"] for r in self.results_a if r["time"]]
        times_b = [r["time"] for r in self.results_b if r["time"]]
        avg_time_a = sum(times_a) / len(times_a) if times_a else 0
        avg_time_b = sum(times_b) / len(times_b) if times_b else 0
 
        # エラー率
        error_a = sum(1 for r in self.results_a if r["error"]) / n if n > 0 else 0
        error_b = sum(1 for r in self.results_b if r["error"]) / n if n > 0 else 0
 
        # 統計的有意性の簡易テスト
        significance = self._chi_square_test(
            sum(1 for r in self.results_a if r["success"]),
            sum(1 for r in self.results_b if r["success"]),
            n,
        )
 
        return {
            "test_name": self.name,
            "total_cases": n,
            "agent_a": {
                "success_rate": round(success_a, 4),
                "avg_time": round(avg_time_a, 2),
                "error_rate": round(error_a, 4),
            },
            "agent_b": {
                "success_rate": round(success_b, 4),
                "avg_time": round(avg_time_b, 2),
                "error_rate": round(error_b, 4),
            },
            "winner": "A" if success_a > success_b else "B" if success_b > success_a else "tie",
            "improvement": round((success_b - success_a) / max(success_a, 0.001) * 100, 1),
            "statistically_significant": significance < 0.05,
            "p_value": round(significance, 4),
        }
 
    def _chi_square_test(self, success_a: int, success_b: int, n: int) -> float:
        """簡易カイ二乗検定"""
        if n == 0:
            return 1.0
        fail_a = n - success_a
        fail_b = n - success_b
        total = 2 * n
        expected = (success_a + success_b) / 2
 
        if expected == 0 or (n - expected) == 0:
            return 1.0
 
        chi2 = ((success_a - expected) ** 2 / expected
                + (fail_a - (n - expected)) ** 2 / (n - expected)
                + (success_b - expected) ** 2 / expected
                + (fail_b - (n - expected)) ** 2 / (n - expected))
 
        # 自由度1のカイ二乗分布の近似p値
        import math
        p_value = math.exp(-chi2 / 2)
        return p_value

# CI/CDでのリグレッション検出
import json
from pathlib import Path
from typing import Optional
 
class RegressionDetector:
    """エージェントのリグレッションを検出"""
 
    def __init__(self, baseline_path: str = "./eval_baseline.json"):
        self.baseline_path = Path(baseline_path)
        self.baseline: Optional[dict] = None
        if self.baseline_path.exists():
            with open(self.baseline_path) as f:
                self.baseline = json.load(f)
 
    def check_regression(self, current_results: dict,
                         thresholds: dict = None) -> dict:
        """現在の結果をベースラインと比較"""
        if not self.baseline:
            return {
                "status": "no_baseline",
                "message": "ベースラインが存在しません。現在の結果をベースラインとして保存します。",
            }
 
        default_thresholds = {
            "success_rate_drop": 0.05,      # 成功率5%以上の低下で警告
            "avg_time_increase": 1.5,       # 平均時間1.5倍以上で警告
            "cost_increase": 1.3,           # コスト1.3倍以上で警告
            "error_rate_increase": 0.03,    # エラー率3%以上の増加で警告
        }
        t = thresholds or default_thresholds
 
        regressions = []
        improvements = []
 
        # 成功率チェック
        baseline_sr = self.baseline.get("success_rate", 0)
        current_sr = current_results.get("success_rate", 0)
        sr_diff = current_sr - baseline_sr
 
        if sr_diff < -t["success_rate_drop"]:
            regressions.append({
                "metric": "success_rate",
                "baseline": baseline_sr,
                "current": current_sr,
                "change": sr_diff,
                "severity": "critical" if sr_diff < -0.10 else "warning",
            })
        elif sr_diff > t["success_rate_drop"]:
            improvements.append({
                "metric": "success_rate",
                "baseline": baseline_sr,
                "current": current_sr,
                "change": sr_diff,
            })
 
        # 時間チェック
        baseline_time = self.baseline.get("avg_time", 0)
        current_time = current_results.get("avg_time", 0)
        if baseline_time > 0:
            time_ratio = current_time / baseline_time
            if time_ratio > t["avg_time_increase"]:
                regressions.append({
                    "metric": "avg_time",
                    "baseline": baseline_time,
                    "current": current_time,
                    "change_ratio": time_ratio,
                    "severity": "warning",
                })
 
        # エラー率チェック
        baseline_err = self.baseline.get("error_rate", 0)
        current_err = current_results.get("error_rate", 0)
        err_diff = current_err - baseline_err
 
        if err_diff > t["error_rate_increase"]:
            regressions.append({
                "metric": "error_rate",
                "baseline": baseline_err,
                "current": current_err,
                "change": err_diff,
                "severity": "critical" if err_diff > 0.10 else "warning",
            })
 
        has_critical = any(r["severity"] == "critical" for r in regressions)
 
        return {
            "status": "regression" if regressions else "ok",
            "has_critical": has_critical,
            "regressions": regressions,
            "improvements": improvements,
            "recommendation": (
                "デプロイを中止してください" if has_critical
                else "警告を確認してください" if regressions
                else "問題ありません"
            ),
        }
 
    def update_baseline(self, results: dict):
        """ベースラインを更新"""
        with open(self.baseline_path, "w") as f:
            json.dump(results, f, ensure_ascii=False, indent=2)
        self.baseline = results

# .github/workflows/agent-eval.yml
name: Agent Evaluation
 
on:
  pull_request:
    paths:
      - 'agent/**'
      - 'prompts/**'
  schedule:
    - cron: '0 0 * * *'  # 毎日深夜
 
jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
 
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
 
      - name: Install dependencies
        run: pip install -r requirements.txt
 
      - name: Run evaluation suite
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
        run: |
          python -m agent.evaluate \
            --test-suite tests/agent_eval/ \
            --output results/eval_$(date +%Y%m%d).json \
            --baseline results/baseline.json
 
      - name: Check for regressions
        run: |
          python -m agent.check_regression \
            --current results/eval_$(date +%Y%m%d).json \
            --baseline results/baseline.json \
            --fail-on-critical
 
      - name: Upload results
        uses: actions/upload-artifact@v4
        with:
          name: eval-results
          path: results/
 
      - name: Comment on PR
        if: github.event_name == 'pull_request'
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const results = JSON.parse(
              fs.readFileSync('results/regression_check.json', 'utf8')
            );
            const body = `## Agent Evaluation Results
 
            | Metric | Baseline | Current | Status |
            |--------|----------|---------|--------|
            | Success Rate | ${results.baseline_sr} | ${results.current_sr} | ${results.sr_status} |
            | Avg Time | ${results.baseline_time}s | ${results.current_time}s | ${results.time_status} |
            | Error Rate | ${results.baseline_err} | ${results.current_err} | ${results.err_status} |
            `;
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: body
            });

# Prometheusメトリクスのエクスポート
from prometheus_client import Counter, Histogram, Gauge, start_http_server
 
# メトリクス定義
AGENT_TASKS_TOTAL = Counter(
    "agent_tasks_total",
    "Total number of agent tasks",
    ["status", "model"]
)
 
AGENT_TASK_DURATION = Histogram(
    "agent_task_duration_seconds",
    "Agent task duration in seconds",
    ["model"],
    buckets=[1, 5, 10, 30, 60, 120, 300, 600]
)
 
AGENT_COST_USD = Counter(
    "agent_cost_usd_total",
    "Total cost in USD",
    ["model"]
)
 
AGENT_TOKENS_TOTAL = Counter(
    "agent_tokens_total",
    "Total tokens used",
    ["model", "direction"]  # direction: input/output
)
 
AGENT_TOOL_CALLS = Counter(
    "agent_tool_calls_total",
    "Total tool calls",
    ["tool_name", "status"]  # status: success/error
)
 
AGENT_ACTIVE_SESSIONS = Gauge(
    "agent_active_sessions",
    "Number of active agent sessions"
)
 
class PrometheusMonitor:
    """Prometheusメトリクス出力"""
 
    def __init__(self, port: int = 9090):
        start_http_server(port)
 
    def record_task_complete(self, model: str, success: bool,
                              duration: float, cost: float,
                              input_tokens: int, output_tokens: int):
        status = "success" if success else "failure"
        AGENT_TASKS_TOTAL.labels(status=status, model=model).inc()
        AGENT_TASK_DURATION.labels(model=model).observe(duration)
        AGENT_COST_USD.labels(model=model).inc(cost)
        AGENT_TOKENS_TOTAL.labels(model=model, direction="input").inc(input_tokens)
        AGENT_TOKENS_TOTAL.labels(model=model, direction="output").inc(output_tokens)
 
    def record_tool_call(self, tool_name: str, success: bool):
        status = "success" if success else "error"
        AGENT_TOOL_CALLS.labels(tool_name=tool_name, status=status).inc()
 
    def set_active_sessions(self, count: int):
        AGENT_ACTIVE_SESSIONS.set(count)

# アラートルールの定義
from dataclasses import dataclass
from typing import Callable, Optional
from datetime import datetime, timedelta
import logging
 
logger = logging.getLogger(__name__)
 
@dataclass
class AlertRule:
    """アラートルールの定義"""
    name: str
    condition: Callable[[dict], bool]
    severity: str  # critical, warning, info
    message_template: str
    cooldown_minutes: int = 30  # 同じアラートの最小間隔
    last_fired: Optional[datetime] = None
 
class AlertManager:
    """エージェントのアラート管理"""
 
    def __init__(self):
        self.rules: list[AlertRule] = []
        self.alert_history: list[dict] = []
 
    def add_rule(self, rule: AlertRule):
        self.rules.append(rule)
 
    def check(self, metrics: dict):
        """メトリクスをチェックしてアラートを発火"""
        now = datetime.now()
 
        for rule in self.rules:
            # クールダウンチェック
            if rule.last_fired:
                elapsed = (now - rule.last_fired).total_seconds() / 60
                if elapsed < rule.cooldown_minutes:
                    continue
 
            try:
                if rule.condition(metrics):
                    self._fire_alert(rule, metrics, now)
            except Exception as e:
                logger.error(f"アラートルール {rule.name} の評価エラー: {e}")
 
    def _fire_alert(self, rule: AlertRule, metrics: dict, now: datetime):
        """アラートを発火"""
        alert = {
            "name": rule.name,
            "severity": rule.severity,
            "message": rule.message_template.format(**metrics),
            "timestamp": now.isoformat(),
            "metrics_snapshot": metrics,
        }
        self.alert_history.append(alert)
        rule.last_fired = now
 
        if rule.severity == "critical":
            logger.critical(f"[CRITICAL] {alert['message']}")
            self._send_notification(alert)
        elif rule.severity == "warning":
            logger.warning(f"[WARNING] {alert['message']}")
 
    def _send_notification(self, alert: dict):
        """通知を送信（Slack, PagerDuty等）"""
        # 実装例: Slack Webhook
        pass
 
# アラートルールの設定
alert_manager = AlertManager()
 
alert_manager.add_rule(AlertRule(
    name="low_success_rate",
    condition=lambda m: m.get("success_rate_1h", 1) < 0.7,
    severity="critical",
    message_template="成功率が70%を下回りました: {success_rate_1h:.1%}",
    cooldown_minutes=15,
))
 
alert_manager.add_rule(AlertRule(
    name="high_cost",
    condition=lambda m: m.get("cost_1h", 0) > 10.0,
    severity="warning",
    message_template="直近1時間のコストが$10を超えました: ${cost_1h:.2f}",
    cooldown_minutes=60,
))
 
alert_manager.add_rule(AlertRule(
    name="high_error_rate",
    condition=lambda m: m.get("error_rate_1h", 0) > 0.15,
    severity="critical",
    message_template="エラー率が15%を超えました: {error_rate_1h:.1%}",
    cooldown_minutes=15,
))
 
alert_manager.add_rule(AlertRule(
    name="high_latency",
    condition=lambda m: m.get("p99_latency_seconds", 0) > 120,
    severity="warning",
    message_template="P99レイテンシが120秒を超えました: {p99_latency_seconds:.0f}秒",
    cooldown_minutes=30,
))