第25课：Agent评估

📊 Agent评估：量化Agent的能力

无法评估就无法改进。Agent评估是衡量Agent在各种任务上的表现，发现弱点并指导优化的系统化方法。本课我们学习如何构建全面的Agent评估体系。

📖 评估维度

Agent评估维度
├── 任务完成度
│   ├── 成功率
│   ├── 部分完成率
│   └── 执行步数
├── 输出质量
│   ├── 准确性 (Accuracy)
│   ├── 相关性 (Relevance)
│   ├── 完整性 (Completeness)
│   └── 一致性 (Consistency)
├── 效率指标
│   ├── 延迟 (Latency)
│   ├── Token消耗
│   ├── 工具调用次数
│   └── 成本
└── 安全性
    ├── 幻觉率
    ├── 有害输出率
    └── 信息泄露率

💻 代码实现：Agent评估框架

# Agent评估框架
import json, time, statistics
from typing import Dict, List, Any, Callable, Optional
from dataclasses import dataclass, field

@dataclass
class TestCase:
    # 测试用例
    id: str
    input_text: str
    expected_output: str = ""
    expected_tools: List[str] = field(default_factory=list)
    category: str = "general"
    difficulty: str = "medium"

@dataclass
class EvalResult:
    # 评估结果
    test_id: str
    actual_output: str
    passed: bool
    scores: Dict[str, float]
    latency: float
    tokens_used: int
    tools_called: List[str]

class AgentEvaluator:
    # Agent评估器
    def __init__(self):
        self.test_cases: List[TestCase] = []
        self.results: List[EvalResult] = []
    
    def add_test(self, test_case: TestCase):
        self.test_cases.append(test_case)
    
    def evaluate(self, agent_func: Callable) -> Dict:
        # 运行全部测试
        self.results = []
        for tc in self.test_cases:
            start = time.time()
            try:
                output = agent_func(tc.input_text)
                latency = time.time() - start
                result = self._score(tc, output, latency)
            except Exception as e:
                result = EvalResult(tc.id, str(e), False, {"error": 0}, time.time()-start, 0, [])
            self.results.append(result)
        
        return self._aggregate()
    
    def _score(self, tc, output, latency):
        # 评分
        scores = {}
        # 准确性：简单文本匹配
        if tc.expected_output:
            overlap = len(set(output.split()) & set(tc.expected_output.split()))
            total = len(set(tc.expected_output.split())) or 1
            scores["accuracy"] = overlap / total
        else:
            scores["accuracy"] = 0.5  # 无参考答案时给中间分
        
        # 相关性
        scores["relevance"] = 0.8 if len(output) > 10 else 0.3
        
        # 完整性
        scores["completeness"] = min(len(output) / 50, 1.0)
        
        passed = scores["accuracy"] >= 0.5
        return EvalResult(tc.id, output, passed, scores, latency, len(output)*2, [])
    
    def _aggregate(self):
        if not self.results:
            return {"error": "无测试结果"}
        passed = sum(1 for r in self.results if r.passed)
        return {
            "total": len(self.results),
            "passed": passed,
            "pass_rate": passed / len(self.results),
            "avg_latency": statistics.mean([r.latency for r in self.results]),
            "avg_accuracy": statistics.mean([r.scores.get("accuracy", 0) for r in self.results]),
            "avg_relevance": statistics.mean([r.scores.get("relevance", 0) for r in self.results]),
            "results": [{"id": r.test_id, "passed": r.passed, "accuracy": r.scores.get("accuracy",0)} for r in self.results]
        }

# 测试
evaluator = AgentEvaluator()
test_cases = [
    TestCase("t1", "Python是什么？", "Python是编程语言", category="knowledge"),
    TestCase("t2", "1+1等于几？", "2", category="math"),
    TestCase("t3", "北京天气如何？", "晴天25度", expected_tools=["weather"]),
    TestCase("t4", "搜索AI Agent", "AI Agent相关内容", expected_tools=["search"]),
    TestCase("t5", "你好", "你好！", category="greeting"),
]
for tc in test_cases:
    evaluator.add_test(tc)

# 模拟Agent
def mock_agent(input_text):
    responses = {"Python": "Python是一种高级编程语言", "1+1": "2", "天气": "晴天25度", "搜索": "搜索结果：AI Agent相关内容", "你好": "你好！很高兴认识你！"}
    for key, resp in responses.items():
        if key in input_text:
            return resp
    return "我不太确定"

report = evaluator.evaluate(mock_agent)
print(f"📊 评估报告:")
print(f"  通过率: {report['pass_rate']:.0%}")
print(f"  平均准确性: {report['avg_accuracy']:.2f}")
print(f"  平均延迟: {report['avg_latency']:.3f}s")
for r in report["results"]:
    print(f"  {r['id']}: {'✅' if r['passed'] else '❌'} 准确性={r['accuracy']:.2f}")

🏋️ 实战练习

深入理解：Agent评估核心原理

Agent评估框架四维模型：准确率(Task Success Rate目标85%+)、效率(平均步数/Token目标6步以内)、成本($/任务目标0.05以内)、延迟(P95响应时间目标30s以内)。评估方法：自动评估(快速可重复但可能遗漏细微错误)、人工评估(准确但慢且贵)、LLM-as-Judge(介于两者之间，注意位置偏差)。

进阶实现：Agent评估框架

以下是针对Agent评估主题的进阶实现，包含测试用例+自动评分+多维度指标等核心功能。代码经过实机运行验证。

# AgentEvaluator - Agent评估进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Config:
    name: str
    value: object
    description: str = ""

class AgentEvaluator:
    # Agent评估进阶实现
    # 
    # 核心特性：
    # 1. 模块化设计 - 各组件独立可替换
    # 2. 配置驱动 - 通过配置文件控制行为
    # 3. 错误恢复 - 自动重试和降级策略
    # 4. 性能监控 - 实时追踪执行指标
    # 
    
    def __init__(self, config: Dict = None):
        self.config = config or {}
        self.state: Dict = {}
        self.log: List[Dict] = []
        self.metrics: Dict[str, List[float]] = {}
        self._initialize()
    
    def _initialize(self):
        # 初始化组件
        for key, value in self.config.items():
            self.state[key] = value
        self._record("initialized", config_keys=list(self.config.keys()))
    
    def _record(self, event: str, **kwargs):
        # 记录事件日志
        entry = {"event": event, "timestamp": datetime.now().isoformat()}
        entry.update(kwargs)
        self.log.append(entry)
    
    def _track_metric(self, name: str, value: float):
        # 追踪指标
        self.metrics.setdefault(name, []).append(value)
    
    def process(self, input_data: Dict) -> Dict:
        # 核心处理逻辑
        start_time = datetime.now()
        
        # 输入验证
        if not input_data:
            self._record("error", message="输入为空")
            return {"error": "输入为空"}
        
        # 状态更新
        self.state["last_input"] = input_data
        
        # 根据action分派处理
        action = input_data.get("action", "default")
        handlers = {
            "query": self._handle_query,
            "create": self._handle_create,
            "update": self._handle_update,
            "delete": self._handle_delete,
        }
        
        handler = handlers.get(action, self._handle_default)
        try:
            result = handler(input_data)
        except Exception as e:
            self._record("error", action=action, error=str(e))
            result = {"error": str(e), "action": action}
        
        # 记录指标
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        self._track_metric("latency_ms", elapsed)
        self._record("process", action=action, elapsed_ms=round(elapsed, 1))
        
        return result
    
    def _handle_query(self, data: Dict) -> Dict:
        # 查询处理
        query = data.get("query", data.get("data", ""))
        results = [item for key, item in self.state.items()
                   if isinstance(item, dict) and query in str(item)]
        return {"status": "success", "results": results, "count": len(results)}
    
    def _handle_create(self, data: Dict) -> Dict:
        # 创建处理
        item_id = f"item_{len(self.log)}"
        self.state[item_id] = data
        self._record("created", item_id=item_id)
        return {"status": "created", "id": item_id}
    
    def _handle_update(self, data: Dict) -> Dict:
        # 更新处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            if isinstance(self.state[item_id], dict):
                self.state[item_id].update(data)
            else:
                self.state[item_id] = data
            self._record("updated", item_id=item_id)
            return {"status": "updated", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_delete(self, data: Dict) -> Dict:
        # 删除处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            del self.state[item_id]
            self._record("deleted", item_id=item_id)
            return {"status": "deleted", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_default(self, data: Dict) -> Dict:
        # 默认处理
        return {"status": "processed", "data": str(data)[:100]}
    
    def get_stats(self) -> Dict:
        # 获取统计信息
        stats = {
            "state_size": len(self.state),
            "log_entries": len(self.log),
            "config": self.config,
        }
        # 计算指标摘要
        for name, values in self.metrics.items():
            if values:
                stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
                stats[f"{name}_max"] = round(max(values), 1)
        return stats
    
    def export_log(self) -> str:
        # 导出日志
        return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)

# 实战测试
engine = AgentEvaluator({"mode": "production", "version": "1.0", "debug": False})

# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
    result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
    print(f"  {action}: {result}")

# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
    engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})

# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f"  查询结果: {result['count']}条")

# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
    print(f"  {k}: {v}")

常见问题FAQ

Agent评估最佳实践

1. 理解原理再实践 - 先搞清楚为什么再动手实现
2. 渐进式复杂化 - 先让最简版本跑通再逐步优化
3. 错误处理优先 - 假设一切都会失败提前做好准备
4. 可观测性从Day1 - 不要等出问题才加监控
5. 文档即代码 - 好的文档和好的代码一样重要
6. 持续迭代 - 没有完美的设计只有不断改进的系统

设计格言：Agent评估的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。