在 AI 应用开发中，Agent 已经从一个前沿概念变成了工程实践的核心组件。无论是处理复杂任务的自主决策系统，还是与用户进行多轮对话的智能助手，Agent 的性能表现直接决定了产品的可用性和可靠性。但很多团队在评估 Agent 时，往往只关注任务是否完成这个最终结果，忽略了执行过程中的效率、稳定性和资源消耗，导致线上环境出现各种预料之外的问题。
一个真正可靠的 Agent 评估体系需要同时关注三个层面：任务级别的成功率（Task-level Success）、执行轨迹的质量（Trajectory Evaluation），以及系统工程指标（System Engineering Metrics）。这三个维度分别对应了业务目标达成度、执行过程优化空间和系统稳定性保障，缺少任何一个都会让评估结果失真。
本文将从工程实践角度，详细拆解如何构建完整的 Agent 性能评估体系。无论你是正在开发第一个 Agent 项目，还是需要优化现有系统的评估流程，都能找到可落地的方案和排查思路。
1. 理解 Agent 评估的三个核心维度
在深入具体指标之前，需要先明确每个评估维度的定位和相互关系。这三个维度不是孤立的，而是构成了一个从微观到宏观的完整观察链。
1.1 Task-level Success：业务目标的达成度
Task-level Success 衡量的是 Agent 是否完成了用户指定的任务目标。这是最直观的评估维度，也是业务方最关心的指标。但“成功”的定义需要根据具体场景来明确，不能简单理解为“程序没有报错”。
在实际项目中，Task-level Success 通常通过以下方式定义和测量：
- 二进制判断：任务要么完全成功，要么完全失败。适用于目标明确、结果可量化的场景，如“查询今日天气”、“生成一份报告”。
- 部分成功度评分：对于复杂任务，可以按完成质量打分。例如文档总结任务，可以按信息完整度、语言流畅度、格式规范性等维度分别评分。
- 人工评估校准：在自动化评估不够可靠时，需要引入人工判断作为黄金标准。
关键是要在项目初期就明确成功的标准，避免后期因为定义模糊导致评估结果争议。
1.2 轨迹评估：执行过程的效率与质量
轨迹评估关注的是 Agent 从开始到结束的整个执行路径。即使最终任务成功了，如果执行过程绕了远路、犯了不必要的错误，或者资源消耗过大，这样的 Agent 在实际应用中也会带来体验和成本问题。
轨迹评估的核心价值在于：
- 识别优化机会：发现 Agent 决策中的低效模式，如重复尝试、不必要的工具调用等。
- 评估稳定性：观察在不同输入条件下，Agent 的行为是否一致可靠。
- 调试辅助：当任务失败时，轨迹记录提供了完整的排查线索。
常见的轨迹评估指标包括步骤数量、工具调用次数、回溯次数、思考时间分布等。这些指标需要结合具体任务类型来解读，没有绝对的好坏标准。
1.3 系统工程指标：系统层面的可靠性
系统工程指标关注的是 Agent 作为系统组件的运行状况，包括性能、资源消耗、可用性等。这些指标决定了 Agent 能否在生产环境中稳定服务。
对于需要 7x24 小时运行的 Agent 系统，系统工程指标的重要性甚至超过前两个维度，因为系统层面的故障会导致所有任务都无法执行。典型的系统工程指标包括：
- 响应时间：从接收请求到返回结果的延迟分布。
- 吞吐量：单位时间内能处理的任务数量。
- 资源使用率：CPU、内存、网络、API 调用次数等消耗情况。
- 错误率：系统级错误（如超时、内存溢出）的发生频率。
- 可用性：系统正常服务的时间比例。
2. 构建 Task-level Success 评估体系
Task-level Success 评估的关键在于设计可靠的验证机制。根据任务类型的不同，验证方式也需要相应调整。
2.1 明确成功标准与验证方法
在开始评估前，必须明确每个任务类型的成功标准。以下表格展示了不同场景下的成功定义和验证方法：
| 任务类型 | 成功标准 | 验证方法 | 注意事项 |
|---|---|---|---|
| 信息查询 | 返回信息准确完整 | 与权威数据源对比 | 注意信息时效性和来源可靠性 |
| 内容生成 | 符合格式要求和内容质量 | 自动化检查+人工抽样 | 质量评估主观性强，需要明确评分标准 |
| 数据分析 | 分析结论正确，图表规范 | 对比预期输出 | 考虑数据边界情况和异常处理 |
| 工具调用 | 正确执行操作并返回结果 | 检查操作结果状态 | 注意权限问题和环境差异 |
对于自动化验证，常用的技术方案包括：
- 规则匹配：使用正则表达式、关键字匹配等检查输出是否符合预期格式。
- 相似度计算：使用文本相似度算法（如 BLEU、ROUGE）对比生成内容与参考答案。
- 代码执行：对于生成代码的任务，通过实际执行验证功能正确性。
- API 调用验证：对于调用外部工具的任务，检查工具返回的结果状态。
2.2 实现自动化评估流水线
手动评估无法满足大规模测试的需求，需要建立自动化的评估流水线。以下是一个基于 Python 的简单评估框架示例：
import json from typing import Dict, Any, List class TaskEvaluator: def __init__(self, evaluation_rules: Dict[str, Any]): self.rules = evaluation_rules def evaluate_success(self, task_input: str, agent_output: str, expected_output: str = None) -> Dict[str, Any]: """评估单个任务的完成情况""" results = { 'task_input': task_input, 'agent_output': agent_output, 'success': False, 'score': 0.0, 'details': {} } # 根据任务类型应用不同的评估规则 task_type = self._classify_task(task_input) if task_type == 'information_query': results.update(self._evaluate_query(task_input, agent_output, expected_output)) elif task_type == 'content_generation': results.update(self._evaluate_generation(task_input, agent_output)) # 其他任务类型的评估逻辑... return results def _evaluate_query(self, task_input: str, agent_output: str, expected: str) -> Dict[str, Any]: """评估信息查询类任务""" # 实现具体的验证逻辑 success = self._check_accuracy(agent_output, expected) score = self._calculate_similarity(agent_output, expected) return { 'success': success, 'score': score, 'details': {'similarity': score, 'expected': expected} } def batch_evaluate(self, test_cases: List[Dict]) -> Dict[str, Any]: """批量评估测试用例""" total_success = 0 total_score = 0.0 detailed_results = [] for case in test_cases: result = self.evaluate_success( case['input'], case['output'], case.get('expected') ) detailed_results.append(result) if result['success']: total_success += 1 total_score += result['score'] success_rate = total_success / len(test_cases) if test_cases else 0 avg_score = total_score / len(test_cases) if test_cases else 0 return { 'success_rate': success_rate, 'average_score': avg_score, 'total_cases': len(test_cases), 'detailed_results': detailed_results } # 使用示例 evaluator = TaskEvaluator({ 'query_threshold': 0.8, # 相似度阈值 'generation_criteria': ['completeness', 'accuracy', 'fluency'] }) test_cases = [ {'input': '今天北京天气如何？', 'output': '北京今天晴，气温15-25度', 'expected': '晴，15-25度'}, # 更多测试用例... ] results = evaluator.batch_evaluate(test_cases) print(f"任务成功率: {results['success_rate']:.2%}")2.3 建立测试用例库与基准线
可靠的评估需要覆盖各种场景的测试用例库。测试用例应该包括：
- 正常场景：典型的使用情况，确保基本功能正常。
- 边界场景：输入边界值、特殊字符、极端情况等。
- 错误场景：无效输入、权限不足、资源不可用等情况。
- 复杂场景：需要多步推理或工具调用的复杂任务。
建立基准线（Baseline）也很重要，可以用简单的规则系统或早期版本作为对比基准，衡量新版本的改进程度。
3. 深入轨迹评估：从执行路径发现优化机会
轨迹评估需要记录和分析 Agent 的完整执行过程，这比单纯看最终结果能提供更多优化线索。
3.1 轨迹数据采集与存储
首先需要建立轨迹记录机制，捕获关键的执行信息：
import time from datetime import datetime from typing import List, Dict, Any import json class ExecutionTracker: def __init__(self): self.current_trajectory = [] self.start_time = None def start_tracking(self, task_id: str, task_input: str): """开始跟踪一个新的任务执行""" self.current_trajectory = [] self.start_time = time.time() self.record_event('task_start', { 'task_id': task_id, 'input': task_input, 'timestamp': datetime.now().isoformat() }) def record_event(self, event_type: str, details: Dict[str, Any]): """记录执行事件""" event = { 'type': event_type, 'timestamp': time.time() - self.start_time, 'details': details } self.current_trajectory.append(event) def record_llm_call(self, prompt: str, response: str, tokens_used: int): """记录LLM调用""" self.record_event('llm_call', { 'prompt_preview': prompt[:200], # 记录前200字符作为预览 'response_preview': response[:200], 'tokens_used': tokens_used }) def record_tool_call(self, tool_name: str, parameters: Dict, result: Any, success: bool): """记录工具调用""" self.record_event('tool_call', { 'tool': tool_name, 'parameters': parameters, 'success': success, 'result_preview': str(result)[:200] if result else None }) def record_error(self, error_type: str, error_message: str, recovery_action: str = None): """记录错误信息""" self.record_event('error', { 'error_type': error_type, 'message': error_message, 'recovery_action': recovery_action }) def get_trajectory_summary(self) -> Dict[str, Any]: """生成轨迹摘要""" if not self.current_trajectory: return {} llm_calls = [e for e in self.current_trajectory if e['type'] == 'llm_call'] tool_calls = [e for e in self.current_trajectory if e['type'] == 'tool_call'] errors = [e for e in self.current_trajectory if e['type'] == 'error'] total_time = self.current_trajectory[-1]['timestamp'] if self.current_trajectory else 0 return { 'total_steps': len(self.current_trajectory), 'llm_calls': len(llm_calls), 'tool_calls': len(tool_calls), 'errors': len(errors), 'total_time': total_time, 'events': self.current_trajectory } # 在Agent执行过程中集成轨迹跟踪 tracker = ExecutionTracker() def execute_agent_task(task_input: str): tracker.start_tracking('task_001', task_input) try: # Agent执行逻辑 # 在每个关键步骤调用tracker记录事件 tracker.record_llm_call("思考如何解决问题...", "首先需要查询天气信息", 150) tracker.record_tool_call('weather_api', {'city': '北京'}, {'temp': 20}, True) # ... 更多执行步骤 except Exception as e: tracker.record_error('execution_error', str(e)) raise return tracker.get_trajectory_summary()3.2 轨迹质量指标分析
收集到轨迹数据后，需要从多个维度分析执行质量：
| 指标类别 | 具体指标 | 分析目的 | 优化方向 |
|---|---|---|---|
| 效率指标 | 总步骤数、LLM调用次数、工具调用次数 | 识别执行路径是否简洁 | 减少不必要的步骤，合并相似操作 |
| 时间指标 | 各步骤耗时、总执行时间、思考时间占比 | 发现性能瓶颈 | 优化慢速步骤，设置超时机制 |
| 质量指标 | 错误次数、回退次数、重复操作次数 | 评估执行稳定性 | 加强错误处理，改进决策逻辑 |
| 资源指标 | Token消耗量、API调用成本 | 控制运营成本 | 优化提示词，缓存重复结果 |
以下代码展示了如何分析轨迹数据：
class TrajectoryAnalyzer: def __init__(self, trajectory_data: List[Dict]): self.data = trajectory_data def calculate_efficiency_metrics(self) -> Dict[str, Any]: """计算效率相关指标""" total_steps = len(self.data) llm_calls = len([e for e in self.data if e['type'] == 'llm_call']) tool_calls = len([e for e in self.data if e['type'] == 'tool_call']) # 计算步骤密度（有效步骤占比） effective_steps = llm_calls + tool_calls step_density = effective_steps / total_steps if total_steps > 0 else 0 return { 'total_steps': total_steps, 'llm_calls': llm_calls, 'tool_calls': tool_calls, 'step_density': step_density, 'calls_per_step': (llm_calls + tool_calls) / total_steps if total_steps > 0 else 0 } def identify_inefficient_patterns(self) -> List[Dict]: """识别低效执行模式""" patterns = [] # 检查重复的工具调用 tool_calls = [e for e in self.data if e['type'] == 'tool_call'] tool_usage = {} for call in tool_calls: tool_name = call['details']['tool'] params = str(call['details']['parameters']) key = f"{tool_name}_{params}" tool_usage[key] = tool_usage.get(key, 0) + 1 repeated_calls = {k: v for k, v in tool_usage.items() if v > 1} if repeated_calls: patterns.append({ 'pattern': 'repeated_tool_calls', 'description': '相同参数的工具被多次调用', 'details': repeated_calls, 'suggestion': '考虑缓存工具调用结果或优化决策逻辑' }) # 检查长时间的思考步骤 thinking_events = [e for e in self.data if e['type'] == 'llm_call'] long_thinking = [e for e in thinking_events if e['details'].get('tokens_used', 0) > 1000] if long_thinking: patterns.append({ 'pattern': 'excessive_thinking', 'description': '存在消耗大量token的思考步骤', 'details': {'count': len(long_thinking)}, 'suggestion': '优化提示词设计，减少不必要的推理步骤' }) return patterns def generate_optimization_report(self) -> Dict[str, Any]: """生成完整的优化报告""" efficiency = self.calculate_efficiency_metrics() patterns = self.identify_inefficient_patterns() # 计算总体评分（简化版） base_score = 100 penalty = len(patterns) * 10 # 每个低效模式扣10分 final_score = max(0, base_score - penalty) return { 'efficiency_metrics': efficiency, 'inefficient_patterns': patterns, 'optimization_score': final_score, 'recommendations': [p['suggestion'] for p in patterns] }3.3 轨迹对比与模式分析
通过对比不同版本或不同配置下 Agent 的执行轨迹，可以发现改进效果和新的问题模式：
def compare_trajectories(baseline_traj, improved_traj): """对比两个版本的执行轨迹""" baseline_analyzer = TrajectoryAnalyzer(baseline_traj) improved_analyzer = TrajectoryAnalyzer(improved_traj) baseline_metrics = baseline_analyzer.calculate_efficiency_metrics() improved_metrics = improved_analyzer.calculate_efficiency_metrics() comparison = {} for key in baseline_metrics.keys(): if isinstance(baseline_metrics[key], (int, float)): improvement = improved_metrics[key] - baseline_metrics[key] percent_change = (improvement / baseline_metrics[key]) * 100 if baseline_metrics[key] != 0 else 0 comparison[key] = { 'baseline': baseline_metrics[key], 'improved': improved_metrics[key], 'change': improvement, 'percent_change': percent_change } return comparison4. 系统工程指标：保障生产环境稳定性
系统工程指标关注的是 Agent 作为服务组件的运行状况，这些指标直接影响到系统的可用性和可维护性。
4.1 关键系统工程指标定义与监控
在生产环境中，需要监控以下核心指标：
| 指标类别 | 具体指标 | 监控频率 | 告警阈值 | 应对措施 |
|---|---|---|---|---|
| 性能指标 | P95/P99延迟、QPS | 实时 | P99 > 5s 或 QPS下降50% | 扩容、优化代码 |
| 资源指标 | CPU使用率、内存占用 | 每分钟 | CPU > 80% 持续5分钟 | 检查内存泄漏、优化算法 |
| 可用性指标 | 错误率、超时率 | 每5分钟 | 错误率 > 1% 或超时率 > 5% | 重启服务、回滚版本 |
| 业务指标 | 任务成功率、用户满意度 | 每小时 | 成功率下降10% | 检查模型退化、数据变化 |
实现基础监控的代码示例：
import time import psutil import logging from datetime import datetime from collections import deque from typing import Dict, List class SystemMetricsCollector: def __init__(self, window_size: int = 100): self.window_size = window_size self.response_times = deque(maxlen=window_size) self.error_counts = deque(maxlen=window_size) self.start_time = time.time() def record_request(self, response_time: float, success: bool): """记录请求指标""" self.response_times.append(response_time) if not success: self.error_counts.append(1) else: self.error_counts.append(0) def get_current_metrics(self) -> Dict[str, float]: """获取当前系统指标""" # 系统资源使用情况 cpu_percent = psutil.cpu_percent(interval=1) memory_info = psutil.virtual_memory() # 应用性能指标 if self.response_times: avg_response_time = sum(self.response_times) / len(self.response_times) p95_response_time = sorted(self.response_times)[int(len(self.response_times) * 0.95)] error_rate = sum(self.error_counts) / len(self.error_counts) if self.error_counts else 0 else: avg_response_time = p95_response_time = error_rate = 0 # 计算QPS（最近一分钟） current_time = time.time() recent_requests = [rt for rt in self.response_times if current_time - rt < 60] # 最近60秒的请求 qps = len(recent_requests) / 60 if recent_requests else 0 return { 'timestamp': datetime.now().isoformat(), 'cpu_percent': cpu_percent, 'memory_percent': memory_info.percent, 'avg_response_time': avg_response_time, 'p95_response_time': p95_response_time, 'error_rate': error_rate, 'qps': qps, 'uptime': current_time - self.start_time } def check_alert_conditions(self, metrics: Dict) -> List[str]: """检查是否需要触发告警""" alerts = [] if metrics['cpu_percent'] > 80: alerts.append(f"CPU使用率过高: {metrics['cpu_percent']}%") if metrics['memory_percent'] > 85: alerts.append(f"内存使用率过高: {metrics['memory_percent']}%") if metrics['p95_response_time'] > 5.0: # 5秒 alerts.append(f"P95响应时间过长: {metrics['p95_response_time']}s") if metrics['error_rate'] > 0.01: # 1% alerts.append(f"错误率过高: {metrics['error_rate']:.2%}") return alerts # 使用示例 collector = SystemMetricsCollector() # 在每次请求处理完成后记录指标 def handle_agent_request(request_data): start_time = time.time() success = False try: # 处理请求的逻辑 result = process_agent_task(request_data) success = True return result except Exception as e: logging.error(f"请求处理失败: {e}") raise finally: response_time = time.time() - start_time collector.record_request(response_time, success) # 定期检查系统状态 metrics = collector.get_current_metrics() alerts = collector.check_alert_conditions(metrics) if alerts: logging.warning(f"系统告警: {alerts}") # 发送告警通知...4.2 建立监控仪表盘与告警机制
单纯的指标收集不够，需要建立可视化的监控仪表盘和及时的告警机制：
import json from typing import Dict, List import requests class MonitoringDashboard: def __init__(self, dashboard_url: str, alert_webhook: str = None): self.dashboard_url = dashboard_url self.alert_webhook = alert_webhook self.metric_history = [] def push_metrics(self, metrics: Dict): """推送指标到监控系统""" self.metric_history.append(metrics) # 保留最近1000个数据点 if len(self.metric_history) > 1000: self.metric_history = self.metric_history[-1000:] # 实际项目中这里会推送到Prometheus、Datadog等监控系统 # 示例中简单打印到日志 logging.info(f"指标更新: {json.dumps(metrics, indent=2)}") def send_alert(self, alert_type: str, message: str, severity: str = 'warning'): """发送告警通知""" alert_data = { 'type': alert_type, 'message': message, 'severity': severity, 'timestamp': datetime.now().isoformat() } if self.alert_webhook: try: requests.post(self.alert_webhook, json=alert_data, timeout=5) except Exception as e: logging.error(f"告警发送失败: {e}") # 同时记录到日志 logging.warning(f"告警: {severity.upper()} - {alert_type}: {message}") def generate_daily_report(self) -> Dict: """生成每日报告""" if not self.metric_history: return {} today_metrics = [m for m in self.metric_history if datetime.fromisoformat(m['timestamp']).date() == datetime.today().date()] if not today_metrics: return {} # 计算各项指标的统计信息 response_times = [m['avg_response_time'] for m in today_metrics] error_rates = [m['error_rate'] for m in today_metrics] cpu_usage = [m['cpu_percent'] for m in today_metrics] report = { 'date': datetime.today().date().isoformat(), 'total_requests': len(today_metrics), 'avg_response_time': sum(response_times) / len(response_times), 'max_response_time': max(response_times), 'avg_error_rate': sum(error_rates) / len(error_rates), 'max_cpu_usage': max(cpu_usage), 'availability': (1 - sum(error_rates) / len(error_rates)) * 100 } return report4.3 容量规划与性能测试
在生产部署前，需要进行充分的性能测试和容量规划：
import asyncio from concurrent.futures import ThreadPoolExecutor import statistics class LoadTester: def __init__(self, agent_endpoint: str, max_workers: int = 10): self.endpoint = agent_endpoint self.max_workers = max_workers async def test_concurrent_load(self, num_requests: int, request_data: List[Dict]) -> Dict: """测试并发负载下的性能表现""" results = { 'total_requests': num_requests, 'successful_requests': 0, 'failed_requests': 0, 'response_times': [], 'throughput': 0 } start_time = time.time() # 使用线程池模拟并发请求 with ThreadPoolExecutor(max_workers=self.max_workers) as executor: loop = asyncio.get_event_loop() futures = [ loop.run_in_executor(executor, self._send_single_request, data) for data in request_data[:num_requests] ] responses = await asyncio.gather(*futures, return_exceptions=True) end_time = time.time() total_time = end_time - start_time # 分析结果 for response in responses: if isinstance(response, Exception): results['failed_requests'] += 1 else: results['successful_requests'] += 1 if 'response_time' in response: results['response_times'].append(response['response_time']) results['throughput'] = results['successful_requests'] / total_time if total_time > 0 else 0 # 计算统计信息 if results['response_times']: results['avg_response_time'] = statistics.mean(results['response_times']) results['p95_response_time'] = statistics.quantiles(results['response_times'], n=20)[18] # 95% results['p99_response_time'] = statistics.quantiles(results['response_times'], n=100)[98] # 99% else: results['avg_response_time'] = results['p95_response_time'] = results['p99_response_time'] = 0 results['success_rate'] = results['successful_requests'] / num_requests if num_requests > 0 else 0 return results def _send_single_request(self, request_data: Dict) -> Dict: """发送单个请求（模拟实际调用）""" # 这里替换为实际的API调用逻辑 start_time = time.time() try: # 模拟处理时间（实际项目中调用真实接口） processing_time = 0.1 + (0.2 * random.random()) # 100-300ms time.sleep(processing_time) # 模拟随机失败（测试错误处理） if random.random() < 0.02: # 2%的失败率 raise Exception("模拟API调用失败") return { 'success': True, 'response_time': time.time() - start_time, 'data': {'result': '模拟响应数据'} } except Exception as e: return { 'success': False, 'error': str(e), 'response_time': time.time() - start_time } def determine_optimal_capacity(self, target_qps: int, max_latency: float = 2.0) -> Dict: """确定最优的并发配置""" test_cases = [ {'concurrency': 1, 'duration': 30}, {'concurrency': 5, 'duration': 30}, {'concurrency': 10, 'duration': 30}, {'concurrency': 20, 'duration': 30}, {'concurrency': 50, 'duration': 30}, ] capacity_results = [] for case in test_cases: # 模拟不同并发级别的测试 result = self._simulate_load_test(case['concurrency'], case['duration']) capacity_results.append({ 'concurrency': case['concurrency'], 'achieved_qps': result['qps'], 'avg_latency': result['avg_latency'], 'error_rate': result['error_rate'] }) # 找出满足目标QPS且延迟低于阈值的配置 viable_configs = [ config for config in capacity_results if config['achieved_qps'] >= target_qps * 0.8 # 达到目标80%以上 and config['avg_latency'] <= max_latency and config['error_rate'] <= 0.01 # 错误率低于1% ] if viable_configs: # 选择并发数最小的可行配置 optimal = min(viable_configs, key=lambda x: x['concurrency']) return { 'optimal_concurrency': optimal['concurrency'], 'expected_qps': optimal['achieved_qps'], 'expected_latency': optimal['avg_latency'], 'all_test_results': capacity_results } else: return { 'optimal_concurrency': None, 'message': '没有找到满足要求的配置，需要优化性能或调整目标', 'all_test_results': capacity_results }5. 综合评估与持续改进
将三个维度的评估结果整合起来，形成完整的 Agent 性能视图，并建立持续改进机制。
5.1 建立综合评分体系
为每个维度分配权重，计算综合评分：
class ComprehensiveEvaluator: def __init__(self, task_success_weight: float = 0.5, trajectory_weight: float = 0.3, system_weight: float = 0.2): self.weights = { 'task_success': task_success_weight, 'trajectory': trajectory_weight, 'system': system_weight } def calculate_comprehensive_score(self, task_success_metrics: Dict, trajectory_metrics: Dict, system_metrics: Dict) -> Dict[str, Any]: """计算综合评分""" # 任务成功率评分（0-100分） success_rate = task_success_metrics.get('success_rate', 0) task_score = success_rate * 100 # 轨迹效率评分（基于优化报告） trajectory_score = trajectory_metrics.get('optimization_score', 0) # 系统稳定性评分（基于错误率和响应时间） error_rate = system_metrics.get('error_rate', 0) avg_latency = system_metrics.get('avg_response_time', 0) # 错误率评分（错误率越低分数越高） error_score = max(0, 100 - (error_rate * 10000)) # 错误率1%得90分 # 延迟评分（延迟越低分数越高） latency_score = max(0, 100 - (avg_latency * 10)) # 100ms得99分，1s得90分 system_score = (error_score + latency_score) / 2 # 加权综合评分 comprehensive_score = ( task_score * self.weights['task_success'] + trajectory_score * self.weights['trajectory'] + system_score * self.weights['system'] ) return { 'comprehensive_score': comprehensive_score, 'component_scores': { 'task_success': task_score, 'trajectory_efficiency': trajectory_score, 'system_stability': system_score }, 'weights': self.weights, 'recommendations': self._generate_recommendations( task_score, trajectory_score, system_score ) } def _generate_recommendations(self, task_score: float, trajectory_score: float, system_score: float) -> List[str]: """根据各维度评分生成改进建议""" recommendations = [] if task_score < 80: recommendations.append("重点优化任务成功率：增加测试用例覆盖，改进提示词设计") if trajectory_score < 70: recommendations.append("优化执行效率：减少不必要的步骤，缓存重复结果") if system_score < 90: recommendations.append("提升系统稳定性：加强错误处理，优化资源使用") # 根据相对强弱给出优先级建议 scores = [task_score, trajectory_score, system_score] min_score_index = scores.index(min(scores)) priority_areas = ['任务成功率', '执行效率', '系统稳定性'] recommendations.append(f"优先改进：{priority_areas[min_score_index]}") return recommendations5.2 建立持续评估流水线
将评估流程自动化，集成到CI/CD pipeline中：
class ContinuousEvaluationPipeline: def __init__(self, test_suite: Dict, performance_thresholds: Dict): self.test_suite = test_suite self.thresholds = performance_thresholds def run_full_evaluation(self, agent_version: str) -> Dict[str, Any]: """运行完整的评估流程""" results = { 'version': agent_version, 'timestamp': datetime.now().isoformat(), 'passed': True, 'details': {} } # 1. 任务成功率评估 task_results = self._evaluate_task_success() results['details']['task_success'] = task_results if task_results['success_rate'] < self.thresholds['min_success_rate']: results['passed'] = False results['failure_reason'] = f"任务成功率低于阈值: {task_results['success_rate']:.2%}" # 2. 轨迹效率评估 trajectory_results = self._evaluate_trajectory_efficiency() results['details']['trajectory_efficiency'] = trajectory_results if trajectory_results['optimization_score'] < self.thresholds['min_efficiency_score']: results['passed'] = False results['failure_reason'] = f"执行效率评分低于阈值: {trajectory_results['optimization_score']}" # 3. 系统性能评估 system_results = self._evaluate_system_performance() results['details']['system_performance'] = system_results if system_results['error_rate'] > self.thresholds['max_error_rate']: results['passed'] = False results['failure_reason'] = f"系统错误率超过阈值: {system_results['error_rate']:.2%}" # 4. 综合评分 evaluator = ComprehensiveEvaluator() comprehensive_score = evaluator.calculate_comprehensive_score( task_results, trajectory_results, system_results ) results['comprehensive_score'] = comprehensive_score return常见输入问题与缓冲区清理方案)
