42be31dd1f
活动级读模型单一出口：一次取齐报告 / 探索 / 分析 / 对比四大数据源。
markdown handler 从 30 行拼装逻辑缩到 3 行；分析执行器同步迁移。
comparison.py 内部的 8 次 load_campaign_report 调用暂不动（跨请求冗余，
缓存收益有限，改动风险高）。
- 新增 load_campaign_view(session, campaign) -> dict[str, Any]
- 返回 {report, exploration, analysis, comparison} 四键
- 迁移 markdown handler 和分析执行器两个调用点
- 4 个新测试覆盖 view 的组装逻辑
409 lines
17 KiB
Python
409 lines
17 KiB
Python
"""Report generation for evaluation runs.
|
||
|
||
Generation only: read the DB / model objects and build the report dict.
|
||
Formatting lives in ``report_render`` (pure dict → HTML/Markdown/JSON).
|
||
"""
|
||
|
||
from collections import defaultdict
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Optional
|
||
|
||
from sqlmodel import Session
|
||
|
||
from agenteval.evaluation.case_verdict import build_case_evidence, resolve_case_verdicts
|
||
from agenteval.evaluation.metrics import aggregate_runs
|
||
from agenteval.evaluation.report_render import render_html, render_json, render_markdown
|
||
from agenteval.models import Campaign, EvalRun, RunStatus, RunSummary
|
||
from agenteval.storage.db import DATA_DIR, iso_utc
|
||
from agenteval.storage.repository import RunRepository, ScenarioRepository, TargetRepository
|
||
from agenteval.utils.llm import extract_reply_text
|
||
|
||
|
||
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": [], "all_replied": True})
|
||
sent = turn.get_sent_message()
|
||
reply = turn.get_reply()
|
||
if reply is None:
|
||
case_map[turn.case_id]["all_replied"] = False
|
||
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": [], "all_replied": True})
|
||
case_map[result.case_id]["results"].append(
|
||
{
|
||
"rule_type": result.rule_type,
|
||
"passed": result.passed,
|
||
"score": result.score,
|
||
"reason": result.reason,
|
||
}
|
||
)
|
||
|
||
summary = run.summary or RunSummary()
|
||
errored_case_ids = {e.get("case_id") for e in summary.case_errors}
|
||
# 权威判定：引擎经 combine_case_outcome 算一次写入 summary，读路径只读不重算。
|
||
# resolve_case_verdicts 统一处理「权威优先、老 run 近似回退」（唯一落点）。
|
||
evidence = build_case_evidence(turns, results)
|
||
verdicts = resolve_case_verdicts(
|
||
case_outcomes=summary.case_outcomes,
|
||
evidence=evidence,
|
||
errored_case_ids=errored_case_ids,
|
||
)
|
||
|
||
cases = []
|
||
for case_id in sorted(case_map.keys()):
|
||
item = case_map[case_id]
|
||
verdict = verdicts[case_id]
|
||
cases.append(
|
||
{
|
||
"case_id": case_id,
|
||
"passed": verdict.passed,
|
||
"connectivity": verdict.connectivity,
|
||
"turns": sorted(item["turns"], key=lambda x: x["round"]),
|
||
"results": item["results"],
|
||
}
|
||
)
|
||
|
||
total_cases = summary.total_cases
|
||
passed_cases = summary.passed_cases
|
||
connectivity_count = sum(1 for c in cases if c["connectivity"])
|
||
judged_total = total_cases - connectivity_count
|
||
# 判定型通过率由 build_run_summary 入库，读路径只读；老 run 缺字段时按同一口径回退近似
|
||
judged_pass_rate = summary.judged_pass_rate
|
||
if judged_pass_rate is None and judged_total > 0:
|
||
judged_pass_rate = round((passed_cases - connectivity_count) / judged_total, 4)
|
||
|
||
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 "未知",
|
||
"scenario_version": run.scenario_version,
|
||
"triggered_by": run.triggered_by.value,
|
||
"status": run.status.value,
|
||
"started_at": iso_utc(run.started_at),
|
||
"completed_at": iso_utc(run.completed_at),
|
||
"summary": {
|
||
"total_cases": total_cases,
|
||
"passed_cases": passed_cases,
|
||
"failed_cases": summary.failed_cases,
|
||
"total_rules": summary.total_rules,
|
||
"passed_rules": summary.passed_rules,
|
||
"pass_rate": summary.pass_rate if summary.pass_rate is not None else 0.0,
|
||
"connectivity_cases": connectivity_count,
|
||
"judged_pass_rate": judged_pass_rate,
|
||
},
|
||
"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")
|
||
# 同场景还须同考纲版本才可比（ADR-0001）
|
||
if report_a.get("scenario_version") != report_b.get("scenario_version"):
|
||
raise ValueError(
|
||
"compare report requires the same scenario version "
|
||
f"(A: v{report_a.get('scenario_version')}, B: v{report_b.get('scenario_version')})"
|
||
)
|
||
|
||
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):
|
||
# None 仅表示该 run 没有这个用例；判定本身读权威 passed 字段
|
||
return c["passed"] if c else None
|
||
|
||
case_diffs.append(
|
||
{
|
||
"case_id": cid,
|
||
"connectivity": bool((ca and ca.get("connectivity")) or (cb and cb.get("connectivity"))),
|
||
"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"),
|
||
"scenario_version": report_a.get("scenario_version"),
|
||
"triggered_by": report_a.get("triggered_by"),
|
||
"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"),
|
||
"scenario_version": report_b.get("scenario_version"),
|
||
"triggered_by": report_b.get("triggered_by"),
|
||
"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 _to_utc(dt: Optional[datetime]) -> Optional[datetime]:
|
||
if dt is None:
|
||
return None
|
||
return dt.replace(tzinfo=timezone.utc) if dt.tzinfo is None else dt
|
||
|
||
|
||
def _run_window_offset(campaign: Campaign, run: EvalRun) -> float:
|
||
"""Position a Run within the (time-scaled) service window, in seconds.
|
||
|
||
``(started_at − campaign.started_at) × time_scale`` clamped to
|
||
``[0, window_seconds]`` — the single口径 shared by the report's bucketing and
|
||
the flat timeline, so a compressed dev run lands identically in both. Unstarted
|
||
Runs (either timestamp missing) sit at offset 0.
|
||
"""
|
||
window = float(campaign.window_seconds)
|
||
campaign_start = _to_utc(campaign.started_at)
|
||
run_start = _to_utc(run.started_at)
|
||
if campaign_start is None or run_start is None:
|
||
return 0.0
|
||
offset = (run_start - campaign_start).total_seconds() * campaign.time_scale
|
||
return max(0.0, min(offset, window))
|
||
|
||
|
||
def summarize_campaign_progress(campaign: Campaign, runs: list[EvalRun]) -> dict[str, Any]:
|
||
"""Compact list-row progress: completed vs *planned* total, plus pass_rate.
|
||
|
||
Unlike ``campaign_progress`` (live window position), this powers the list
|
||
view. ``planned_total`` is the sum of plan-entry counts — a fixed target the
|
||
campaign works toward, so the progress bar fills from 0 rather than tracking
|
||
a growing spawned count. ``overall_pass_rate`` reuses ``aggregate_runs`` so
|
||
the list figure matches the report exactly (ADR-0002: failures count as 0.0).
|
||
"""
|
||
return {
|
||
"completed_runs": sum(1 for r in runs if r.status == RunStatus.COMPLETED),
|
||
"planned_total": sum(entry.count for entry in campaign.plan),
|
||
"overall_pass_rate": aggregate_runs(runs)["pass_rate"],
|
||
}
|
||
|
||
|
||
def build_campaign_timeline(
|
||
campaign: Campaign,
|
||
runs: list[EvalRun],
|
||
*,
|
||
scenario_names: Optional[dict[str, str]] = None,
|
||
) -> list[dict[str, Any]]:
|
||
"""Flatten a campaign's child Runs into per-Run timeline entries.
|
||
|
||
Unlike ``generate_campaign_report`` (12-bucket aggregation), this returns one
|
||
entry per Run, sorted by window offset, for a process-timeline view. The
|
||
offset reuses the report's口径 — ``(started_at − campaign.started_at) ×
|
||
time_scale`` clamped to ``[0, window_seconds]`` — so a compressed dev run
|
||
lands at the same window position it reports. ``pass_rate`` / ``avg_latency_ms``
|
||
are read straight from each Run's summary (no re-aggregation); unstarted Runs
|
||
sit at offset 0.
|
||
"""
|
||
scenario_names = scenario_names or {}
|
||
|
||
entries = []
|
||
for run in runs:
|
||
offset = _run_window_offset(campaign, run)
|
||
summary = run.summary
|
||
entries.append({
|
||
"run_id": run.id,
|
||
"scenario_id": run.scenario_id,
|
||
"scenario_name": scenario_names.get(run.scenario_id, (run.scenario_id or "")[:8]),
|
||
"offset_seconds": round(offset, 3),
|
||
"status": run.status.value,
|
||
"pass_rate": summary.pass_rate if summary is not None else None,
|
||
"avg_latency_ms": summary.avg_latency_ms if summary is not None else None,
|
||
"started_at": iso_utc(run.started_at),
|
||
})
|
||
entries.sort(key=lambda e: e["offset_seconds"])
|
||
return entries
|
||
|
||
|
||
def generate_campaign_report(
|
||
campaign: Campaign,
|
||
runs: list[EvalRun],
|
||
*,
|
||
scenario_names: Optional[dict[str, str]] = None,
|
||
bucket_count: int = 12,
|
||
) -> dict[str, Any]:
|
||
"""Build a dual-axis periodic report for a campaign from its child Runs.
|
||
|
||
Axis 1 (time trend): child Runs bucketed by their position in the service
|
||
window, each bucket carrying pass_rate / availability / latency. Axis 2
|
||
(capability summary): the same measures grouped by scenario across the whole
|
||
window. Pure function — no I/O; ``scenario_names`` maps ids to display names.
|
||
|
||
``time_scale`` is used *only* to place each Run into the right window-time
|
||
bucket (so a compressed dev run still reports "hour 0-2, 2-4, ..."); it never
|
||
changes any aggregated number, keeping figures comparable across lines.
|
||
"""
|
||
scenario_names = scenario_names or {}
|
||
window = float(campaign.window_seconds)
|
||
bucket_seconds = window / bucket_count if bucket_count else window
|
||
|
||
# ── Axis 1: time trend ────────────────────────────────────────────────
|
||
buckets: dict[int, list[EvalRun]] = defaultdict(list)
|
||
for run in runs:
|
||
offset = _run_window_offset(campaign, run)
|
||
idx = min(int(offset / bucket_seconds), bucket_count - 1) if bucket_seconds else 0
|
||
buckets[idx].append(run)
|
||
|
||
time_trend = []
|
||
for idx in range(bucket_count):
|
||
agg = aggregate_runs(buckets.get(idx, []))
|
||
time_trend.append({
|
||
"bucket_index": idx,
|
||
"start_seconds": round(idx * bucket_seconds, 3),
|
||
"end_seconds": round((idx + 1) * bucket_seconds, 3),
|
||
**agg,
|
||
})
|
||
|
||
# ── Axis 2: capability summary (by scenario) ──────────────────────────
|
||
by_scenario: dict[str, list[EvalRun]] = defaultdict(list)
|
||
for run in runs:
|
||
by_scenario[run.scenario_id].append(run)
|
||
capability_summary = []
|
||
for sid, sruns in by_scenario.items():
|
||
agg = aggregate_runs(sruns)
|
||
capability_summary.append({
|
||
"scenario_id": sid,
|
||
"scenario_name": scenario_names.get(sid, (sid or "")[:8]),
|
||
**agg,
|
||
})
|
||
capability_summary.sort(key=lambda s: s["run_count"], reverse=True)
|
||
|
||
overall = aggregate_runs(runs)
|
||
return {
|
||
"campaign_id": campaign.id,
|
||
"name": campaign.name,
|
||
"target_id": campaign.target_id,
|
||
"status": campaign.status.value,
|
||
"window_seconds": campaign.window_seconds,
|
||
"time_scale": campaign.time_scale,
|
||
"started_at": iso_utc(campaign.started_at),
|
||
"completed_at": iso_utc(campaign.completed_at),
|
||
"summary": {
|
||
"total_runs": len(runs),
|
||
"completed_runs": sum(1 for r in runs if r.status == RunStatus.COMPLETED),
|
||
"overall_pass_rate": overall["pass_rate"],
|
||
"overall_availability": overall["availability"],
|
||
"avg_latency_ms": overall["avg_latency_ms"],
|
||
},
|
||
"time_trend": time_trend,
|
||
"capability_summary": capability_summary,
|
||
}
|
||
|
||
|
||
def load_campaign_report(session: Session, campaign: Campaign) -> dict[str, Any]:
|
||
"""取数 + 聚合一步完成：报告 / 分析 / 对比 / 导出共用的活动报告 dict 取法。"""
|
||
runs = RunRepository(session).list_by_campaign(campaign.id)
|
||
return generate_campaign_report(campaign, runs, scenario_names=ScenarioRepository(session).name_map())
|
||
|
||
|
||
def load_campaign_view(session: Session, campaign: Campaign) -> dict[str, Any]:
|
||
"""活动级读模型单一出口：一次取齐报告 / 探索 / 分析 / 对比。
|
||
|
||
四个数据源总是被一起取（markdown 导出、分析执行器、前端报告抽屉），
|
||
统一入口消除调用方的拼装逻辑。各子模块的取法保持不变，此处只做组合。
|
||
"""
|
||
from agenteval.evaluation.comparison import load_comparison_view
|
||
from agenteval.exploration.summary import summarize_campaign_exploration
|
||
from agenteval.storage.repository import CampaignAnalysisRepository
|
||
|
||
report = load_campaign_report(session, campaign)
|
||
exploration = summarize_campaign_exploration(session, campaign.id)
|
||
|
||
analysis_row = CampaignAnalysisRepository(session).get_by_campaign(campaign.id)
|
||
analysis = analysis_row.get_result() if analysis_row and analysis_row.status == "completed" else None
|
||
|
||
comparison_view = load_comparison_view(session, campaign)
|
||
comparison = comparison_view if comparison_view.get("status") != "none" else None
|
||
|
||
return {
|
||
"report": report,
|
||
"exploration": exploration,
|
||
"analysis": analysis,
|
||
"comparison": comparison,
|
||
}
|
||
|
||
|
||
def save_report(run_id: str, fmt: str = "html", output_dir: Optional[Path] = None) -> Path:
|
||
"""Generate a run report and save it to disk in the requested format."""
|
||
output_dir = output_dir or DATA_DIR / "reports"
|
||
output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
report = generate_report(run_id)
|
||
renderers = {
|
||
"html": (render_html, "html"),
|
||
"json": (render_json, "json"),
|
||
"markdown": (render_markdown, "md"),
|
||
}
|
||
if fmt not in renderers:
|
||
raise ValueError(f"unsupported report format: {fmt}")
|
||
render, ext = renderers[fmt]
|
||
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
path = output_dir / f"report_{run_id}_{timestamp}.{ext}"
|
||
path.write_text(render(report), encoding="utf-8")
|
||
return path
|