1
@dataclass
2
class TaskResult:
3
    task_id: str
4
    user_intent: str
5
    agent_response: str
6
    tools_used: list[str]
7
    completed: bool  # 人工标注
8
    partial_completed: bool
9

10
    # 自动指标
11
    intent_match_score: float  # 0-1
12
    key_points_covered: list[str]

1
def evaluate_task_completion(result: TaskResult) -> dict:
2
    """自动化评估指标"""
3
    metrics = {
4
        # 1. 关键实体覆盖
5
        "entity_recall": len(result.covered_entities) / len(result.required_entities),
6

7
        # 2. 意图匹配
8
        "intent_match": result.intent_match_score,
9

10
        # 3. 工具调用准确率
11
        "tool_accuracy": sum(result.correct_tool_calls) / len(result.tool_calls),
12

13
        # 4. 幻觉检测
14
        "hallucination_score": 1 - detect_hallucination_rate(result.response)
15
    }
16
    return metrics

1
@dataclass
2
class TaskScoreWeights:
3
    """不同场景的权重配置"""
4
    intent_match: float = 0.3
5
    entity_recall: float = 0.25
6
    tool_accuracy: float = 0.25
7
    factuality: float = 0.2
8

9
# 不同场景的权重
10
WEIGHTS = {
11
    "research": TaskScoreWeights(intent_match=0.2, entity_recall=0.35, tool_accuracy=0.15, factuality=0.3),
12
    "coding": TaskScoreWeights(intent_match=0.3, entity_recall=0.1, tool_accuracy=0.4, factuality=0.2),
13
    "customer_support": TaskScoreWeights(intent_match=0.4, entity_recall=0.2, tool_accuracy=0.2, factuality=0.2),
14
}
15

16
def compute_task_score(result: TaskResult, scenario: str = "research") -> float:
17
    weights = WEIGHTS[scenario]
18
    metrics = evaluate_task_completion(result)
19
    return (
20
        weights.intent_match * metrics["intent_match"]
21
        + weights.entity_recall * metrics["entity_recall"]
22
        + weights.tool_accuracy * metrics["tool_accuracy"]
23
        + weights.factuality * metrics["hallucination_score"]
24
    )

1
class FactualityChecker:
2
    """检查 Agent 回答中的事实准确性"""
3

4
    def __init__(self, knowledge_base):
5
        self.kb = knowledge_base
6

7
    async def check(self, response: str, references: list[str]) -> dict:
8
        """将回答中的每个声明与参考来源对照"""
9
        claims = await self._extract_claims(response)
10
        results = []
11

12
        for claim in claims:
13
            evidence = await self.kb.search(claim, k=3)
14
            is_supported = await self._verify_claim(claim, evidence)
15
            results.append({
16
                "claim": claim,
17
                "supported": is_supported,
18
                "evidence": evidence[:1] if evidence else None,
19
            })
20

21
        supported_count = sum(1 for r in results if r["supported"])
22
        return {
23
            "total_claims": len(results),
24
            "supported": supported_count,
25
            "factuality_score": supported_count / len(results) if results else 0,
26
            "details": results,
27
        }
28

29
    async def _extract_claims(self, text: str) -> list[str]:
30
        """从回答中提取可验证的声明"""
31
        extraction_prompt = f"""从以下文本中提取可验证的事实声明。
32
每个声明应该是独立、可核验的句子。
33

34
文本: {text}
35

36
输出 JSON 列表。"""
37
        response = await llm.complete(extraction_prompt)
38
        return parse_json_list(response)
39

40
    async def _verify_claim(self, claim: str, evidence: list) -> bool:
41
        """判断声明是否有证据支持"""
42
        verification_prompt = f"""判断以下声明是否被证据支持。
43

44
声明: {claim}
45
证据: {evidence}
46

47
回答 "supported" 或 "not_supported"。"""
48
        result = await llm.complete(verification_prompt)
49
        return "supported" in result.lower()

1
def evaluate_tool_calls(ground_truth: list, predictions: list) -> dict:
2
    """评估工具调用准确率"""
3
    return {
4
        "precision": len(set(ground_truth) & set(predictions)) / len(set(predictions)),
5
        "recall": len(set(ground_truth) & set(predictions)) / len(set(ground_truth)),
6
        "order_accuracy": ground_truth == predictions,
7
    }

1
@dataclass
2
class ToolCallEvaluation:
3
    """工具调用全链路评估"""
4
    # 1. 工具选择是否正确
5
    correct_tool_selected: bool
6
    # 2. 参数是否完整
7
    all_params_filled: bool
8
    # 3. 参数值是否正确
9
    correct_param_values: dict[str, bool]  # param_name -> correct?
10
    # 4. 调用时机是否合理
11
    call_timing_appropriate: bool  # 是否在不该调用时调用了
12
    # 5. 结果处理是否得当
13
    result_utilized: bool  # 工具返回的结果是否被正确利用
14

15
def evaluate_tool_call_chain(
16
    agent_trace: list[dict],
17
    ground_truth: list[dict]
18
) -> dict:
19
    """评估整个工具调用链"""
20
    metrics = {
21
        "tool_selection_accuracy": 0.0,
22
        "param_completeness": 0.0,
23
        "param_accuracy": 0.0,
24
        "ordering_correctness": 0.0,
25
        "unnecessary_calls": 0,
26
    }
27

28
    # 逐步骤对比
29
    for i, (actual, expected) in enumerate(zip(agent_trace, ground_truth)):
30
        if actual.get("tool") == expected.get("tool"):
31
            metrics["tool_selection_accuracy"] += 1
32

33
            # 检查参数
34
            actual_params = actual.get("params", {})
35
            expected_params = expected.get("params", {})
36

37
            filled = sum(1 for k in expected_params if k in actual_params)
38
            metrics["param_completeness"] += filled / len(expected_params)
39

40
            correct = sum(
41
                1 for k in expected_params
42
                if actual_params.get(k) == expected_params[k]
43
            )
44
            metrics["param_accuracy"] += correct / len(expected_params)
45

46
    n = len(ground_truth)
47
    metrics["tool_selection_accuracy"] /= n
48
    metrics["param_completeness"] /= n
49
    metrics["param_accuracy"] /= n
50

51
    # 检查多余的调用
52
    metrics["unnecessary_calls"] = max(0, len(agent_trace) - len(ground_truth))
53

54
    return metrics

1
@dataclass
2
class CostMetrics:
3
    prompt_tokens: int
4
    completion_tokens: int
5
    total_tokens: int
6
    latency_ms: int
7
    cost_usd: float
8

9
def calculate_cost(result: TaskResult) -> CostMetrics:
10
    """计算任务成本"""
11
    return CostMetrics(
12
        prompt_tokens=result.prompt_tokens,
13
        completion_tokens=result.completion_tokens,
14
        total_tokens=result.total_tokens,
15
        latency_ms=result.latency_ms,
16
        cost_usd=calculate_cost_usd(result.total_tokens)
17
    )

1
from datetime import datetime
2

3
@dataclass
4
class CostRecord:
5
    """单次调用的成本记录"""
6
    timestamp: datetime
7
    agent_name: str
8
    task_type: str
9
    model: str
10
    prompt_tokens: int
11
    completion_tokens: int
12
    latency_ms: float
13
    cost_usd: float
14
    success: bool
15

16
class CostTracker:
17
    """成本追踪器"""
18

19
    def __init__(self):
20
        self.records: list[CostRecord] = []
21

22
    def record(self, **kwargs):
23
        self.records.append(CostRecord(timestamp=datetime.now(), **kwargs))
24

25
    def summary(self, group_by: str = "agent_name") -> dict:
26
        """按维度汇总成本"""
27
        groups = {}
28
        for r in self.records:
29
            key = getattr(r, group_by)
30
            if key not in groups:
31
                groups[key] = {"total_cost": 0, "total_tokens": 0, "count": 0, "success": 0}
32
            groups[key]["total_cost"] += r.cost_usd
33
            groups[key]["total_tokens"] += r.prompt_tokens + r.completion_tokens
34
            groups[key]["count"] += 1
35
            groups[key]["success"] += 1 if r.success else 0
36
        return groups
37

38
    def cost_per_success(self) -> float:
39
        """计算每次成功任务的平均成本"""
40
        successful = [r for r in self.records if r.success]
41
        if not successful:
42
            return float("inf")
43
        return sum(r.cost_usd for r in successful) / len(successful)

1
# 常见模型定价（2026年初参考价格，单位: USD / 1M tokens）
2
MODEL_PRICING = {
3
    "gpt-4o": {"input": 2.50, "output": 10.00},
4
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
5
    "claude-sonnet-4": {"input": 3.00, "output": 15.00},
6
    "claude-haiku-3.5": {"input": 0.80, "output": 4.00},
7
    "deepseek-chat": {"input": 0.14, "output": 0.28},
8
    "deepseek-reasoner": {"input": 0.55, "output": 2.19},
9
    "gemini-2.0-flash": {"input": 0.10, "output": 0.40},
10
    "gemini-2.5-pro": {"input": 1.25, "output": 10.00},
11
}
12

13
def estimate_monthly_cost(
14
    daily_requests: int,
15
    avg_prompt_tokens: int = 2000,
16
    avg_completion_tokens: int = 500,
17
    avg_tool_calls_per_request: int = 3,
18
    model: str = "gpt-4o",
19
) -> dict:
20
    """估算月度成本"""
21
    pricing = MODEL_PRICING[model]
22

23
    # 每次 LLM 调用的成本
24
    cost_per_call = (
25
        avg_prompt_tokens * pricing["input"] / 1_000_000
26
        + avg_completion_tokens * pricing["output"] / 1_000_000
27
    )
28

29
    # 每次请求的总 LLM 调用次数 = 初始调用 + 工具调用次数 * 2（调用+结果处理） + 最终汇总
30
    calls_per_request = 1 + avg_tool_calls_per_request * 2 + 1
31

32
    daily_cost = daily_requests * calls_per_request * cost_per_call
33
    monthly_cost = daily_cost * 30
34

35
    return {
36
        "model": model,
37
        "cost_per_call": cost_per_call,
38
        "calls_per_request": calls_per_request,
39
        "daily_cost_usd": daily_cost,
40
        "monthly_cost_usd": monthly_cost,
41
    }

1
class AgentBenchRunner:
2
    """AgentBench 评估运行器"""
3

4
    TASKS = [
5
        "os",           # 操作系统操作
6
        "webshop",      # 网购任务
7
        "web_browsing", # 网页浏览
8
        "digital_card", # 数字卡牌游戏
9
        "lateral_thinking", # 侧向思维谜题
10
        "householding", # 家务任务
11
        "textcraft",    # 文字冒险
12
        "mind2web",     # 网页任务执行
13
    ]
14

15
    async def evaluate_agent(self, agent, tasks: list[str] | None = None) -> dict:
16
        tasks = tasks or self.TASKS
17
        results = {}
18

19
        for task in tasks:
20
            task_env = self._create_env(task)
21
            score = 0
22
            total = 0
23

24
            for episode in task_env.episodes:
25
                obs = task_env.reset(episode)
26
                done = False
27
                steps = 0
28

29
                while not done and steps < task_env.max_steps:
30
                    action = await agent.act(obs, task_env.available_actions)
31
                    obs, reward, done, info = task_env.step(action)
32
                    score += reward
33
                    steps += 1
34

35
                total += 1
36

37
            results[task] = {
38
                "score": score,
39
                "total_episodes": total,
40
                "average_score": score / total if total > 0 else 0,
41
            }
42

43
        return results

1
class WebArenaEvaluator:
2
    """WebArena 评估器"""
3

4
    async def evaluate(self, agent, test_cases: list[dict]) -> dict:
5
        results = {
6
            "total": len(test_cases),
7
            "success": 0,
8
            "partial_success": 0,
9
            "failure": 0,
10
            "avg_steps": 0,
11
        }
12

13
        for case in test_cases:
14
            # 启动浏览器环境
15
            env = WebEnvironment(
16
                start_url=case["start_url"],
17
                intent=case["intent"],
18
                target_url=case.get("target_url"),
19
            )
20

21
            steps = 0
22
            done = False
23

24
            while not done and steps < 30:
25
                # 获取页面状态
26
                page_state = env.get_state()
27

28
                # Agent 决策
29
                action = await agent.decide(page_state, case["intent"])
30

31
                # 执行动作
32
                env.execute(action)
33
                steps += 1
34

35
                # 检查是否完成
36
                if env.check_completion(case["eval_criteria"]):
37
                    results["success"] += 1
38
                    done = True
39
                elif steps >= 30:
40
                    # 超时，检查部分完成
41
                    if env.check_partial_completion(case["eval_criteria"]):
42
                        results["partial_success"] += 1
43
                    else:
44
                        results["failure"] += 1
45

46
            results["avg_steps"] += steps
47

48
        results["avg_steps"] /= results["total"]
49
        results["success_rate"] = results["success"] / results["total"]
50
        return results

1
from agenteval import AgentEvaluator
2

3
evaluator = AgentEvaluator(
4
    tasks=load_benchmark("GAIA"),
5
    metrics=["task_completion", "tool_usage", "cost"]
6
)
7

8
results = evaluator.evaluate(your_agent)
9
print(f"Overall Score: {results.overall_score}")

1
import hashlib
2
from dataclasses import dataclass
3

4
@dataclass
5
class ABTestConfig:
6
    test_name: str
7
    variants: dict[str, float]  # variant_name -> traffic_ratio
8
    metrics: list[str]
9
    min_sample_size: int = 1000
10
    significance_level: float = 0.05
11

12
class ABTestEngine:
13
    """Agent A/B 测试引擎"""
14

15
    def __init__(self):
16
        self.results: dict[str, list[dict]] = {}
17
        self.active_tests: dict[str, ABTestConfig] = {}
18

19
    def register(self, config: ABTestConfig):
20
        self.active_tests[config.test_name] = config
21
        self.results[config.test_name] = {v: [] for v in config.variants}
22

23
    def assign_variant(self, test_name: str, user_id: str) -> str:
24
        """根据用户 ID 确定性分配变体"""
25
        config = self.active_tests[test_name]
26
        hash_val = int(hashlib.md5(f"{test_name}:{user_id}".encode()).hexdigest(), 16)
27
        bucket = (hash_val % 100) / 100
28

29
        cumulative = 0
30
        for variant, ratio in config.variants.items():
31
            cumulative += ratio
32
            if bucket <= cumulative:
33
                return variant
34

35
        return list(config.variants.keys())[-1]
36

37
    def record_result(self, test_name: str, variant: str, metrics: dict):
38
        """记录一次实验结果"""
39
        self.results[test_name][variant].append(metrics)
40

41
    def analyze(self, test_name: str) -> dict:
42
        """分析实验结果"""
43
        config = self.active_tests[test_name]
44
        report = {}
45

46
        for metric in config.metrics:
47
            variant_stats = {}
48
            for variant, data in self.results[test_name].items():
49
                values = [d[metric] for d in data if metric in d]
50
                variant_stats[variant] = {
51
                    "mean": sum(values) / len(values) if values else 0,
52
                    "std": self._std(values),
53
                    "n": len(values),
54
                }
55

56
            # t-test 检验显著性
57
            variants = list(variant_stats.keys())
58
            if len(variants) == 2:
59
                is_significant = self._t_test(
60
                    variant_stats[variants[0]],
61
                    variant_stats[variants[1]],
62
                    config.significance_level,
63
                )
64
            else:
65
                is_significant = False
66

67
            report[metric] = {
68
                "variant_stats": variant_stats,
69
                "is_significant": is_significant,
70
            }
71

72
        return report
73

74
    def _std(self, values: list[float]) -> float:
75
        if len(values) < 2:
76
            return 0
77
        mean = sum(values) / len(values)
78
        return (sum((v - mean) ** 2 for v in values) / (len(values) - 1)) ** 0.5
79

80
    def _t_test(self, a: dict, b: dict, alpha: float) -> bool:
81
        """简化的 t 检验"""
82
        from scipy import stats
83
        if a["n"] < 30 or b["n"] < 30:
84
            return False
85
        # 实际实现中使用 scipy.stats.ttest_ind 进行显著性检验
86
        return True  # 简化示例，生产环境使用完整的 t 检验

1
async def llm_judge评估(response: str, criteria: list[str]) -> float:
2
    """用更强 LLM 评估 Agent 输出"""
3
    judge_prompt = f"""
4
    评估以下 Agent 响应：
5

6
    响应：{response}
7

8
    评估标准：
9
    {chr(10).join(criteria)}
10

11
    给出 0-10 的评分和简要理由。
12
    """
13
    result = await judge_llm.complete(judge_prompt)
14
    return parse_score(result)

1
class RobustLLMJudge:
2
    """健壮的 LLM-as-Judge 评估器"""
3

4
    def __init__(self, judge_model: str = "claude-opus-4"):
5
        self.judge_model = judge_model
6
        self.calibration_samples = []  # 标定样本
7

8
    async def judge(
9
        self,
10
        query: str,
11
        response: str,
12
        reference: str | None = None,
13
        rubric: list[str] | None = None,
14
    ) -> dict:
15
        """评估单条 Agent 响应"""
16
        rubric_text = "\n".join(f"- {r}" for r in (rubric or self._default_rubric()))
17

18
        prompt = f"""你是一个专业的 Agent 输出评估专家。请严格按照评分标准评估。
19

20
用户问题: {query}
21

22
Agent 回答: {response}
23
{"参考答案: " + reference if reference else ""}
24

25
评分标准:
26
{rubric_text}
27

28
请以 JSON 格式输出:
29
{{
30
    "scores": {{
31
        "relevance": <0-10>,
32
        "accuracy": <0-10>,
33
        "completeness": <0-10>,
34
        "clarity": <0-10>,
35
        "safety": <0-10>
36
    }},
37
    "overall": <0-10>,
38
    "reasoning": "<简要说明评分理由>",
39
    "improvement_suggestions": ["<建议1>", "<建议2>"]
40
}}"""
41

42
        result = await llm.complete(prompt, model=self.judge_model)
43
        parsed = parse_json(result)
44

45
        return {
46
            "scores": parsed.get("scores", {}),
47
            "overall": parsed.get("overall", 0),
48
            "reasoning": parsed.get("reasoning", ""),
49
            "suggestions": parsed.get("improvement_suggestions", []),
50
        }
51

52
    async def judge_pairwise(
53
        self,
54
        query: str,
55
        response_a: str,
56
        response_b: str,
57
    ) -> dict:
58
        """成对比较评估：减少 LLM 评分的不稳定性"""
59
        prompt = f"""比较以下两个 Agent 对同一问题的回答，判断哪个更好。
60

61
问题: {query}
62

63
回答 A: {response_a}
64

65
回答 B: {response_b}
66

67
输出 JSON:
68
{{
69
    "winner": "A" 或 "B" 或 "tie",
70
    "confidence": <0.0-1.0>,
71
    "reasoning": "<简要说明>"
72
}}"""
73

74
        result = await llm.complete(prompt, model=self.judge_model)
75
        return parse_json(result)
76

77
    async def batch_judge(
78
        self,
79
        queries_and_responses: list[tuple[str, str]],
80
        reference_answers: list[str] | None = None,
81
    ) -> list[dict]:
82
        """批量评估，自动计算统计信息"""
83
        results = []
84
        for i, (query, response) in enumerate(queries_and_responses):
85
            ref = reference_answers[i] if reference_answers else None
86
            result = await self.judge(query, response, reference=ref)
87
            results.append(result)
88

89
        # 汇总统计
90
        overall_scores = [r["overall"] for r in results]
91
        return {
92
            "individual_results": results,
93
            "summary": {
94
                "mean_score": sum(overall_scores) / len(overall_scores),
95
                "min_score": min(overall_scores),
96
                "max_score": max(overall_scores),
97
                "score_distribution": self._compute_distribution(overall_scores),
98
            }
99
        }
100

101
    def _default_rubric(self) -> list[str]:
102
        return [
103
            "相关性: 回答是否切题",
104
            "准确性: 事实是否正确",
105
            "完整性: 是否覆盖了所有要点",
106
            "清晰度: 表达是否清晰易懂",
107
            "安全性: 是否包含有害内容",
108
        ]

1
class EvaluationPipeline:
2
    """自动化评估流水线"""
3

4
    def __init__(self, config: dict):
5
        self.judge = RobustLLMJudge(config.get("judge_model"))
6
        self.cost_tracker = CostTracker()
7
        self.benchmarks = config.get("benchmarks", [])
8

9
    async def evaluate(self, agent, test_set: list[dict]) -> dict:
10
        """运行完整评估"""
11
        results = {
12
            "task_metrics": [],
13
            "tool_metrics": [],
14
            "cost_metrics": [],
15
            "judge_scores": [],
16
        }
17

18
        for test_case in test_set:
19
            # 运行 Agent
20
            agent_result = await agent.run(test_case["query"])
21

22
            # 1. 任务完成指标
23
            task_metrics = evaluate_task_completion(agent_result)
24
            results["task_metrics"].append(task_metrics)
25

26
            # 2. 工具调用指标
27
            tool_metrics = evaluate_tool_call_chain(
28
                agent_result.trace,
29
                test_case.get("expected_tools", [])
30
            )
31
            results["tool_metrics"].append(tool_metrics)
32

33
            # 3. 成本指标
34
            cost = calculate_cost(agent_result)
35
            results["cost_metrics"].append(cost)
36

37
            # 4. LLM-as-Judge
38
            judge_result = await self.judge.judge(
39
                query=test_case["query"],
40
                response=agent_result.response,
41
                reference=test_case.get("reference_answer"),
42
            )
43
            results["judge_scores"].append(judge_result)
44

45
        # 汇总报告
46
        return self._generate_report(results)
47

48
    def _generate_report(self, results: dict) -> dict:
49
        """生成评估报告"""
50
        task_scores = [m.get("intent_match", 0) for m in results["task_metrics"]]
51
        judge_scores = [r["overall"] for r in results["judge_scores"]]
52
        costs = [c.cost_usd for c in results["cost_metrics"]]
53

54
        return {
55
            "overall_score": sum(judge_scores) / len(judge_scores),
56
            "task_completion_rate": sum(1 for s in task_scores if s >= 0.8) / len(task_scores),
57
            "avg_cost_per_task": sum(costs) / len(costs),
58
            "cost_efficiency": (sum(judge_scores) / len(judge_scores)) / (sum(costs) / len(costs)),
59
            "details": results,
60
        }

1
class EvaluationReport:
2
    """评估报告生成器"""
3

4
    def generate(self, results: dict) -> str:
5
        report = f"""
6
# Agent 评估报告
7

8
## 概要
9
- 评估时间: {datetime.now().isoformat()}
10
- 测试用例数: {results['total_cases']}
11
- 整体评分: {results['overall_score']:.2f} / 10
12

13
## 任务完成
14
| 指标 | 值 | 目标 | 状态 |
15
|------|-----|------|------|
16
| 完成率 | {results['completion_rate']:.1%} | > 85% | {'' if results['completion_rate'] > 0.85 else ''} |
17
| 意图匹配 | {results['intent_match']:.2f} | > 0.9 | {'' if results['intent_match'] > 0.9 else ''} |
18

19
## 工具使用
20
| 指标 | 值 | 目标 | 状态 |
21
|------|-----|------|------|
22
| 准确率 | {results['tool_accuracy']:.1%} | > 90% | {'' if results['tool_accuracy'] > 0.9 else ''} |
23
| 参数完整 | {results['param_completeness']:.1%} | > 95% | {'' if results['param_completeness'] > 0.95 else ''} |
24

25
## 成本效率
26
| 指标 | 值 | 目标 | 状态 |
27
|------|-----|------|------|
28
| 平均延迟 | {results['avg_latency_ms']:.0f}ms | < 3000ms | {'' if results['avg_latency_ms'] < 3000 else ''} |
29
| 每任务成本 | ${results['cost_per_task']:.4f} | < $0.10 | {'' if results['cost_per_task'] < 0.1 else ''} |
30
"""
31
        return report