Agent能力评估框架：从单任务到多任务综合评测
为什么Agent评估比LLM评估更难 LLM评估可以简化为"输入→输出→对比"，因为LLM是单轮的。但Agent是多步、多工具、有状态的，评估维度爆炸式增长： 不仅看最终答案对不对，还要看过程是否合理 不仅看单任务表现，还要看跨任务迁移能力 不仅看成功率，还要看效率、成本、安全性 本文构建一个从单任务到多任务的综合评估框架。 第一层：单任务评测 任务级指标矩阵 from dataclasses import dataclass, field from typing import Any from enum import Enum class MetricCategory(Enum): ACCURACY = "准确性" EFFICIENCY = "效率" SAFETY = "安全性" ROBUSTNESS = "鲁棒性" @dataclass class TaskMetric: name: str category: MetricCategory value: float weight: float = 1.0 description: str = "" @dataclass class TaskResult: task_id: str task_type: str success: bool metrics: list[TaskMetric] = field(default_factory=list) steps_taken: int = 0 tools_used: list[str] = field(default_factory=list) error_log: list[str] = field(default_factory=list) def weighted_score(self) -> float: total_weight = sum(m.weight for m in self.metrics) weighted_sum = sum(m.value * m.weight for m in self.metrics) return weighted_sum / total_weight if total_weight > 0 else 0.0 核心指标定义 指标 计算方式 说明 任务成功率 成功次数/总次数 基础指标 步骤效率 最优步数/实际步数 0-1，越高越好 工具选择准确率 正确工具调用/总调用 反映工具使用能力 格式合规率 格式正确输出/总输出 结构化输出能力 错误恢复率 恢复成功次数/错误次数 容错能力 成本效率 最优成本/实际成本 token消耗评估 安全违规率 违规次数/总次数 越低越好 评测执行框架 import asyncio from abc import ABC, abstractmethod class TaskEvaluator(ABC): @abstractmethod async def evaluate(self, agent, task) -> TaskResult: pass class CodeGenerationEvaluator(TaskEvaluator): async def evaluate(self, agent, task) -> TaskResult: # 执行Agent result = await agent.run(task["prompt"]) metrics = [] # 1. 功能正确性（单元测试通过率） test_pass = await self._run_tests(result.code, task["test_cases"]) metrics.append(TaskMetric( name="functional_correctness", category=MetricCategory.ACCURACY, value=test_pass, weight=2.0 )) # 2. 步骤效率 optimal_steps = task.get("optimal_steps", 3) actual_steps = result.steps efficiency = min(optimal_steps / actual_steps, 1.0) if actual_steps > 0 else 0 metrics.append(TaskMetric( name="step_efficiency", category=MetricCategory.EFFICIENCY, value=efficiency, weight=1.0 )) # 3. 代码质量（LLM-as-Judge） quality_score = await self._llm_judge( result.code, criteria=["可读性", "性能", "安全性"] ) metrics.append(TaskMetric( name="code_quality", category=MetricCategory.ACCURACY, value=quality_score, weight=1.5 )) # 4. 安全检查 violations = self._check_safety(result.code) metrics.append(TaskMetric( name="safety_compliance", category=MetricCategory.SAFETY, value=1.0 - violations / max(len(result.code.split("\n")), 1), weight=1.0 )) return TaskResult( task_id=task["id"], task_type="code_generation", success=test_pass > 0.8, metrics=metrics, steps_taken=actual_steps, tools_used=result.tools_used, error_log=result.errors ) async def _run_tests(self, code: str, test_cases: list) -> float: passed = 0 for tc in test_cases: try: result = self._execute_code(code, tc["input"]) if result == tc["expected"]: passed += 1 except Exception: pass return passed / len(test_cases) if test_cases else 0 async def _llm_judge(self, code: str, criteria: list[str]) -> float: prompt = f"评估以下代码质量，维度{criteria}，给出0-1的分数:\n{code}" score = await judge_llm(prompt) return score 第二层：多任务综合评测 能力维度模型 class CapabilityModel: """Agent能力维度定义""" CAPABILITIES = { "reasoning": "逻辑推理", "coding": "代码生成", "extraction": "信息抽取", "planning": "任务规划", "tool_use": "工具使用", "creativity": "创意生成", "safety": "安全合规", "multilingual": "多语言能力" } def __init__(self): self.dimension_scores: dict[str, list[float]] = { dim: [] for dim in self.CAPABILITIES } def add_result(self, capability: str, score: float): if capability in self.dimension_scores: self.dimension_scores[capability].append(score) def aggregate(self) -> dict[str, float]: return { dim: sum(scores) / len(scores) if scores else 0.0 for dim, scores in self.dimension_scores.items() } 雷达图生成 import numpy as np class RadarChart: def __init__(self, capabilities: dict[str, float]): self.capabilities = capabilities self.angles = np.linspace(0, 2 * np.pi, len(capabilities), endpoint=False).tolist() self.angles += self.angles[:1] def to_plotly_data(self) -> dict: values = list(self.capabilities.values()) values += values[:1] return { "type": "scatterpolar", "r": values, "theta": list(self.capabilities.keys()) + [list(self.capabilities.keys())[0]], "fill": "toself", "name": "Agent能力" } 跨任务迁移评测 class TransferEvaluator: """评估Agent的跨任务迁移能力""" async def evaluate_transfer(self, agent, source_task: dict, target_task: dict) -> float: """在源任务上训练/调整后，在目标任务上的表现""" # 1. 记录基线表现 baseline = await self._run_task(agent, target_task) # 2. 在源任务上的经验 await self._run_task(agent, source_task) # 3. 再测目标任务 after = await self._run_task(agent, target_task) # 4. 迁移增益 transfer_gain = after - baseline # 5. 归一化 max_possible = 1.0 - baseline normalized = transfer_gain / max_possible if max_possible > 0 else 0 return normalized 第三层：系统级评估 长期稳定性评测 from collections import defaultdict import statistics class StabilityMonitor: def __init__(self, window_size: int = 100): self.window_size = window_size self.results: dict[str, list[float]] = defaultdict(list) def record(self, task_type: str, score: float): self.results[task_type].append(score) if len(self.results[task_type]) > self.window_size: self.results[task_type].pop(0) def get_stability_metrics(self) -> dict[str, dict]: metrics = {} for task_type, scores in self.results.items(): if len(scores) < 10: continue metrics[task_type] = { "mean": statistics.mean(scores), "stdev": statistics.stdev(scores), "min": min(scores), "max": max(scores), "cv": statistics.stdev(scores) / statistics.mean(scores) if statistics.mean(scores) > 0 else 0, "trend": self._trend(scores) } return metrics def _trend(self, scores: list[float]) -> str: if len(scores) < 5: return "insufficient_data" first_half = statistics.mean(scores[:len(scores)//2]) second_half = statistics.mean(scores[len(scores)//2:]) if second_half > first_half * 1.05: return "improving" elif second_half < first_half * 0.95: return "declining" return "stable" 评测报告模板 def generate_eval_report(agent_name: str, results: list[TaskResult]) -> str: capability = CapabilityModel() for r in results: # 映射任务类型到能力维度 dim_map = { "code_generation": "coding", "logical_reasoning": "reasoning", "information_extraction": "extraction", "task_planning": "planning" } dim = dim_map.get(r.task_type, "reasoning") capability.add_result(dim, r.weighted_score()) scores = capability.aggregate() report = f"""# Agent评测报告: {agent_name} ## 总览 - 评测任务数: {len(results)} - 平均得分: {sum(r.weighted_score() for r in results)/len(results):.2f} - 整体成功率: {sum(r.success for r in results)/len(results):.1%} ## 能力雷达 {scores} ## 详细指标 | 任务类型 | 成功率 | 平均步骤 | 平均得分 | |----------|--------|----------|----------| """ by_type = {} for r in results: by_type.setdefault(r.task_type, []).append(r) for ttype, task_results in by_type.items(): success_rate = sum(r.success for r in task_results) / len(task_results) avg_steps = sum(r.steps_taken for r in task_results) / len(task_results) avg_score = sum(r.weighted_score() for r in task_results) / len(task_results) report += f"| {ttype} | {success_rate:.1%} | {avg_steps:.1f} | {avg_score:.2f} |\n" return report 评测集构建原则 原则 说明 示例 覆盖性 覆盖所有能力维度 每个维度≥20题 难度梯度 简单/中等/困难均匀 3:5:2比例 防污染 避免训练数据泄露 用私有数据集 可扩展 支持动态新增 模块化任务定义 可复现 固定随机种子 temperature=0 总结 Agent评估是一个多层次工程：单任务看指标，多任务看雷达，长期看稳定性。核心原则： ...
