第十三章：评估与质量保障¶

from typing import List

def evaluate_step_level(trajectory: List["ToolCall"],
                       ground_truth: List["ToolCall"]) -> dict:
    """步骤级评估"""

    correct_tools = 0
    correct_calls = 0
    successful = 0
    total = len(trajectory)

    for i, predicted_call in enumerate(trajectory):
        # 工具是否正确
        if predicted_call.tool_name == ground_truth[i].tool_name:
            correct_tools += 1

        # 调用是否完全正确(工具+参数)
        if (predicted_call.tool_name == ground_truth[i].tool_name and
            predicted_call.args == ground_truth[i].args):
            correct_calls += 1

        # 执行是否成功
        if predicted_call.success:
            successful += 1

    return {
        "tool_accuracy": correct_tools / total,
        "call_accuracy": correct_calls / total,
        "execution_success_rate": successful / total,
    }

from typing import List

def evaluate_trajectory_level(trajectory: List["ToolCall"],
                              optimal_trajectory: List["ToolCall"]) -> dict:
    """轨迹级评估"""

    actual_length = len(trajectory)
    optimal_length = len(optimal_trajectory)

    # 调用效率
    efficiency = optimal_length / actual_length if actual_length > 0 else 0

    # 错误恢复
    errors = sum(1 for call in trajectory if not call.success)
    recovered = sum(1 for i, call in enumerate(trajectory)
                   if not call.success and i+1 < len(trajectory) and trajectory[i+1].success)
    recovery_rate = recovered / errors if errors > 0 else 0

    # 重复调用
    duplicates = sum(1 for i in range(len(trajectory)-1)
                    if trajectory[i].tool_name == trajectory[i+1].tool_name)
    duplicate_rate = duplicates / actual_length if actual_length > 0 else 0

    return {
        "efficiency": efficiency,
        "error_recovery_rate": recovery_rate,
        "duplicate_rate": duplicate_rate,
        "trajectory_length": actual_length,
        "optimal_length": optimal_length,
    }

from datetime import datetime
from typing import List

def evaluate_task_level(results: List["TaskResult"]) -> dict:
    """任务级评估"""

    successful = sum(1 for r in results if r.success)
    total = len(results)

    # 任务成功率
    success_rate = successful / total

    # 执行时间
    execution_times = [r.end_time - r.start_time for r in results]
    avg_time = sum(execution_times) / len(execution_times)

    # Token效率
    tokens = [r.tokens_used for r in results]
    avg_tokens = sum(tokens) / len(tokens)

    # 成本效率(假设0.001美元/1000Token)
    costs = [t * 0.001 / 1000 for t in tokens]
    total_cost = sum(costs)
    avg_cost = total_cost / len(results)

    return {
        "success_rate": success_rate,
        "avg_execution_time_sec": avg_time.total_seconds(),
        "avg_tokens_per_task": avg_tokens,
        "total_cost_usd": total_cost,
        "avg_cost_per_task": avg_cost,
    }

from dataclasses import dataclass

@dataclass
class EvaluationMetrics:
    """完整的评估指标集"""

    # ======== 步骤级 ========
    tool_accuracy: float              # 工具选择准确率
    parameter_accuracy: float         # 参数准确率
    execution_success_rate: float     # 执行成功率

    # ======== 轨迹级 ========
    trajectory_efficiency: float      # 轨迹效率(最优/实际)
    error_recovery_rate: float        # 错误恢复率
    duplicate_rate: float             # 重复调用率

    # ======== 任务级 ========
    task_success_rate: float          # 任务成功率
    avg_execution_time: float         # 平均执行时间(秒)
    avg_tokens_per_task: float        # 平均Token消耗
    avg_cost_per_task: float          # 平均成本(美元)

    # ======== 综合指标 ========
    overall_quality_score: float      # 综合质量评分(0-100)

    def compute_overall_score(self) -> float:
        """
        计算综合质量评分

        权重配置：
        - 任务成功率：40%(最关键)
        - 轨迹效率：30%(关键)
        - 步骤准确率：20%(基础)
        - 成本效率：10%(现实考虑)
        """
        score = (
            self.task_success_rate * 0.4 * 100 +
            self.trajectory_efficiency * 0.3 * 100 +
            ((self.tool_accuracy + self.parameter_accuracy) / 2) * 0.2 * 100 +
            max(0, (1 - self.avg_cost_per_task / 0.1)) * 0.1 * 100  # 假设0.1美元为基准
        )
        return min(100, max(0, score))

from abc import ABC, abstractmethod
from typing import Any, Dict, List
import logging

logger = logging.getLogger(__name__)

class Evaluator(ABC):
    """评估器基类"""

    @abstractmethod
    def evaluate(self, prediction: Any, ground_truth: Any) -> Dict[str, float]:
        """评估单个样本"""
        pass

    @abstractmethod
    def aggregate(self, results: List[Dict[str, float]]) -> EvaluationMetrics:
        """聚合多个样本的评估结果"""
        pass

class StepLevelEvaluator(Evaluator):
    """步骤级评估器"""

    def evaluate(self, predicted_call: ToolCall, ground_truth_call: ToolCall) -> Dict[str, float]:
        """评估单个工具调用"""
        tool_correct = predicted_call.tool_name == ground_truth_call.tool_name
        params_correct = predicted_call.args == ground_truth_call.args
        success = predicted_call.success

        return {
            "tool_correct": 1.0 if tool_correct else 0.0,
            "params_correct": 1.0 if params_correct else 0.0,
            "execution_success": 1.0 if success else 0.0,
        }

    def aggregate(self, results: List[Dict[str, float]]) -> Dict[str, float]:
        """聚合结果"""
        if not results:
            return {"tool_accuracy": 0, "param_accuracy": 0, "success_rate": 0}

        tool_sum = sum(r["tool_correct"] for r in results)
        param_sum = sum(r["params_correct"] for r in results)
        success_sum = sum(r["execution_success"] for r in results)
        total = len(results)

        return {
            "tool_accuracy": tool_sum / total,
            "param_accuracy": param_sum / total,
            "success_rate": success_sum / total,
        }

class TrajectoryLevelEvaluator(Evaluator):
    """轨迹级评估器"""

    def evaluate(self, trajectory: List[ToolCall],
                optimal_trajectory: List[ToolCall]) -> Dict[str, float]:
        """评估工具调用序列"""

        actual_len = len(trajectory)
        optimal_len = len(optimal_trajectory)
        efficiency = optimal_len / actual_len if actual_len > 0 else 0

        errors = sum(1 for call in trajectory if not call.success)
        if errors > 0:
            recovery = sum(1 for i in range(len(trajectory)-1)
                         if not trajectory[i].success and trajectory[i+1].success)
            recovery_rate = recovery / errors
        else:
            recovery_rate = 0

        duplicates = sum(1 for i in range(len(trajectory)-1)
                        if trajectory[i].tool_name == trajectory[i+1].tool_name)
        duplicate_rate = duplicates / actual_len if actual_len > 0 else 0

        return {
            "efficiency": efficiency,
            "recovery_rate": recovery_rate,
            "duplicate_rate": duplicate_rate,
        }

    def aggregate(self, results: List[Dict[str, float]]) -> Dict[str, float]:
        """聚合结果"""
        if not results:
            return {"avg_efficiency": 0, "avg_recovery_rate": 0, "avg_duplicate_rate": 0}

        avg_efficiency = sum(r["efficiency"] for r in results) / len(results)
        avg_recovery = sum(r["recovery_rate"] for r in results) / len(results)
        avg_duplicate = sum(r["duplicate_rate"] for r in results) / len(results)

        return {
            "avg_efficiency": avg_efficiency,
            "avg_recovery_rate": avg_recovery,
            "avg_duplicate_rate": avg_duplicate,
        }

class TaskLevelEvaluator(Evaluator):
    """任务级评估器"""

    def evaluate(self, result: TaskResult) -> Dict[str, float]:
        """评估单个任务"""
        success = 1.0 if result.success else 0.0
        time_seconds = (result.end_time - result.start_time).total_seconds()
        cost = result.tokens_used * 0.001 / 1000

        return {
            "success": success,
            "time_seconds": time_seconds,
            "tokens": float(result.tokens_used),
            "cost": cost,
        }

    def aggregate(self, results: List[Dict[str, float]]) -> Dict[str, float]:
        """聚合结果"""
        if not results:
            return {"success_rate": 0, "avg_time": 0, "avg_tokens": 0, "avg_cost": 0}

        success_sum = sum(r["success"] for r in results)
        success_rate = success_sum / len(results)

        avg_time = sum(r["time_seconds"] for r in results) / len(results)
        avg_tokens = sum(r["tokens"] for r in results) / len(results)
        avg_cost = sum(r["cost"] for r in results) / len(results)

        return {
            "success_rate": success_rate,
            "avg_time_seconds": avg_time,
            "avg_tokens": avg_tokens,
            "avg_cost_usd": avg_cost,
        }

from unittest.mock import Mock, patch, MagicMock
import asyncio

class MockLLMResponse:
    """模拟LLM响应"""
    def __init__(self, tool_calls: list):
        self.tool_calls = tool_calls

class E2ETestWithMocks:
    """使用Mock的端到端测试"""

    def setup_mock_llm(self):
        """设置Mock LLM"""
        self.mock_llm = Mock()

        # 预设第一步响应：读取文件
        self.mock_llm.generate.side_effect = [
            MockLLMResponse([
                {"tool_name": "read_file", "args": {"path": "data.txt"}}
            ]),
            MockLLMResponse([
                {"tool_name": "analyze_text", "args": {"text": "file content"}}
            ]),
            MockLLMResponse([
                {"tool_name": "write_file", "args": {"path": "result.txt", "content": "analysis"}}
            ])
        ]

    def setup_mock_tools(self):
        """设置Mock工具"""
        self.mock_tools = {
            "read_file": Mock(return_value="file content"),
            "analyze_text": Mock(return_value="analysis result"),
            "write_file": Mock(return_value=True),
        }

    async def test_file_analysis_workflow(self):
        """测试文件分析工作流"""

        self.setup_mock_llm()
        self.setup_mock_tools()

        # 创建Agent(使用Mock)
        agent = Agent(
            llm=self.mock_llm,
            tools=self.mock_tools
        )

        # 运行任务
        result = await agent.run("分析data.txt文件并保存结果到result.txt")

        # 验证
        assert result.success == True
        assert self.mock_tools["read_file"].called
        assert self.mock_tools["analyze_text"].called
        assert self.mock_tools["write_file"].called

        # 验证调用顺序
        calls = self.mock_llm.generate.call_args_list
        assert len(calls) == 3  # 三步调用

import os
from anthropic import Anthropic

class E2ETestWithRealLLM:
    """使用真实LLM的端到端测试"""

    def __init__(self):
        self.client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
        self.test_data_dir = "/tmp/e2e_test_data"
        os.makedirs(self.test_data_dir, exist_ok=True)

    async def test_real_file_analysis(self):
        """使用真实LLM的测试"""

        # 准备测试数据
        test_file = os.path.join(self.test_data_dir, "sample.txt")
        with open(test_file, "w") as f:
            f.write("This is a sample document for testing.\nIt contains multiple lines.")

        # 定义真实工具
        tools = [
            {
                "name": "read_file",
                "description": "读取文件内容",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "path": {"type": "string", "description": "文件路径"}
                    },
                    "required": ["path"]
                }
            },
            {
                "name": "write_file",
                "description": "写入文件",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "path": {"type": "string"},
                        "content": {"type": "string"}
                    },
                    "required": ["path", "content"]
                }
            }
        ]

        # 系统消息
        system_prompt = """你是一个文件分析助手。
        可用工具：read_file(读取文件)、write_file(写入文件)
        任务：读取文件并分析其内容,然后将分析结果写入输出文件。
        """

        # 初始消息
        messages = [
            {"role": "user", "content": f"请分析{test_file}的内容,并将结果写入 {os.path.join(self.test_data_dir, 'result.txt')}"}
        ]

        # 迭代交互
        max_iterations = 5
        iteration = 0
        tool_calls_made = []

        while iteration < max_iterations:
            iteration += 1

            # 调用真实LLM
            response = self.client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=1024,
                system=system_prompt,
                tools=tools,
                messages=messages
            )

            # 检查是否需要工具调用
            if response.stop_reason == "tool_use":
                tool_use_block = next(
                    (block for block in response.content if block.type == "tool_use"),
                    None
                )

                if tool_use_block:
                    tool_name = tool_use_block.name
                    tool_input = tool_use_block.input

                    tool_calls_made.append({
                        "tool": tool_name,
                        "input": tool_input
                    })

                    # 执行真实工具
                    if tool_name == "read_file":
                        with open(tool_input["path"], "r") as f:
                            tool_result = f.read()
                    elif tool_name == "write_file":
                        with open(tool_input["path"], "w") as f:
                            f.write(tool_input["content"])
                        tool_result = "文件已写入"
                    else:
                        tool_result = "未知工具"

                    # 继续对话
                    messages.append({"role": "assistant", "content": response.content})
                    messages.append({
                        "role": "user",
                        "content": [
                            {
                                "type": "tool_result",
                                "tool_use_id": tool_use_block.id,
                                "content": tool_result
                            }
                        ]
                    })
            else:
                # LLM完成
                break

        # 验证结果
        result_file = os.path.join(self.test_data_dir, "result.txt")
        assert os.path.exists(result_file), "结果文件未创建"

        with open(result_file, "r") as f:
            result_content = f.read()

        assert len(result_content) > 0, "结果文件为空"
        assert len(tool_calls_made) >= 2, f"工具调用不足,只有{len(tool_calls_made)}次"

        return {
            "success": True,
            "tool_calls": len(tool_calls_made),
            "iterations": iteration,
            "result": result_content
        }

import pytest
import tempfile
import json
from pathlib import Path

class TestFixture:
    """测试夹具基类"""

    @pytest.fixture(scope="session")
    def test_data_dir(self):
        """创建临时测试数据目录"""
        with tempfile.TemporaryDirectory() as tmpdir:
            yield tmpdir

    @pytest.fixture
    def sample_files(self, test_data_dir):
        """创建示例文件"""
        files = {}

        # 创建文本文件
        text_file = Path(test_data_dir) / "sample.txt"
        text_file.write_text("Sample text content for testing")
        files["text"] = str(text_file)

        # 创建JSON文件
        json_file = Path(test_data_dir) / "data.json"
        json_file.write_text(json.dumps({"key": "value", "number": 42}))
        files["json"] = str(json_file)

        # 创建CSV文件
        csv_file = Path(test_data_dir) / "data.csv"
        csv_file.write_text("name,age,city\nAlice,30,NYC\nBob,25,LA")
        files["csv"] = str(csv_file)

        return files

    @pytest.fixture
    def mock_tools(self):
        """Mock工具集合"""
        from unittest.mock import Mock

        return {
            "read_file": Mock(return_value="mocked file content"),
            "analyze_text": Mock(return_value="analysis result"),
            "list_files": Mock(return_value=["file1.txt", "file2.txt"]),
        }

    @pytest.fixture
    def agent_config(self):
        """Agent配置"""
        return {
            "model": "claude-sonnet-4-6",
            "max_iterations": 10,
            "timeout": 30,
            "tools_enabled": ["read_file", "analyze_text", "write_file"]
        }

class RegressionTestBaseline:
    """回归测试基线管理"""

    def __init__(self, baseline_file: str = "test_baseline.json"):
        self.baseline_file = baseline_file
        self.baseline = self._load_baseline()

    def _load_baseline(self) -> dict:
        """加载基线"""
        try:
            with open(self.baseline_file, "r") as f:
                return json.load(f)
        except FileNotFoundError:
            return {}

    def save_baseline(self, metrics: dict):
        """保存新基线"""
        with open(self.baseline_file, "w") as f:
            json.dump(metrics, f, indent=2)
        print(f"基线已保存: {self.baseline_file}")

    def check_regression(self, current_metrics: dict, threshold: float = 0.05) -> list:
        """
        检查回归

        Args:
            current_metrics: 当前指标
            threshold: 允许的降低幅度(5%)

        Returns:
            回归列表：[(指标名, 基线值, 当前值, 降低%)]
        """
        regressions = []

        for metric_name, baseline_value in self.baseline.items():
            if metric_name not in current_metrics:
                continue

            current_value = current_metrics[metric_name]

            # 对于"越高越好"的指标(accuracy, success_rate等)
            if "accuracy" in metric_name or "success" in metric_name or "rate" in metric_name:
                if current_value < baseline_value * (1 - threshold):
                    degradation = (baseline_value - current_value) / baseline_value
                    regressions.append((
                        metric_name,
                        baseline_value,
                        current_value,
                        degradation * 100
                    ))

            # 对于"越低越好"的指标(latency, cost等)
            elif "time" in metric_name or "cost" in metric_name or "latency" in metric_name:
                if current_value > baseline_value * (1 + threshold):
                    degradation = (current_value - baseline_value) / baseline_value
                    regressions.append((
                        metric_name,
                        baseline_value,
                        current_value,
                        degradation * 100
                    ))

        return regressions

# 使用示例
baseline_mgr = RegressionTestBaseline()

# 首次运行：保存基线
baseline_mgr.save_baseline({
    "success_rate": 0.95,
    "avg_tokens": 250,
    "avg_time_sec": 5.2
})

# 后续运行：检查回归
current = {
    "success_rate": 0.92,  # 降低了
    "avg_tokens": 280,     # 增加了
    "avg_time_sec": 4.8    # 降低了(好的)
}

regressions = baseline_mgr.check_regression(current)
if regressions:
    print("⚠ 检测到性能退化：")
    for metric, baseline, current, degradation in regressions:
        print(f"  {metric}: {baseline} → {current} (降低{degradation:.1f}%)")

# tests/test_e2e.py
import pytest
import asyncio
from pathlib import Path

class TestE2EWorkflows:
    """端到端工作流测试"""

    @pytest.mark.asyncio
    async def test_simple_file_read(self, sample_files, mock_tools):
        """简单的文件读取"""
        # 这个应该成功
        pass

    @pytest.mark.asyncio
    async def test_multi_step_workflow(self, sample_files, mock_tools):
        """多步工作流"""
        # 读取 → 分析 → 写入
        pass

    @pytest.mark.asyncio
    async def test_error_recovery(self, sample_files, mock_tools):
        """错误恢复测试"""
        # 工具调用失败后是否能恢复
        pass

    @pytest.mark.asyncio
    async def test_edge_cases(self, sample_files):
        """边界情况"""
        # 空文件、大文件、特殊字符等
        pass

    def test_token_efficiency(self):
        """Token效率测试"""
        # 平均调用数、Token使用
        pass

    def test_performance_baseline(self):
        """性能基线测试"""
        # 与基线对比
        pass

import json
import random
from pathlib import Path

class GAIABenchmark:
    """GAIA基准实现"""

    def __init__(self, data_dir: str = "./data/gaia"):
        self.data_dir = Path(data_dir)
        self.tasks = self._load_tasks()

    def _load_tasks(self) -> dict:
        """加载GAIA任务"""
        tasks = {"level1": [], "level2": [], "level3": []}

        for level in ["level1", "level2", "level3"]:
            level_dir = self.data_dir / level
            if level_dir.exists():
                for task_file in level_dir.glob("*.json"):
                    with open(task_file) as f:
                        task = json.load(f)
                        tasks[level].append(task)

        return tasks

    def get_sample(self, level: str, size: int = 10) -> list:
        """获取该等级的样本"""
        tasks = self.tasks.get(level, [])
        return random.sample(tasks, min(size, len(tasks)))

    async def run_evaluation(self, agent, level: str, size: int = 10):
        """在该等级上运行评估"""
        tasks = self.get_sample(level, size)
        if not tasks:
            return {
                "level": level,
                "success_rate": 0.0,
                "total": 0,
                "results": [],
                "warning": f"No GAIA tasks found for {level} in {self.data_dir}",
            }
        results = []

        for task in tasks:
            try:
                # 执行任务
                result = await agent.execute(
                    instruction=task["instruction"],
                    ground_truth=task.get("ground_truth")
                )

                # 评估
                is_correct = self._evaluate_task(result, task)
                results.append({
                    "task_id": task["id"],
                    "correct": is_correct,
                    "result": result
                })
            except Exception as e:
                results.append({
                    "task_id": task["id"],
                    "correct": False,
                    "error": str(e)
                })

        # 计算成功率
        success_rate = sum(1 for r in results if r["correct"]) / len(results)
        return {
            "level": level,
            "success_rate": success_rate,
            "total": len(results),
            "results": results
        }

    def _evaluate_task(self, result, task) -> bool:
        """评估任务是否正确"""
        # 简化:字符串匹配
        ground_truth = task.get("ground_truth", "")
        return ground_truth.lower() in str(result).lower()

# GAIA基准评估使用示例
benchmark = GAIABenchmark()

# 在Level 1上测试
level1_results = asyncio.run(benchmark.run_evaluation(agent, "level1", size=10))
print(f"Level 1 Success Rate: {level1_results['success_rate']:.1%}")

class WebArenaBenchmark:
    """WebArena基准实现"""

    def __init__(self, domain: str = "ecommerce"):
        """
        domain: 'ecommerce', 'social_media', 'forum', 'government'
        """
        self.domain = domain
        self.tasks = self._get_sample_tasks()

    def _get_sample_tasks(self) -> list:
        """获取示例任务"""
        tasks = {
            "ecommerce": [
                {
                    "id": "shop_001",
                    "instruction": "在Amazon上搜索'笔记本电脑',按价格从低到高排序",
                    "expected_steps": ["navigate", "search", "sort"],
                    "ground_truth": "sorted by price ascending"
                },
                {
                    "id": "shop_002",
                    "instruction": "添加商品到购物车并查看总价",
                    "expected_steps": ["add_to_cart", "view_cart"],
                    "ground_truth": "total price displayed"
                }
            ],
            "social_media": [
                {
                    "id": "social_001",
                    "instruction": "在Twitter上发布一条包含#AgentBench标签的推文",
                    "expected_steps": ["navigate", "compose", "post"],
                    "ground_truth": "tweet posted"
                }
            ]
        }
        return tasks.get(self.domain, [])

    async def run_evaluation(self, agent, size: int = 5):
        """运行评估"""
        tasks = self.tasks[:size]
        results = []

        for task in tasks:
            try:
                # 模拟网页交互
                trajectory = await agent.interact_with_web(
                    instruction=task["instruction"]
                )

                # 检查是否完成了预期步骤
                completed_steps = [step for step in task["expected_steps"]
                                 if step in [call.tool_name for call in trajectory]]

                success = len(completed_steps) == len(task["expected_steps"])

                results.append({
                    "task_id": task["id"],
                    "success": success,
                    "steps_completed": len(completed_steps),
                    "steps_expected": len(task["expected_steps"])
                })
            except Exception as e:
                results.append({
                    "task_id": task["id"],
                    "success": False,
                    "error": str(e)
                })

        success_rate = sum(1 for r in results if r["success"]) / len(results)
        return {
            "domain": self.domain,
            "success_rate": success_rate,
            "total": len(results),
            "results": results
        }

import json
from datetime import datetime
import pandas as pd

class BenchmarkReport:
    """基准评估报告"""

    def __init__(self, agent_name: str):
        self.agent_name = agent_name
        self.results = []
        self.timestamp = datetime.now().isoformat()

    def add_result(self, benchmark_name: str, metrics: dict):
        """添加评估结果"""
        self.results.append({
            "benchmark": benchmark_name,
            "metrics": metrics,
            "timestamp": datetime.now().isoformat()
        })

    def generate_report(self, output_file: str = "benchmark_report.json"):
        """生成报告"""
        report = {
            "agent_name": self.agent_name,
            "generated_at": self.timestamp,
            "benchmarks": self.results,
            "summary": self._compute_summary()
        }

        with open(output_file, "w") as f:
            json.dump(report, f, indent=2)

        return report

    def _compute_summary(self) -> dict:
        """计算汇总统计"""
        success_rates = [r["metrics"].get("success_rate", 0) for r in self.results]

        return {
            "avg_success_rate": sum(success_rates) / len(success_rates) if success_rates else 0,
            "best_benchmark": max(self.results, key=lambda x: x["metrics"].get("success_rate", 0))
                             ["benchmark"] if self.results else None,
            "benchmarks_count": len(self.results)
        }

    def to_dataframe(self) -> pd.DataFrame:
        """转为DataFrame用于分析"""
        data = []
        for result in self.results:
            row = {"benchmark": result["benchmark"]}
            row.update(result["metrics"])
            data.append(row)
        return pd.DataFrame(data)

    def print_summary(self):
        """打印汇总"""
        summary = self._compute_summary()
        print(f"\n{'='*60}")
        print(f"Agent: {self.agent_name}")
        print(f"Generated: {self.timestamp}")
        print(f"{'='*60}")
        print(f"\nBenchmark Results:")
        print(f"{'Benchmark':<20} {'Success Rate':<15}")
        print(f"{'-'*35}")

        for result in self.results:
            benchmark = result["benchmark"]
            sr = result["metrics"].get("success_rate", 0)
            print(f"{benchmark:<20} {sr:>6.1%}")

        print(f"\n{'='*60}")
        print(f"Average Success Rate: {summary['avg_success_rate']:.1%}")
        print(f"Best Performing: {summary['best_benchmark']}")
        print(f"{'='*60}\n")

# 综合基准评估使用示例
report = BenchmarkReport("MiniHarness-v1.0")

# 评估GAIA
gaia_benchmark = GAIABenchmark()
level1_result = asyncio.run(gaia_benchmark.run_evaluation(agent, "level1", size=10))
report.add_result("GAIA Level 1", {"success_rate": level1_result["success_rate"]})

# 评估WebArena
web_benchmark = WebArenaBenchmark("ecommerce")
web_result = asyncio.run(web_benchmark.run_evaluation(agent, size=5))
report.add_result("WebArena E-commerce", {"success_rate": web_result["success_rate"]})

# 生成报告
report.generate_report()
report.print_summary()

class AgentStandardsReadiness:
    """示意性的智能体标准化准备度检查"""

    def __init__(self):
        self.requirements = {
            "functional_classification": False,
            "safety_assessment": False,
            "performance_evaluation": False,
            "reliability_assessment": False,
            "documentation": False,
            "testing_coverage": False,
        }

    def check_compliance(self, system) -> dict:
        """检查系统合规性"""

        results = {
            "functional_classification": self._check_function_classification(system),
            "safety_assessment": self._check_safety(system),
            "performance_evaluation": self._check_performance(system),
            "reliability_assessment": self._check_reliability(system),
            "documentation": self._check_documentation(system),
            "testing_coverage": self._check_testing(system),
        }

        compliance_score = sum(results.values()) / len(results)
        results["overall_compliance"] = compliance_score

        return results

    def _check_function_classification(self, system) -> float:
        """功能分类评估"""
        # 检查系统是否有明确的任务分类
        return 0.8 if hasattr(system, "task_categories") else 0.0

    def _check_safety(self, system) -> float:
        """安全评估"""
        # 检查是否有安全防护
        has_permissions = hasattr(system, "permission_engine")
        has_sandboxing = hasattr(system, "sandbox")
        return (has_permissions + has_sandboxing) / 2

    def _check_performance(self, system) -> float:
        """性能评估"""
        # 检查是否有性能指标
        return 0.7 if hasattr(system, "metrics") else 0.0

    def _check_reliability(self, system) -> float:
        """可靠性评估"""
        # 检查是否有错误处理和恢复机制
        has_error_handling = hasattr(system, "error_handler")
        has_recovery = hasattr(system, "recovery_mechanism")
        return (has_error_handling + has_recovery) / 2

    def _check_documentation(self, system) -> float:
        """文档检查"""
        # 检查是否有完整文档
        return 0.8  # 假设有文档

    def _check_testing(self, system) -> float:
        """测试覆盖率检查"""
        # 检查是否有测试套件
        return 0.75 if hasattr(system, "test_suite") else 0.0

# 使用
compliance = AgentStandardsReadiness()
results = compliance.check_compliance(harness_system)

print(f"智能体标准化准备度评分: {results['overall_compliance']:.1%}")
for criterion, score in results.items():
    if criterion != "overall_compliance":
        print(f"  {criterion}: {score:.1%}")

from dataclasses import dataclass
from datetime import datetime
from typing import List, Dict

@dataclass
class ProductionMetrics:
    """生产环境关键指标"""

    # 可用性指标
    uptime_percent: float              # 系统可用性
    error_rate: float                  # 错误率
    crash_rate: float                  # 崩溃率

    # 质量指标
    success_rate: float                # 任务成功率
    avg_task_duration_sec: float       # 平均任务耗时
    timeout_rate: float                # 超时率

    # 效率指标
    avg_tokens_per_task: float         # 平均Token消耗
    tool_call_accuracy: float          # 工具调用准确率
    error_recovery_rate: float         # 错误恢复率

    # 用户满意度
    user_satisfaction_score: float     # 用户满意度评分
    complaint_count: int               # 投诉数

    # 成本指标
    total_cost_usd: float              # 总成本
    cost_per_successful_task: float    # 每个成功任务的成本

    timestamp: datetime = None

class MetricsCollector:
    """指标收集器"""

    def __init__(self, window_size_minutes: int = 60):
        self.window_size = window_size_minutes * 60  # 转为秒
        self.metrics_buffer = []

    def record_execution(self,
                        task_id: str,
                        success: bool,
                        duration_sec: float,
                        tokens_used: int,
                        cost_usd: float,
                        tool_calls: int,
                        correct_tool_calls: int):
        """记录一次执行"""

        self.metrics_buffer.append({
            "task_id": task_id,
            "success": success,
            "duration": duration_sec,
            "tokens": tokens_used,
            "cost": cost_usd,
            "tool_calls": tool_calls,
            "correct_calls": correct_tool_calls,
            "timestamp": datetime.now()
        })

    def compute_metrics(self) -> ProductionMetrics:
        """计算当前指标"""

        if not self.metrics_buffer:
            return None

        # 筛选时间窗口内的数据
        now = datetime.now()
        current_data = [
            m for m in self.metrics_buffer
            if (now - m["timestamp"]).total_seconds() < self.window_size
        ]

        if not current_data:
            return None

        # 计算各项指标
        total = len(current_data)
        successful = sum(1 for m in current_data if m["success"])
        errors = total - successful

        success_rate = successful / total if total > 0 else 0
        error_rate = errors / total if total > 0 else 0

        avg_duration = sum(m["duration"] for m in current_data) / total if total > 0 else 0
        avg_tokens = sum(m["tokens"] for m in current_data) / total if total > 0 else 0
        total_cost = sum(m["cost"] for m in current_data)

        # 工具调用准确率
        total_calls = sum(m["tool_calls"] for m in current_data)
        correct_calls = sum(m["correct_calls"] for m in current_data)
        tool_accuracy = correct_calls / total_calls if total_calls > 0 else 0

        cost_per_success = total_cost / successful if successful > 0 else float('inf')

        return ProductionMetrics(
            uptime_percent=99.9,  # 从基础设施监控获取
            error_rate=error_rate,
            crash_rate=0.0,  # 从日志获取
            success_rate=success_rate,
            avg_task_duration_sec=avg_duration,
            timeout_rate=sum(1 for m in current_data if m["duration"] > 30) / total,
            avg_tokens_per_task=avg_tokens,
            tool_call_accuracy=tool_accuracy,
            error_recovery_rate=0.8,  # 需要额外逻辑计算
            user_satisfaction_score=0.0,  # 需要反馈收集
            complaint_count=0,
            total_cost_usd=total_cost,
            cost_per_successful_task=cost_per_success,
            timestamp=now
        )

from scipy import stats
import numpy as np

class AnomalyDetector:
    """异常检测"""

    def __init__(self, history_window: int = 100):
        self.history_window = history_window
        self.history = {
            "success_rate": [],
            "avg_duration": [],
            "error_rate": [],
            "tokens_per_task": []
        }

    def add_observation(self,
                       success_rate: float,
                       avg_duration: float,
                       error_rate: float,
                       tokens_per_task: float):
        """添加观测值"""

        self.history["success_rate"].append(success_rate)
        self.history["avg_duration"].append(avg_duration)
        self.history["error_rate"].append(error_rate)
        self.history["tokens_per_task"].append(tokens_per_task)

        # 保持历史窗口大小
        for key in self.history:
            if len(self.history[key]) > self.history_window:
                self.history[key].pop(0)

    def detect_anomalies(self) -> Dict[str, bool]:
        """检测异常"""

        anomalies = {}

        for metric, values in self.history.items():
            if len(values) < 10:  # 需要足够的历史数据
                continue

            # 使用Z-score检测异常
            if len(values) > 0:
                z_scores = np.abs(stats.zscore(values[-10:]))  # 最近10个数据
                current_z = z_scores[-1]

                # Z-score > 2.5 认为异常
                anomalies[metric] = current_z > 2.5

        return anomalies

    def get_recommendation(self, anomalies: Dict[str, bool]) -> str:
        """基于异常给出建议"""

        if not any(anomalies.values()):
            return "系统状态正常"

        recommendations = []

        if anomalies.get("success_rate"):
            recommendations.append("成功率异常下降,建议检查模型更新或工具配置")

        if anomalies.get("avg_duration"):
            recommendations.append("执行时间异常增加,建议检查系统负载或网络延迟")

        if anomalies.get("error_rate"):
            recommendations.append("错误率异常升高,建议查看错误日志")

        if anomalies.get("tokens_per_task"):
            recommendations.append("Token消耗异常增加,建议优化prompt工程")

        return "; ".join(recommendations)

# 异常检测使用示例
detector = AnomalyDetector()

# 模拟一段时间的正常操作
for _ in range(50):
    detector.add_observation(
        success_rate=0.92 + np.random.normal(0, 0.02),
        avg_duration=5.0 + np.random.normal(0, 0.5),
        error_rate=0.08 + np.random.normal(0, 0.02),
        tokens_per_task=250 + np.random.normal(0, 20)
    )

# 模拟一个异常事件
for _ in range(5):
    detector.add_observation(
        success_rate=0.70,  # 异常下降
        avg_duration=8.5,   # 异常增加
        error_rate=0.30,    # 异常增加
        tokens_per_task=400 # 异常增加
    )

anomalies = detector.detect_anomalies()
print(f"检测到异常: {anomalies}")
print(f"建议: {detector.get_recommendation(anomalies)}")