为什么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评估是一个多层次工程：单任务看指标，多任务看雷达，长期看稳定性。核心原则：
- 多维度：不只是"做对了没"，还要看"做得好不好"
- 分层次：单任务→跨任务→系统级，逐层深入
- 可对比：标准化评测集 + 固定参数，确保结果可比较
- 防过拟合：定期更新评测集，防止Agent针对测试集优化
好的评估框架是Agent持续进步的基石。没有度量就没有改进。