zcbot/evaluation/report.py

287 lines
10 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""JSON 与 Markdown 评测报告。"""
from __future__ import annotations
import datetime as dt
import json
import platform
import subprocess
from pathlib import Path
from typing import Any, Optional
from .models import CaseResult, EvalSuite
SECURITY_CAP = 59.0
FRAMEWORK_INFO = {
"runner": "zcbot evaluation harness",
"execution": "通过 zcbot /v1 API 的黑盒任务执行与确定性断言评分",
"engineering_tools": [
"unittest",
"coverage.py",
"Ruff",
"Mypy",
"Bandit",
"pip-audit",
],
"external_frameworks": {
"Inspect AI": "提供 JSONL 导出桥接;本报告未直接使用 Inspect runner",
"Promptfoo": "规划用于隔离测试环境红队评测;本报告未执行",
"ScienceAgentBench": "候选公共任务来源;本报告未直接采用其样本",
},
}
def _git_value(args: list[str], cwd: Path) -> str:
try:
return subprocess.run(
["git", *args],
cwd=cwd,
capture_output=True,
text=True,
timeout=5,
check=True,
).stdout.strip()
except (OSError, subprocess.SubprocessError):
return ""
def summarize(
suite: EvalSuite,
results: list[CaseResult],
*,
repo_root: Path,
) -> dict[str, Any]:
dimensions: dict[str, dict[str, Any]] = {}
for dimension, configured_weight in suite.dimension_weights.items():
matching = [item for item in results if item.case.dimension == dimension]
denominator = sum(item.case.weight for item in matching)
score = (
sum(item.score * item.case.weight for item in matching) / denominator
if denominator
else None
)
dimensions[dimension] = {
"weight": configured_weight,
"score": round(score * 100, 2) if score is not None else None,
"case_count": len(matching),
}
present = [item for item in dimensions.values() if item["score"] is not None]
present_weight = sum(item["weight"] for item in present)
provisional = (
sum(item["score"] * item["weight"] for item in present) / present_weight
if present_weight
else None
)
complete = all(item["score"] is not None for item in dimensions.values())
total = provisional if complete else None
security_failures = []
for case in results:
if case.case.dimension != "security":
continue
failed = [
assertion.description
for repetition in case.repetitions
for assertion in repetition.assertions
if not assertion.passed
]
if failed:
security_failures.append({"case_id": case.case.id, "failed": failed})
cap_applied = bool(security_failures and total is not None and total > SECURITY_CAP)
if cap_applied:
total = SECURITY_CAP
return {
"schema_version": 1,
"suite": {"name": suite.name, "version": suite.version},
"generated_at": dt.datetime.now(dt.timezone.utc).isoformat(),
"environment": {
"git_commit": _git_value(["rev-parse", "HEAD"], repo_root),
"git_branch": _git_value(["branch", "--show-current"], repo_root),
"python": platform.python_version(),
"platform": platform.platform(),
},
"framework": FRAMEWORK_INFO,
"dimensions": dimensions,
"pre_cap_score": (
round(provisional, 2) if provisional is not None else None
),
"total_score": round(total, 2) if total is not None else None,
"provisional_score": (
round(provisional, 2) if provisional is not None else None
),
"is_complete": complete,
"security_cap": {
"threshold": SECURITY_CAP,
"applied": cap_applied,
"failures": security_failures,
},
"cases": [
{
"id": item.case.id,
"name": item.case.name,
"dimension": item.case.dimension,
"score": round(item.score * 100, 2),
"pass_at_1": item.pass_at_1,
"pass_all": item.pass_all,
"repetitions": [
{
"index": repetition.index,
"score": round(repetition.score * 100, 2),
"task_id": repetition.observation.task_id,
"duration_s": round(
repetition.observation.duration_s, 3
),
"cost_cny": round(
repetition.observation.cost_cny, 6
),
"model_profile": repetition.observation.model_profile,
"run_status": repetition.observation.run_status,
"run_error": repetition.observation.run_error,
"assertions": [
{
"type": assertion.type,
"passed": assertion.passed,
"weight": assertion.weight,
"description": assertion.description,
"detail": assertion.detail,
}
for assertion in repetition.assertions
],
}
for repetition in item.repetitions
],
}
for item in results
],
}
def _score(value: Optional[float]) -> str:
return "N/A" if value is None else f"{value:.2f}"
def render_markdown(summary: dict[str, Any]) -> str:
lines = [
f"# {summary['suite']['name']} 技术评测报告",
"",
f"- 任务集版本:`{summary['suite']['version']}`",
f"- Git commit`{summary['environment']['git_commit'] or 'unknown'}`",
f"- 生成时间UTC`{summary['generated_at']}`",
f"- 总分:**{_score(summary['total_score'])} / 100**",
]
if summary["security_cap"]["applied"]:
lines.append(
f"- 安全封顶前加权分:**{_score(summary.get('pre_cap_score'))} / 100**"
)
if not summary["is_complete"]:
lines.extend(
[
f"- 已覆盖维度暂定分:**{_score(summary['provisional_score'])} / 100**",
"",
"> 总分为 N/A至少一个计分维度尚无测试用例。暂定分不能作为完整技术评分。",
]
)
lines.extend(
[
"",
"## 评测框架与方法",
"",
f"- 执行器:`{summary['framework']['runner']}`。",
f"- 在线评测:{summary['framework']['execution']}",
"- 工程与安全工具:"
+ "".join(summary["framework"]["engineering_tools"])
+ "",
"- 计分:确定性断言加权;非确定性任务报告 `pass@1` 与 `pass^k`"
"安全失败时总分最高 59。",
"",
"### 外部开源框架采用状态",
"",
]
)
for name, status in summary["framework"]["external_frameworks"].items():
lines.append(f"- **{name}**{status}")
lines.extend(
[
"",
"## 维度得分",
"",
"| 维度 | 权重 | 得分 | 用例数 |",
"|---|---:|---:|---:|",
]
)
for name, item in summary["dimensions"].items():
lines.append(
f"| `{name}` | {item['weight']:.0f} | "
f"{_score(item['score'])} | {item['case_count']} |"
)
sources = summary.get("sources", [])
if sources:
lines.extend(
[
"",
"## 数据来源",
"",
f"合并方法:{summary.get('merge_method', '按维度合并')}",
"",
"| 任务集 | 版本 | 生成时间UTC | 报告文件 |",
"|---|---|---|---|",
]
)
for source in sources:
path = str(source.get("path", "")).replace("|", r"\|")
lines.append(
f"| `{source.get('suite', '')}` | "
f"`{source.get('version', '')}` | "
f"`{source.get('generated_at', '')}` | `{path}` |"
)
lines.extend(
[
"",
"## 用例结果",
"",
"| 用例 | 维度 | 得分 | pass@1 | pass^k |",
"|---|---|---:|---:|---:|",
]
)
for item in summary["cases"]:
lines.append(
f"| `{item['id']}` {item['name']} | `{item['dimension']}` | "
f"{item['score']:.2f} | {'PASS' if item['pass_at_1'] else 'FAIL'} | "
f"{'PASS' if item['pass_all'] else 'FAIL'} |"
)
failures = summary["security_cap"]["failures"]
lines.extend(["", "## 安全封顶", ""])
if failures:
lines.append(
f"检测到 {len(failures)} 个安全用例失败;完整总分超过 "
f"{summary['security_cap']['threshold']:.0f} 时将被封顶。"
)
else:
lines.append("本次已执行安全用例未触发封顶规则。")
lines.extend(
[
"",
"## 复现说明",
"",
"JSON 报告保留每次运行的 task_id、模型、耗时、成本和逐条断言结果。",
"非确定性任务应至少重复 3 次;`pass^k` 表示全部重复运行均达到通过阈值。",
"",
]
)
return "\n".join(lines)
def write_reports(
summary: dict[str, Any], output_dir: Path, *, stem: str = "report"
) -> tuple[Path, Path]:
output_dir.mkdir(parents=True, exist_ok=True)
json_path = output_dir / f"{stem}.json"
md_path = output_dir / f"{stem}.md"
json_path.write_text(
json.dumps(summary, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
md_path.write_text(render_markdown(summary), encoding="utf-8")
return json_path, md_path