评测报告
评测对象：{{ report.target_name }}（{{ report.target_id }}）
评测场景：{{ report.scenario_name }}（{{ report.scenario_id }}）
执行时间：{{ report.started_at }} 至 {{ report.completed_at or '进行中' }}
{{ report.summary.total_cases }}
用例总数
{{ report.summary.passed_cases }}
通过用例
{{ report.summary.total_rules }}
规则总数
{{ "%.2f"|format(report.summary.pass_rate * 100) }}%
规则通过率
{% for case in report.cases %}
用例 {{ case.case_id }}
{% for turn in case.turns %}
用户消息
{{ turn.sent_text }}
智能体回复（{{ turn.latency_ms }}ms）
{{ turn.reply_text or '（无回复）' }}
{% endfor %}
{% for result in case.results %}
{{ '通过' if result.passed else '失败' }}
{{ result.rule_type }}: {{ result.reason }}
{% endfor %}
{% endfor %}
"""
def _extract_text(data: Any) -> str:
return extract_reply_text(data)
def generate_report(run_id: str, session=None) -> dict[str, Any]:
"""Build a structured report dict for a run."""
run_repo = RunRepository(session)
target_repo = TargetRepository(session)
scenario_repo = ScenarioRepository(session)
run = run_repo.get(run_id)
if not run:
raise ValueError(f"run not found: {run_id}")
target = target_repo.get(run.target_id)
scenario = scenario_repo.get(run.scenario_id)
turns = run_repo.get_turns(run_id)
results = run_repo.get_results(run_id)
# Group by case
case_map: dict[str, dict[str, Any]] = {}
for turn in turns:
case_map.setdefault(turn.case_id, {"turns": [], "results": []})
sent = turn.get_sent_message()
reply = turn.get_reply()
case_map[turn.case_id]["turns"].append(
{
"round": turn.round_index,
"sent_text": _extract_text(sent.get("msgBody")),
"reply_text": _extract_text(reply.get("msgBody") if reply else None),
"latency_ms": turn.latency_ms,
"question_msg_id": turn.question_msg_id,
}
)
for result in results:
case_map.setdefault(result.case_id, {"turns": [], "results": []})
case_map[result.case_id]["results"].append(
{
"rule_type": result.rule_type,
"passed": result.passed,
"score": result.score,
"reason": result.reason,
}
)
cases = []
for case_id in sorted(case_map.keys()):
item = case_map[case_id]
cases.append(
{
"case_id": case_id,
"turns": sorted(item["turns"], key=lambda x: x["round"]),
"results": item["results"],
}
)
summary = run.summary or {}
return {
"run_id": run.id,
"target_id": run.target_id,
"target_name": target.name if target else "未知",
"scenario_id": run.scenario_id,
"scenario_name": scenario.name if scenario else "未知",
"status": run.status.value,
"started_at": iso_utc(run.started_at),
"completed_at": iso_utc(run.completed_at),
"summary": {
"total_cases": summary.get("total_cases", 0),
"passed_cases": summary.get("passed_cases", 0),
"failed_cases": summary.get("failed_cases", 0),
"total_rules": summary.get("total_rules", 0),
"passed_rules": summary.get("passed_rules", 0),
"pass_rate": summary.get("pass_rate", 0.0),
},
"cases": cases,
}
def generate_compare_report(run_id_1: str, run_id_2: str, session=None) -> dict[str, Any]:
"""Build a side-by-side comparison dict for two runs of the same scenario."""
report_a = generate_report(run_id_1, session)
report_b = generate_report(run_id_2, session)
# Cross-scenario case_ids never overlap, so every case would be flagged
# "changed" and the diff would be meaningless — reject early.
if report_a.get("scenario_id") != report_b.get("scenario_id"):
raise ValueError("compare report requires both runs to use the same scenario")
def _summary_delta(key: str) -> float:
return report_b["summary"][key] - report_a["summary"][key]
# Case-level diff: match by case_id
cases_a = {c["case_id"]: c for c in report_a.get("cases", [])}
cases_b = {c["case_id"]: c for c in report_b.get("cases", [])}
all_case_ids = sorted(set(cases_a) | set(cases_b))
case_diffs = []
for cid in all_case_ids:
ca = cases_a.get(cid)
cb = cases_b.get(cid)
def _case_passed(c):
if not c:
return None
results = c.get("results", [])
if not results:
# No rule results (e.g. errored case) must not count as passed.
return None
return all(r["passed"] for r in results)
case_diffs.append(
{
"case_id": cid,
"run_a_passed": _case_passed(ca),
"run_b_passed": _case_passed(cb),
"changed": _case_passed(ca) != _case_passed(cb),
"run_a_results": ca["results"] if ca else [],
"run_b_results": cb["results"] if cb else [],
}
)
return {
"run_a": {
"run_id": run_id_1,
"target_name": report_a.get("target_name"),
"scenario_name": report_a.get("scenario_name"),
"status": report_a.get("status"),
"started_at": report_a.get("started_at"),
"summary": report_a["summary"],
},
"run_b": {
"run_id": run_id_2,
"target_name": report_b.get("target_name"),
"scenario_name": report_b.get("scenario_name"),
"status": report_b.get("status"),
"started_at": report_b.get("started_at"),
"summary": report_b["summary"],
},
"delta": {
"pass_rate": round(_summary_delta("pass_rate"), 4),
"passed_cases": int(_summary_delta("passed_cases")),
"passed_rules": int(_summary_delta("passed_rules")),
},
"cases": case_diffs,
"changed_cases": sum(1 for c in case_diffs if c["changed"]),
}
def render_markdown_report(run_id: str, session=None) -> str:
"""Render a report as Markdown string."""
report = generate_report(run_id, session)
s = report["summary"]
lines: list[str] = [
f"# 评测报告 — {report.get('scenario_name', run_id)}",
"",
f"**评测对象**: {report.get('target_name', '-')} ",
f"**评测场景**: {report.get('scenario_name', '-')} ",
f"**状态**: {report.get('status', '-')} ",
f"**开始时间**: {report.get('started_at', '-')} ",
f"**完成时间**: {report.get('completed_at', '-')} ",
"",
"## 汇总",
"",
"| 指标 | 数值 |",
"|------|------|",
f"| 总用例数 | {s['total_cases']} |",
f"| 通过用例 | {s['passed_cases']} |",
f"| 失败用例 | {s['failed_cases']} |",
f"| 总规则数 | {s['total_rules']} |",
f"| 通过规则 | {s['passed_rules']} |",
f"| 通过率 | {s['pass_rate'] * 100:.1f}% |",
"",
"## 用例明细",
"",
]
for case in report.get("cases", []):
case_passed = all(r["passed"] for r in case.get("results", []))
badge = "✅" if case_passed else "❌"
lines.append(f"### {badge} 用例 `{case['case_id']}`")
lines.append("")
for turn in case.get("turns", []):
lines.append(f"**第 {turn['round']} 轮**")
lines.append("")
lines.append(f"> **用户**: {turn.get('sent_text', '—')}")
lines.append("")
reply = turn.get("reply_text") or "（无回复）"
lines.append(f"> **智能体**: {reply}")
if turn.get("latency_ms") is not None:
lines.append(f"> *延迟: {turn['latency_ms']}ms*")
lines.append("")
if case.get("results"):
lines.append("**规则评估结果**")
lines.append("")
lines.append("| 规则 | 结果 | 评分 | 说明 |")
lines.append("|------|------|------|------|")
for r in case["results"]:
badge = "✅" if r["passed"] else "❌"
score = f"{r['score']:.2f}" if r.get("score") is not None else "-"
lines.append(f"| {r['rule_type']} | {badge} | {score} | {r.get('reason', '')} |")
lines.append("")
return "\n".join(lines)
def render_json_report(run_id: str, session=None) -> str:
"""Render a report as JSON string."""
report = generate_report(run_id, session)
return json.dumps(report, ensure_ascii=False, indent=2)
def render_html_report(run_id: str, session=None) -> str:
"""Render a report as HTML string."""
report = generate_report(run_id, session)
template = Template(HTML_TEMPLATE)
return template.render(report=report)
def save_report(run_id: str, fmt: str = "html", output_dir: Optional[Path] = None) -> Path:
"""Generate and save a report to disk."""
output_dir = output_dir or DATA_DIR / "reports"
output_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
if fmt == "html":
content = render_html_report(run_id)
path = output_dir / f"report_{run_id}_{timestamp}.html"
elif fmt == "json":
content = render_json_report(run_id)
path = output_dir / f"report_{run_id}_{timestamp}.json"
elif fmt == "markdown":
content = render_markdown_report(run_id)
path = output_dir / f"report_{run_id}_{timestamp}.md"
else:
raise ValueError(f"unsupported report format: {fmt}")
path.write_text(content, encoding="utf-8")
return path