Agent 评估框架：从任务完成率到自主性评级

# AgentBench 评估框架简化示例
from dataclasses import dataclass
from enum import Enum

class TaskCategory(Enum):
    REASONING = "reasoning"
    TOOL_USE = "tool_use"
    WEB_BROWSING = "web_browsing"
    CODING = "coding"
    MULTI_STEP = "multi_step"

@dataclass
class AgentTask:
    id: str
    category: TaskCategory
    prompt: str
    expected_output: str
    tools_available: list[str]
    max_steps: int
    scoring: dict  # 评分标准

@dataclass
class EvaluationResult:
    task_id: str
    success: bool
    score: float          # 0-1
    steps_taken: int
    tokens_used: int
    time_elapsed: float
    error_type: str | None

def evaluate_agent(agent, tasks: list[AgentTask]) -> list[EvaluationResult]:
    results = []
    for task in tasks:
        result = agent.run(task)
        score = score_output(result.output, task.expected_output, task.scoring)
        
        results.append(EvaluationResult(
            task_id=task.id,
            success=score > 0.8,
            score=score,
            steps_taken=result.step_count,
            tokens_used=result.total_tokens,
            time_elapsed=result.elapsed_time,
            error_type=result.error if not score > 0.8 else None
        ))
    return results

# SWE-bench 评估流程
class SWEBenchEvaluator:
    def __init__(self, repo_dir: str):
        self.repo_dir = repo_dir
        self.test_cases = self.load_test_cases()
    
    def evaluate(self, agent, issue_id: str) -> dict:
        issue = self.get_issue(issue_id)
        
        # Agent 尝试修复
        patch = agent.generate_patch(issue)
        
        # 应用补丁
        self.apply_patch(patch)
        
        # 运行测试
        test_results = self.run_tests(issue.test_files)
        
        # 恢复原始状态
        self.reset_repo()
        
        return {
            "issue_id": issue_id,
            "patch_generated": patch is not None,
            "tests_passed": test_results.passed,
            "tests_failed": test_results.failed,
            "resolve_rate": test_results.passed / (test_results.passed + test_results.failed),
        }

import asyncio
from dataclasses import dataclass
from typing import Callable

@dataclass
class EvalConfig:
    agent_factory: Callable
    task_set: str           # "agentbench" | "swebench" | "webarena"
    max_concurrent: int = 5
    timeout_per_task: int = 300
    save_traces: bool = True
    output_dir: str = "./eval_results"

class AgentEvaluationPipeline:
    def __init__(self, config: EvalConfig):
        self.config = config
        self.tasks = self.load_tasks(config.task_set)
    
    async def run(self) -> dict:
        """执行完整评估流水线"""
        semaphore = asyncio.Semaphore(self.config.max_concurrent)
        
        async def run_single(task):
            async with semaphore:
                try:
                    return await asyncio.wait_for(
                        self.evaluate_single(task),
                        timeout=self.config.timeout_per_task
                    )
                except asyncio.TimeoutError:
                    return {"task_id": task.id, "status": "timeout"}
                except Exception as e:
                    return {"task_id": task.id, "status": "error", "error": str(e)}
        
        results = await asyncio.gather(*[run_single(t) for t in self.tasks])
        
        # 生成评估报告
        report = self.generate_report(results)
        return report
    
    async def evaluate_single(self, task) -> dict:
        """评估单个任务"""
        agent = self.config.agent_factory()
        
        # 记录执行轨迹
        trace = []
        start_time = time.time()
        
        try:
            result = await agent.run(task.prompt, tools=task.tools)
            
            # 评分
            score = self.score(result, task)
            
            return {
                "task_id": task.id,
                "status": "completed",
                "score": score,
                "steps": len(trace),
                "time": time.time() - start_time,
                "trace": trace if self.config.save_traces else None,
            }
        except Exception as e:
            return {
                "task_id": task.id,
                "status": "failed",
                "error": str(e),
                "time": time.time() - start_time,
            }
    
    def generate_report(self, results: list) -> dict:
        """生成评估报告"""
        total = len(results)
        completed = [r for r in results if r["status"] == "completed"]
        success = [r for r in completed if r["score"] > 0.8]
        
        return {
            "summary": {
                "total_tasks": total,
                "completed": len(completed),
                "successful": len(success),
                "success_rate": len(success) / total,
                "avg_score": sum(r["score"] for r in completed) / len(completed) if completed else 0,
                "avg_time": sum(r["time"] for r in completed) / len(completed) if completed else 0,
            },
            "by_category": self.breakdown_by_category(results),
            "error_analysis": self.analyze_errors(results),
        }

# 使用 LLM 作为评估者
from openai import OpenAI

judge_client = OpenAI()

def llm_judge_evaluate(
    task: str,
    agent_response: str,
    reference: str | None = None,
    criteria: list[str] = None
) -> dict:
    """使用 LLM 评估 Agent 输出质量"""
    
    criteria = criteria or [
        "任务完成度：是否完成了用户要求的任务",
        "准确性：输出内容是否准确无误",
        "效率：是否以合理的方式完成任务",
        "安全性：是否存在安全风险",
        "可读性：输出是否清晰易懂",
    ]
    
    prompt = f"""
    你是一个专业的 AI Agent 评估者。请评估以下 Agent 的表现。
    
    任务: {task}
    Agent 输出: {agent_response}
    参考答案: {reference or "无"}
    
    评估维度:
    {chr(10).join(f'{i+1}. {c}' for i, c in enumerate(criteria))}
    
    对每个维度给出 1-5 分的评分和理由，最后给出总分。
    输出 JSON 格式。
    """
    
    response = judge_client.chat.completions.create(
        model="gpt-5",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
    )
    
    return response.choices[0].message.content

from dataclasses import dataclass, field
from collections import defaultdict
import time

@dataclass
class ProductionMetrics:
    """生产环境 Agent 监控指标"""
    # 任务指标
    total_tasks: int = 0
    successful_tasks: int = 0
    failed_tasks: int = 0
    avg_completion_time: float = 0
    
    # 质量指标
    user_satisfaction: float = 0  # 1-5
    human_handoff_rate: float = 0  # 转人工率
    retry_rate: float = 0
    complaint_rate: float = 0
    
    # 资源指标
    avg_tokens_per_task: int = 0
    avg_cost_per_task: float = 0
    avg_steps_per_task: int = 0
    
    # 安全指标
    safety_violations: int = 0
    unauthorized_actions: int = 0
    
    def success_rate(self) -> float:
        return self.successful_tasks / self.total_tasks if self.total_tasks > 0 else 0
    
    def health_score(self) -> float:
        """综合健康分数 (0-100)"""
        weights = {
            "success_rate": 0.30,
            "satisfaction": 0.25,
            "efficiency": 0.20,
            "safety": 0.15,
            "cost": 0.10,
        }
        score = (
            self.success_rate() * 100 * weights["success_rate"] +
            self.user_satisfaction / 5 * 100 * weights["satisfaction"] +
            (1 - self.retry_rate) * 100 * weights["efficiency"] +
            (1 - self.safety_violations / max(self.total_tasks, 1)) * 100 * weights["safety"] +
            max(0, 100 - self.avg_cost_per_task * 10) * weights["cost"]
        )
        return round(score, 1)

# 生产环境告警配置
alert_rules = {
    "success_rate_drop": {
        "condition": "success_rate < 0.85",
        "window": "5min",
        "severity": "critical",
        "action": "notify_oncall",
    },
    "high_cost": {
        "condition": "avg_cost_per_task > $2.00",
        "window": "1hour",
        "severity": "warning",
        "action": "notify_team",
    },
    "safety_violation": {
        "condition": "safety_violations > 0",
        "window": "realtime",
        "severity": "critical",
        "action": "block_agent_and_notify",
    },
    "high_handoff": {
        "condition": "human_handoff_rate > 0.15",
        "window": "1hour",
        "severity": "warning",
        "action": "notify_team",
    },
}