AI Agent性能评估体系：任务成功率、轨迹效率与系统工程指标

在 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(maxlenwindow_size) self.error_counts deque(maxlenwindow_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(interval1) 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(fCPU使用率过高: {metrics[cpu_percent]}%) if metrics[memory_percent] 85: alerts.append(f内存使用率过高: {metrics[memory_percent]}%) if metrics[p95_response_time] 5.0: # 5秒 alerts.append(fP95响应时间过长: {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, indent2)}) 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, jsonalert_data, timeout5) 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_workersself.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_exceptionsTrue) 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], n20)[18] # 95% results[p99_response_time] statistics.quantiles(results[response_times], n100)[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, keylambda 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