380 lines
15 KiB
Python
380 lines
15 KiB
Python
"""直接运行 product-design-module skill 的评估（内联 sub-agent 输出）。
|
||
|
||
本脚本：
|
||
1. 读取 evals.json 中的 eval cases
|
||
2. 从指定目录读取 sub-agent 的输出文件
|
||
3. 运行确定性检查（内置 + 自定义）
|
||
4. 生成 benchmark 报告（JSON + Markdown）
|
||
5. 输出改进建议
|
||
"""
|
||
|
||
import json
|
||
import re
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
PROJECT_ROOT = Path(__file__).parent
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
from eval_framework.checker import run_checks, load_custom_checks, CHECK_REGISTRY
|
||
from eval_framework.models import (
|
||
CheckResult, RunResult, CaseResult,
|
||
ModeResult, BenchmarkResult, TimingInfo, GradingSummary
|
||
)
|
||
|
||
|
||
def load_eval_cases(skill_name: str):
|
||
eval_path = PROJECT_ROOT / "evals" / skill_name / "evals.json"
|
||
with open(eval_path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
|
||
|
||
def load_agent_outputs(skill_name: str) -> dict:
|
||
output_dir = PROJECT_ROOT / "eval_output" / skill_name / "real_agent"
|
||
outputs = {}
|
||
for case_dir in sorted(output_dir.glob("case_*")):
|
||
case_id = int(case_dir.name.split("_")[1])
|
||
output_file = case_dir / "output.md"
|
||
if output_file.exists():
|
||
with open(output_file, "r", encoding="utf-8") as f:
|
||
outputs[case_id] = f.read()
|
||
return outputs
|
||
|
||
|
||
def run_evaluation(skill_name: str):
|
||
print(f"\n{'='*60}")
|
||
print(f" 评估 Skill: {skill_name}")
|
||
print(f"{'='*60}\n")
|
||
|
||
# 1. 加载 eval 配置
|
||
eval_data = load_eval_cases(skill_name)
|
||
evals = eval_data["evals"]
|
||
print(f" 加载了 {len(evals)} 个 eval case")
|
||
|
||
# 2. 加载 sub-agent 输出
|
||
agent_outputs = load_agent_outputs(skill_name)
|
||
print(f" 加载了 {len(agent_outputs)} 个 sub-agent 输出")
|
||
|
||
# 3. 加载自定义检查
|
||
custom_checks = load_custom_checks(skill_name)
|
||
if custom_checks:
|
||
print(f" 加载了 {len(custom_checks)} 个自定义检查: {list(custom_checks.keys())}")
|
||
|
||
# 4. 对每个 eval case 运行检查
|
||
case_results = []
|
||
total_checks = 0
|
||
total_passed = 0
|
||
all_check_details = []
|
||
|
||
for eval_case in evals:
|
||
case_id = eval_case["id"]
|
||
if case_id not in agent_outputs:
|
||
print(f"\n [SKIP] Case {case_id}: 无 sub-agent 输出")
|
||
continue
|
||
|
||
output = agent_outputs[case_id]
|
||
expectations = eval_case["expectations"]
|
||
|
||
print(f"\n --- Case {case_id}: {eval_case['prompt'][:50]}... ---")
|
||
print(f" 输出长度: {len(output)} 字符")
|
||
print(f" 期望条件数: {len(expectations)}")
|
||
|
||
start_time = time.time()
|
||
check_results = run_checks(output, expectations, custom_checks)
|
||
elapsed = time.time() - start_time
|
||
|
||
passed = sum(1 for r in check_results if r.passed)
|
||
total = len(check_results)
|
||
total_checks += total
|
||
total_passed += passed
|
||
|
||
print(f" 检查结果: {passed}/{total} 通过 ({passed/total*100:.1f}%)")
|
||
|
||
for r in check_results:
|
||
status = "PASS" if r.passed else "FAIL"
|
||
exp_short = r.expectation[:60] + ("..." if len(r.expectation) > 60 else "")
|
||
print(f" [{status}] {exp_short}")
|
||
if not r.passed:
|
||
print(f" 原因: {r.reason[:80]}")
|
||
all_check_details.append({
|
||
"case_id": case_id,
|
||
"expectation": r.expectation,
|
||
"passed": r.passed,
|
||
"reason": r.reason
|
||
})
|
||
|
||
run_result = RunResult(
|
||
eval_case_id=case_id,
|
||
run_index=1,
|
||
output=output[:500] + "..." if len(output) > 500 else output,
|
||
checks=[r.model_dump() for r in check_results],
|
||
timing=TimingInfo(total_seconds=elapsed)
|
||
)
|
||
|
||
case_result = CaseResult(
|
||
eval_case_id=case_id,
|
||
runs=[run_result],
|
||
pass_rate=1.0 if passed == total else 0.0,
|
||
avg_check_pass_rate=passed / total if total > 0 else 0.0
|
||
)
|
||
case_results.append(case_result)
|
||
|
||
# 5. 聚合结果
|
||
overall_pass_rate = total_passed / total_checks if total_checks > 0 else 0.0
|
||
passed_cases = sum(1 for c in case_results if c.pass_rate == 1.0)
|
||
|
||
# 6. 保存结果
|
||
output_dir = PROJECT_ROOT / "eval_output" / skill_name / "iteration-2"
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 保存 benchmark JSON
|
||
benchmark_data = {
|
||
"skill_name": skill_name,
|
||
"iteration": 2,
|
||
"agent_type": "real_sub_agent",
|
||
"timestamp": time.strftime('%Y-%m-%d %H:%M:%S'),
|
||
"grading": {
|
||
"total_cases": len(case_results),
|
||
"passed_cases": passed_cases,
|
||
"overall_pass_rate": overall_pass_rate,
|
||
"total_checks": total_checks,
|
||
"total_passed": total_passed
|
||
},
|
||
"case_results": []
|
||
}
|
||
|
||
for cr in case_results:
|
||
case_data = {
|
||
"eval_case_id": cr.eval_case_id,
|
||
"pass_rate": cr.pass_rate,
|
||
"avg_check_pass_rate": cr.avg_check_pass_rate,
|
||
"checks": all_check_details
|
||
}
|
||
benchmark_data["case_results"].append(case_data)
|
||
|
||
benchmark_path = output_dir / "benchmark.json"
|
||
with open(benchmark_path, "w", encoding="utf-8") as f:
|
||
json.dump(benchmark_data, f, ensure_ascii=False, indent=2)
|
||
print(f"\n Benchmark JSON 已保存: {benchmark_path}")
|
||
|
||
# 7. 生成 Markdown 报告
|
||
report = generate_markdown_report(benchmark_data, evals, case_results, all_check_details)
|
||
report_path = output_dir / "benchmark.md"
|
||
with open(report_path, "w", encoding="utf-8") as f:
|
||
f.write(report)
|
||
print(f" Benchmark 报告已保存: {report_path}")
|
||
|
||
# 8. 输出改进建议
|
||
suggestions = generate_improvement_suggestions(all_check_details)
|
||
suggestions_path = output_dir / "improvement_suggestions.md"
|
||
with open(suggestions_path, "w", encoding="utf-8") as f:
|
||
f.write(suggestions)
|
||
print(f" 改进建议已保存: {suggestions_path}")
|
||
|
||
return benchmark_data
|
||
|
||
|
||
def generate_markdown_report(benchmark_data, evals, case_results, all_checks):
|
||
lines = []
|
||
lines.append(f"# Skill 评估报告: {benchmark_data['skill_name']}")
|
||
lines.append(f"\n**迭代**: {benchmark_data['iteration']} (真实 Sub-Agent)")
|
||
lines.append(f"**评估时间**: {benchmark_data['timestamp']}")
|
||
lines.append(f"**Agent 类型**: {benchmark_data['agent_type']}")
|
||
lines.append("")
|
||
|
||
grading = benchmark_data["grading"]
|
||
lines.append("## 总体结果")
|
||
lines.append("")
|
||
lines.append("| 指标 | 值 |")
|
||
lines.append("|------|-----|")
|
||
lines.append(f"| Eval Cases 总数 | {grading['total_cases']} |")
|
||
lines.append(f"| 完全通过 Cases | {grading['passed_cases']} |")
|
||
lines.append(f"| 总体检查通过率 | {grading['overall_pass_rate']*100:.1f}% |")
|
||
lines.append(f"| 总检查项 | {grading['total_checks']} |")
|
||
lines.append(f"| 通过检查项 | {grading['total_passed']} |")
|
||
lines.append("")
|
||
|
||
# 各 Case 结果
|
||
lines.append("## 各 Case 结果概览")
|
||
lines.append("")
|
||
lines.append("| Case | Prompt 摘要 | 通过率 | 通过/总数 |")
|
||
lines.append("|------|-----------|--------|----------|")
|
||
|
||
for cr in case_results:
|
||
case_id = cr.eval_case_id
|
||
eval_case = next((e for e in evals if e["id"] == case_id), None)
|
||
prompt_short = eval_case["prompt"][:40] + "..." if eval_case else "N/A"
|
||
rate = cr.avg_check_pass_rate * 100
|
||
case_checks = [c for c in all_checks if c["case_id"] == case_id]
|
||
passed = sum(1 for c in case_checks if c["passed"])
|
||
total = len(case_checks)
|
||
lines.append(f"| {case_id} | {prompt_short} | {rate:.1f}% | {passed}/{total} |")
|
||
lines.append("")
|
||
|
||
# 详细检查结果
|
||
lines.append("## 详细检查结果")
|
||
lines.append("")
|
||
|
||
for cr in case_results:
|
||
case_id = cr.eval_case_id
|
||
eval_case = next((e for e in evals if e["id"] == case_id), None)
|
||
if not eval_case:
|
||
continue
|
||
|
||
lines.append(f"### Case {case_id}")
|
||
lines.append(f"**用户输入**: {eval_case['prompt']}")
|
||
lines.append(f"**通过率**: {cr.avg_check_pass_rate*100:.1f}%")
|
||
lines.append("")
|
||
|
||
case_checks = [c for c in all_checks if c["case_id"] == case_id]
|
||
lines.append("| # | 期望条件 | 结果 | 原因 |")
|
||
lines.append("|---|---------|------|------|")
|
||
for i, check in enumerate(case_checks):
|
||
status = "PASS" if check["passed"] else "FAIL"
|
||
exp = check["expectation"][:50] + ("..." if len(check["expectation"]) > 50 else "")
|
||
reason = check["reason"][:60] + ("..." if len(check["reason"]) > 60 else "")
|
||
lines.append(f"| {i+1} | {exp} | {status} | {reason} |")
|
||
lines.append("")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
def generate_improvement_suggestions(all_checks):
|
||
"""基于检查结果生成改进建议。"""
|
||
lines = []
|
||
lines.append("# Skill 改进建议")
|
||
lines.append("")
|
||
lines.append(f"生成时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
|
||
lines.append("")
|
||
|
||
# 分析失败模式
|
||
failed = [c for c in all_checks if not c["passed"]]
|
||
passed = [c for c in all_checks if c["passed"]]
|
||
|
||
lines.append("## 失败模式分析")
|
||
lines.append("")
|
||
|
||
# 按类别分组
|
||
categories = {
|
||
"三级结构": [],
|
||
"KANO分类": [],
|
||
"ICE评分": [],
|
||
"MindMapNode JSON": [],
|
||
"MindMapNode type约束": [],
|
||
"默认模块归属": [],
|
||
"汇总表": [],
|
||
"产品类型识别": [],
|
||
"特定模块覆盖": [],
|
||
"其他": []
|
||
}
|
||
|
||
for check in failed:
|
||
exp = check["expectation"]
|
||
if "三级" in exp or "结构" in exp or "层级" in exp:
|
||
categories["三级结构"].append(check)
|
||
elif "KANO" in exp and "type" not in exp.lower():
|
||
categories["KANO分类"].append(check)
|
||
elif "ICE" in exp:
|
||
categories["ICE评分"].append(check)
|
||
elif "MindMapNode" in exp and "type" in exp:
|
||
categories["MindMapNode type约束"].append(check)
|
||
elif "MindMapNode" in exp or "JSON" in exp:
|
||
categories["MindMapNode JSON"].append(check)
|
||
elif "默认模块" in exp or "归属" in exp:
|
||
categories["默认模块归属"].append(check)
|
||
elif "汇总表" in exp or "表格" in exp:
|
||
categories["汇总表"].append(check)
|
||
elif "识别" in exp or "混合型" in exp or "B端" in exp or "C端" in exp or "OA" in exp:
|
||
categories["产品类型识别"].append(check)
|
||
elif "包含" in exp and ("模块" in exp or "管理" in exp or "审批" in exp or "移动" in exp):
|
||
categories["特定模块覆盖"].append(check)
|
||
else:
|
||
categories["其他"].append(check)
|
||
|
||
lines.append(f"总失败数: {len(failed)}/{len(all_checks)}")
|
||
lines.append("")
|
||
for cat, checks in categories.items():
|
||
if checks:
|
||
lines.append(f"### {cat} ({len(checks)} 项失败)")
|
||
for c in checks:
|
||
lines.append(f"- Case {c['case_id']}: {c['expectation'][:60]}")
|
||
lines.append(f" 原因: {c['reason'][:80]}")
|
||
lines.append("")
|
||
|
||
# 改进建议
|
||
lines.append("## 具体改进建议")
|
||
lines.append("")
|
||
|
||
if categories["三级结构"]:
|
||
lines.append("### 1. 三级结构检查优化")
|
||
lines.append("- **问题**: 关键词检查无法识别语义等价的结构表达")
|
||
lines.append("- **建议**: 使用自定义检查函数 `three_level_structure` 替代关键词检查")
|
||
lines.append("- **建议**: 在 evals.json 中为结构类 expectation 添加 `check_type` 字段指定检查方式")
|
||
lines.append("")
|
||
|
||
if categories["KANO分类"]:
|
||
lines.append("### 2. KANO 分类检查优化")
|
||
lines.append("- **问题**: 期望 '每个L3功能点标注了KANO分类（基本型/期望型/兴奋型）' 要求三种都出现，但某些 case 可能只有两种")
|
||
lines.append("- **建议**: 使用自定义检查 `kano_classification` 替代关键词检查")
|
||
lines.append("- **建议**: 放宽检查条件——只要存在至少一种 KANO 分类即通过")
|
||
lines.append("")
|
||
|
||
if categories["ICE评分"]:
|
||
lines.append("### 3. ICE 评分检查优化")
|
||
lines.append("- **问题**: 关键词检查无法识别 `ICE：9×9×7=567` 这样的格式")
|
||
lines.append("- **建议**: 使用自定义检查 `ice_score` 替代关键词检查")
|
||
lines.append("")
|
||
|
||
if categories["MindMapNode type约束"]:
|
||
lines.append("### 4. MindMapNode type 约束检查优化")
|
||
lines.append("- **问题**: 关键词检查无法验证 JSON 中的 type 字段值")
|
||
lines.append("- **建议**: 使用自定义检查 `mindmap_type_constraint` 替代关键词检查")
|
||
lines.append("")
|
||
|
||
if categories["默认模块归属"]:
|
||
lines.append("### 5. 默认模块归属检查优化")
|
||
lines.append("- **问题**: 期望 '质量/资源/性能需求单独标注' 但不是所有 case 都需要这些模块")
|
||
lines.append("- **建议**: 使用自定义检查 `default_module_label` 替代关键词检查")
|
||
lines.append("- **建议**: 区分 '必须包含全部4个默认模块' 和 '至少标注了归属' 两种检查级别")
|
||
lines.append("")
|
||
|
||
if categories["产品类型识别"]:
|
||
lines.append("### 6. 产品类型识别检查优化")
|
||
lines.append("- **问题**: 关键词检查要求精确匹配 'B端C端混合型' 等表述，但 agent 可能使用不同表述")
|
||
lines.append("- **建议**: 使用语义化检查——只要输出中体现了 C端和B端的区分即可通过")
|
||
lines.append("")
|
||
|
||
if categories["特定模块覆盖"]:
|
||
lines.append("### 7. 特定模块覆盖检查优化")
|
||
lines.append("- **问题**: 关键词检查要求精确匹配模块名称（如 '租户管理'），但 agent 可能使用不同命名")
|
||
lines.append("- **建议**: 使用同义词匹配或语义检查")
|
||
lines.append("")
|
||
|
||
# 通用建议
|
||
lines.append("## 通用改进建议")
|
||
lines.append("")
|
||
lines.append("### 检查框架优化")
|
||
lines.append("1. **为每个 expectation 指定 check_type**: 在 evals.json 中添加 `check_type` 字段，明确指定使用哪个检查函数")
|
||
lines.append("2. **增加语义检查**: 对于产品类型识别、模块覆盖等，使用 LLM-as-Judge 进行语义评估")
|
||
lines.append("3. **分级检查**: 区分 '必须通过' 和 '建议通过' 两个级别")
|
||
lines.append("")
|
||
lines.append("### Skill 内容优化")
|
||
lines.append("1. **强化输出格式约束**: 在 skill 中更明确地要求输出格式，减少 agent 的自由发挥空间")
|
||
lines.append("2. **添加输出检查清单**: 在 skill 末尾添加自检清单，让 agent 在输出前自行验证")
|
||
lines.append("3. **统一术语**: 在 skill 中明确要求使用特定术语（如 'B端C端混合型'、'能力需求' 等）")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
skill_name = "product-design-module"
|
||
result = run_evaluation(skill_name)
|
||
|
||
print(f"\n{'='*60}")
|
||
print(f" 评估完成!")
|
||
print(f" 总体通过率: {result['grading']['overall_pass_rate']*100:.1f}%")
|
||
print(f" 通过 Cases: {result['grading']['passed_cases']}/{result['grading']['total_cases']}")
|
||
print(f" 通过检查项: {result['grading']['total_passed']}/{result['grading']['total_checks']}")
|
||
print(f"{'='*60}")
|