AI Agent 评测与基准测试体系：从能力评估到安全红队的完整指南

# Agent 评测核心维度框架

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum

class EvaluationDimension(Enum):
    TASK_SUCCESS = "task_success"        # 任务完成率
    TOOL_ACCURACY = "tool_accuracy"      # 工具调用准确率
    PLANNING = "planning"                # 规划能力
    ROBUSTNESS = "robustness"            # 鲁棒性
    SAFETY = "safety"                    # 安全性
    COST_EFFICIENCY = "cost_efficiency"   # 成本效率

@dataclass
class EvaluationMetric:
    name: str
    dimension: EvaluationDimension
    weight: float            # 权重（0-1）
    target_threshold: float  # 达标阈值
    measurement_unit: str    # 度量单位

@dataclass
class AgentEvaluationProfile:
    """Agent 评测配置模板"""
    metrics: List[EvaluationMetric] = field(default_factory=lambda: [
        EvaluationMetric("任务完成率", EvaluationDimension.TASK_SUCCESS, 0.25, 0.85, "%"),
        EvaluationMetric("工具调用准确率", EvaluationDimension.TOOL_ACCURACY, 0.20, 0.90, "%"),
        EvaluationMetric("规划合理性", EvaluationDimension.PLANNING, 0.15, 0.75, "分"),
        EvaluationMetric("异常恢复率", EvaluationDimension.ROBUSTNESS, 0.15, 0.80, "%"),
        EvaluationMetric("安全事件零发生", EvaluationDimension.SAFETY, 0.15, 0.99, "%"),
        EvaluationMetric("单位任务成本", EvaluationDimension.COST_EFFICIENCY, 0.10, 0.50, "美元"),
    ])
    
    def weighted_score(self, results: Dict[str, float]) -> float:
        """计算加权综合评分"""
        total = 0.0
        for metric in self.metrics:
            if metric.name in results:
                total += results[metric.name] * metric.weight
        return total

# AgentBench 评测执行脚本示例

import json
import asyncio
from typing import Dict, List, Optional
from dataclasses import dataclass

@dataclass
class BenchmarkTask:
    task_id: str
    environment: str        # OS/DB/KG/Web/Shop 等
    description: str        # 任务描述
    expected_state: Dict    # 预期最终状态
    max_steps: int          # 最大执行步数
    difficulty: str         # easy/medium/hard

@dataclass
class BenchmarkResult:
    task: BenchmarkTask
    success: bool
    steps_taken: int
    tool_calls: List[Dict]  # 所有工具调用记录
    final_state: Dict
    score: float            # 0-1 部分得分

class AgentBenchRunner:
    """AgentBench 评测运行器"""
    
    def __init__(self, agent, environments: List[str]):
        self.agent = agent
        self.envs = environments
        self.results: List[BenchmarkResult] = []
    
    async def run_task(self, task: BenchmarkTask) -> BenchmarkResult:
        """执行单个评测任务"""
        state = await self._init_environment(task.environment)
        tool_calls = []
        steps = 0
        
        while steps < task.max_steps:
            action = await self.agent.decide(state, task.description)
            if action.type == "tool_call":
                result = await self._execute_tool(action)
                tool_calls.append(action.to_dict())
                state = self._update_state(state, result)
            elif action.type == "submit":
                break
            steps += 1
        
        success = self._verify_state(state, task.expected_state)
        score = self._calculate_partial_score(state, task.expected_state)
        
        return BenchmarkResult(
            task=task, success=success,
            steps_taken=steps, tool_calls=tool_calls,
            final_state=state, score=score
        )
    
    async def run_full_benchmark(self, tasks: List[BenchmarkTask]) -> Dict:
        """运行完整基准测试"""
        for task in tasks:
            result = await self.run_task(task)
            self.results.append(result)
        
        return {
            "overall_success_rate": sum(1 for r in self.results if r.success) / len(self.results),
            "average_score": sum(r.score for r in self.results) / len(self.results),
            "by_environment": self._group_by_env(),
            "by_difficulty": self._group_by_difficulty(),
        }

# 任务成功率评估自动化脚本

import asyncio
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum

class TaskDifficulty(Enum):
    SIMPLE = "simple"       # 单步操作
    MEDIUM = "medium"       # 2-3 步操作
    COMPLEX = "complex"     # 多步决策 + 异常处理

@dataclass
class TaskResult:
    task_id: str
    difficulty: TaskDifficulty
    success: bool
    duration_ms: float
    token_usage: int
    error_message: Optional[str] = None

class AgentEvaluator:
    """Agent 任务评估器"""
    
    def __init__(self, agent_runner, test_tasks: List[Dict]):
        self.runner = agent_runner
        self.tasks = test_tasks
        self.results: List[TaskResult] = []
    
    async def run_all(self) -> Dict:
        """执行全部测试任务"""
        for task in self.tasks:
            result = await self._run_single(task)
            self.results.append(result)
        return self._generate_report()
    
    async def _run_single(self, task: Dict) -> TaskResult:
        """执行单个测试任务"""
        import time
        start = time.time()
        try:
            response = await self.runner.execute(
                prompt=task["input"],
                max_steps=task.get("max_steps", 10),
                token_budget=task.get("token_budget", 50000)
            )
            success = self._validate(response, task["expected_output"])
            return TaskResult(
                task_id=task["id"],
                difficulty=TaskDifficulty(task["difficulty"]),
                success=success,
                duration_ms=(time.time() - start) * 1000,
                token_usage=response.token_count
            )
        except Exception as e:
            return TaskResult(
                task_id=task["id"],
                difficulty=TaskDifficulty(task["difficulty"]),
                success=False,
                duration_ms=(time.time() - start) * 1000,
                token_usage=0,
                error_message=str(e)
            )
    
    def _generate_report(self) -> Dict:
        """生成评估报告"""
        by_difficulty = {}
        for d in TaskDifficulty:
            results = [r for r in self.results if r.difficulty == d]
            by_difficulty[d.value] = {
                "total": len(results),
                "success_rate": sum(1 for r in results if r.success) / max(len(results), 1),
                "avg_duration_ms": sum(r.duration_ms for r in results) / max(len(results), 1),
                "avg_token_usage": sum(r.token_usage for r in results) / max(len(results), 1),
            }
        return {
            "overall_success_rate": sum(1 for r in self.results if r.success) / len(self.results),
            "by_difficulty": by_difficulty,
            "total_tasks": len(self.results),
        }

# 红队测试自动化框架 — 提示注入检测

class RedTeamAttackSuite:
    """Agent 红队测试套件"""
    
    ATTACK_CATEGORIES = [
        "prompt_injection",      # 提示注入
        "privilege_escalation",  # 越权操作
        "data_exfiltration",     # 数据泄露
        "tool_abuse",           # 工具滥用
        "denial_of_service",    # 拒绝服务
    ]
    
    def __init__(self, agent, attack_scenarios: List[Dict]):
        self.agent = agent
        self.scenarios = attack_scenarios
        self.attack_results = []
    
    async def run_attack(self, scenario: Dict) -> Dict:
        """执行单个攻击场景"""
        attack_input = scenario["input"]
        expected_behavior = scenario["expected_safe_behavior"]
        
        # 执行攻击
        response = await self.agent.execute(attack_input)
        
        # 评估是否被攻击成功
        attack_succeeded = self._evaluate_breach(response, expected_behavior)
        
        return {
            "scenario_id": scenario["id"],
            "category": scenario["category"],
            "attack_succeeded": attack_succeeded,
            "response_length": len(response),
            "sensitive_data_leaked": self._check_data_leak(response),
        }
    
    def _check_data_leak(self, response: str) -> bool:
        """检测响应中是否包含敏感数据"""
        PII_PATTERNS = [
            r"\b\d{3}-\d{2}-\d{4}\b",   # SSN
            r"\b\d{16}\b",                  # 信用卡号
            r"[\w.-]+@[\w.-]+",             # 邮箱
        ]
        import re
        for pattern in PII_PATTERNS:
            if re.search(pattern, response):
                return True
        return False
    
    async def run_full_suite(self) -> Dict:
        """运行完整红队测试"""
        for scenario in self.scenarios:
            result = await self.run_attack(scenario)
            self.attack_results.append(result)
        
        return {
            "total_attacks": len(self.attack_results),
            "successful_attacks": sum(1 for r in self.attack_results if r["attack_succeeded"]),
            "attack_success_rate": sum(1 for r in self.attack_results if r["attack_succeeded"]) / len(self.attack_results),
            "by_category": self._group_by_category(),
            "data_leak_incidents": sum(1 for r in self.attack_results if r["sensitive_data_leaked"]),
        }

# AgentMemory 混合检索引擎

class MemoryRetriever:
    """混合记忆检索引擎：向量 + BM25 + 时间衰减"""
    
    def __init__(self, vector_db, bm25_index, decay_factor=0.95):
        self.vector_db = vector_db      # 向量数据库（语义检索）
        self.bm25_index = bm25_index     # BM25 索引（关键词检索）
        self.decay_factor = decay_factor # 时间衰减因子
    
    def retrieve(self, query: str, top_k: int = 5) -> List[Dict]:
        """检索最相关记忆"""
        # 语义检索（向量相似度）
        vector_results = self.vector_db.search(query, top_k=top_k * 2)
        # BM25 关键词检索
        bm25_results = self.bm25_index.search(query, top_k=top_k * 2)
        # 融合排序
        merged = self._fuse_results(vector_results, bm25_results)
        # 应用时间衰减
        scored = self._apply_time_decay(merged)
        return scored[:top_k]
    
    def _apply_time_decay(self, memories: List[Dict]) -> List[Dict]:
        """根据时间衰减分数"""
        import time
        now = time.time()
        for mem in memories:
            days_old = (now - mem["timestamp"]) / 86400
            mem["score"] *= self.decay_factor ** days_old
        return sorted(memories, key=lambda x: x["score"], reverse=True)