4. 评估任务完成率
任务目标
构建一个可复用的Agent任务完成率评估框架：定义标准测试用例集，运行Agent后统计完成率，并对失败用例进行原因归类分析。本案例聚焦于评估流程的闭环实现，不涉及具体Agent内部逻辑优化。
实现方案
采用用例驱动评估模式：每个测试用例包含任务描述、预期结果和判定函数。评估器依次执行用例，记录通过/失败状态，最后输出完成率及失败原因分布。失败原因通过异常类型自动归类，同时支持人工标注补充。
核心代码
import json
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional
from enum import Enum
from collections import Counter
# ---------- 1. 测试用例定义 ----------
class FailureReason(str, Enum):
TIMEOUT = "timeout"
WRONG_OUTPUT = "wrong_output"
EXCEPTION = "exception"
PARTIAL = "partial_completion"
@dataclass
class TestCase:
task_id: str
description: str
input_data: dict
expected_output: object
timeout_sec: float = 10.0
validator: Optional[Callable[[object, object], bool]] = None
def validate(self, actual: object) -> bool:
"""默认使用精确匹配，可传入自定义validator"""
if self.validator:
return self.validator(actual, self.expected_output)
return actual == self.expected_output
# ---------- 2. Agent接口（模拟） ----------
class AgentRunner:
"""实际使用时，将run方法替换为真实Agent调用"""
def run(self, task: dict) -> object:
# 模拟不同失败模式
if task.get("type") == "timeout":
raise TimeoutError("simulated timeout")
if task.get("type") == "exception":
raise ValueError("simulated crash")
if task.get("type") == "wrong":
return "wrong_result"
if task.get("type") == "partial":
return {"completed": False, "progress": 0.5}
return task.get("expected", None)
# ---------- 3. 评估器 ----------
@dataclass
class EvalResult:
task_id: str
passed: bool
actual_output: object = None
failure_reason: Optional[FailureReason] = None
error_msg: str = ""
class CompletionRateEvaluator:
def __init__(self, agent: AgentRunner, cases: List[TestCase]):
self.agent = agent
self.cases = cases
self.results: List[EvalResult] = []
def run_evaluation(self) -> Dict:
self.results.clear()
for case in self.cases:
result = self._execute_single(case)
self.results.append(result)
return self._aggregate()
def _execute_single(self, case: TestCase) -> EvalResult:
try:
actual = self.agent.run(case.input_data)
passed = case.validate(actual)
if passed:
return EvalResult(case.task_id, True, actual)
return EvalResult(case.task_id, False, actual,
FailureReason.WRONG_OUTPUT)
except TimeoutError as e:
return EvalResult(case.task_id, False, None,
FailureReason.TIMEOUT, str(e))
except Exception as e:
return EvalResult(case.task_id, False, None,
FailureReason.EXCEPTION, str(e))
def _aggregate(self) -> Dict:
total = len(self.results)
passed = sum(1 for r in self.results if r.passed)
failed = total - passed
reason_counter = Counter(
r.failure_reason.value for r in self.results if not r.passed
)
return {
"total": total,
"passed": passed,
"failed": failed,
"completion_rate": round(passed / total, 4) if total else 0.0,
"failure_reasons": dict(reason_counter),
"details": [
{"task_id": r.task_id, "passed": r.passed,
"reason": r.failure_reason.value if r.failure_reason else None,
"error": r.error_msg}
for r in self.results
]
}
# ---------- 4. 运行示例 ----------
if __name__ == "__main__":
# 构造标准测试集（实际场景中从JSON/YAML加载）
test_cases = [
TestCase("T1", "正常任务", {"type": "normal", "expected": 42}, 42),
TestCase("T2", "超时任务", {"type": "timeout"}, None, timeout_sec=1),
TestCase("T3", "异常任务", {"type": "exception"}, None),
TestCase("T4", "错误输出", {"type": "wrong"}, "expected_result"),
TestCase("T5", "部分完成", {"type": "partial"},
{"completed": True}, validator=lambda a, e: a.get("completed", False)),
]
agent = AgentRunner()
evaluator = CompletionRateEvaluator(agent, test_cases)
report = evaluator.run_evaluation()
print(json.dumps(report, indent=2, ensure_ascii=False))运行与验证
- 运行方式：将上述代码保存为
eval_completion.py，执行python eval_completion.py。 - 预期输出（确定性结果，非实测）：
completion_rate为0.2（5个用例中仅T1通过），失败原因为timeout、exception、wrong_output、partial_completion各1次。 - 验证判定：检查输出JSON中
total=5、passed=1、failed=4，且每个失败用例的reason与模拟类型一致，即证明评估框架工作正常。
失败原因分析要点
- 超时：需检查Agent推理链路是否存在死循环或外部调用阻塞。
- 异常：定位堆栈信息，修复代码健壮性问题。
- 错误输出：对比预期结果，检查任务理解或生成逻辑偏差。
- 部分完成：需结合业务定义“完成”的判定标准，本案例通过自定义
validator实现灵活判定。
扩展建议
真实场景中，测试用例集建议从历史对话、用户反馈中抽样构建，并覆盖典型难度与边界情况。评估结果可输出为JSON文件用于持续集成，或可视化展示完成率趋势。