From f433ebb970d8fedf556f947acb332e05454a0080 Mon Sep 17 00:00:00 2001 From: sinohqb Date: Thu, 30 Jul 2026 13:55:32 +0800 Subject: [PATCH] feat(campaigns): dual-axis periodic report (time trend + capability) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add generate_campaign_report: a pure aggregator over a campaign's child Runs producing a time-trend axis (Runs bucketed by service-window position) and a capability-summary axis (grouped by scenario), each carrying pass_rate / availability / latency. pass_rate keeps the single-Run case-level meaning and counts execution failures as 0.0 (ADR-0002); time_scale only places Runs into window-time buckets and never alters any figure. Engine summary now records avg_latency_ms to feed the latency axis. Expose GET /api/campaigns/{id}/report (structured) and .../report/markdown (reusing the existing Markdown export path). Adds "可用性/Availability" to the domain glossary. --- CONTEXT.md | 4 + backend/agenteval/evaluation/engine.py | 6 + backend/agenteval/evaluation/report.py | 182 ++++++++++++++++++++- backend/agenteval/web/routers/campaigns.py | 28 +++- tests/integration/test_campaigns_api.py | 54 ++++++ tests/unit/test_campaign_report.py | 153 +++++++++++++++++ 6 files changed, 425 insertions(+), 2 deletions(-) create mode 100644 tests/unit/test_campaign_report.py diff --git a/CONTEXT.md b/CONTEXT.md index f0471ab..2041218 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -70,6 +70,10 @@ _Avoid_: 时长、周期（周期单指这个窗口） 活动在窗口内"何时、对哪个对象、跑哪个场景、多大强度、以何种用户人设"的时间编排。由 OpenClaw 作为"虚拟用户大脑"在活动层生成，并在窗口内的决策点依据已完成时段的结果**自适应调整**后续编排；平台调度器负责耐久执行（派生 Run、重启后续跑、决策点唤醒 OpenClaw）。计划编排已有场景，时段内的具体对话仍由动态用例生成器产出。 _Avoid_: 排程、日程表 +**可用性（Availability）**: +活动周期报告中的一个维度：某时段（或整窗）内**正常完成**的子 Run 占比（completed / 已派生）。与通过率正交——通过率反映"回答质量"（用例级、含故障判不通过，ADR-0002），可用性反映"服务是否可达/执行是否成功"。通道故障导致的失败子 Run 拉低可用性。 +_Avoid_: 在线率、健康度 + ## 模型配置 **模型能力（Capability）**: diff --git a/backend/agenteval/evaluation/engine.py b/backend/agenteval/evaluation/engine.py index f54512c..1f56c77 100644 --- a/backend/agenteval/evaluation/engine.py +++ b/backend/agenteval/evaluation/engine.py @@ -173,6 +173,10 @@ class EvalEngine: total_rules = len(results) passed_rules = sum(1 for r in results if r.passed) + turns = self.run_repo.get_turns(run.id) + latencies = [t.latency_ms for t in turns if t.latency_ms is not None] + avg_latency_ms = round(sum(latencies) / len(latencies), 1) if latencies else None + summary = { "total_cases": total_cases, "passed_cases": passed_cases, @@ -181,6 +185,8 @@ class EvalEngine: "passed_rules": passed_rules, # 通过率是用例级口径（CONTEXT.md）；规则级数字保留在 passed_rules/total_rules "pass_rate": round(passed_cases / total_cases, 4) if total_cases else 0.0, + # 平均时延（毫秒），供活动周期报告的时延轴聚合；无回复轮不计入 + "avg_latency_ms": avg_latency_ms, # 逐用例权威判定（judgement.py 算一次），报告/对比/渲染层只读不重算 "case_outcomes": case_outcomes, } diff --git a/backend/agenteval/evaluation/report.py b/backend/agenteval/evaluation/report.py index e7225e0..1abb7ca 100644 --- a/backend/agenteval/evaluation/report.py +++ b/backend/agenteval/evaluation/report.py @@ -1,12 +1,14 @@ """Report generation for evaluation runs.""" import json -from datetime import datetime +from collections import defaultdict +from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional from jinja2 import Template +from agenteval.models import Campaign, EvalRun, RunStatus 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 @@ -283,6 +285,184 @@ def generate_compare_report(run_id_1: str, run_id_2: str, session=None) -> dict[ } +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 _aggregate_runs(runs: list[EvalRun]) -> dict[str, Any]: + """Aggregate a set of child Runs into pass_rate / availability / latency. + + pass_rate follows the single-Run case-level rate and *includes execution + failures* (ADR-0002): a failed child Run contributes 0.0, so a bad time + slice drags the curve down. availability is the completed fraction (an + orthogonal execution-success signal). latency averages only completed Runs + that recorded one. time_scale never enters these numbers. + """ + n = len(runs) + if n == 0: + return {"run_count": 0, "pass_rate": None, "availability": None, "avg_latency_ms": None} + completed = [r for r in runs if r.status == RunStatus.COMPLETED] + pass_rates = [ + (r.summary or {}).get("pass_rate", 0.0) + if (r.status == RunStatus.COMPLETED and isinstance(r.summary, dict)) + else 0.0 + for r in runs + ] + latencies = [ + r.summary["avg_latency_ms"] + for r in completed + if isinstance(r.summary, dict) and r.summary.get("avg_latency_ms") is not None + ] + return { + "run_count": n, + "pass_rate": round(sum(pass_rates) / n, 4), + "availability": round(len(completed) / n, 4), + "avg_latency_ms": round(sum(latencies) / len(latencies), 1) if latencies else None, + } + + +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 + campaign_start = _to_utc(campaign.started_at) + + # ── Axis 1: time trend ──────────────────────────────────────────────── + buckets: dict[int, list[EvalRun]] = defaultdict(list) + for run in runs: + run_start = _to_utc(run.started_at) + if campaign_start is None or run_start is None: + offset = 0.0 + else: + offset = (run_start - campaign_start).total_seconds() * campaign.time_scale + offset = max(0.0, min(offset, window)) + 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 render_campaign_markdown_report( + campaign: Campaign, + runs: list[EvalRun], + *, + scenario_names: Optional[dict[str, str]] = None, +) -> str: + """Render the dual-axis campaign report as Markdown (reuses the export path).""" + report = generate_campaign_report(campaign, runs, scenario_names=scenario_names) + s = report["summary"] + + def _pct(v: Optional[float]) -> str: + return "—" if v is None else f"{v * 100:.1f}%" + + def _ms(v: Optional[float]) -> str: + return "—" if v is None else f"{v:.0f}ms" + + lines: list[str] = [ + f"# 活动周期报告 — {report['name']}", + "", + f"**状态**: {report['status']} ", + f"**窗口**: {report['window_seconds']}s（倍速 {report['time_scale']}） ", + f"**开始时间**: {report['started_at'] or '-'} ", + f"**完成时间**: {report['completed_at'] or '-'} ", + "", + "## 汇总", + "", + "| 指标 | 数值 |", + "|------|------|", + f"| 子运行总数 | {s['total_runs']} |", + f"| 已完成 | {s['completed_runs']} |", + f"| 整窗通过率 | {_pct(s['overall_pass_rate'])} |", + f"| 整窗可用性 | {_pct(s['overall_availability'])} |", + f"| 平均时延 | {_ms(s['avg_latency_ms'])} |", + "", + "## 时间趋势", + "", + "| 时段(秒) | 运行数 | 通过率 | 可用性 | 时延 |", + "|------|------|------|------|------|", + ] + for b in report["time_trend"]: + lines.append( + f"| {b['start_seconds']:.0f}–{b['end_seconds']:.0f} | {b['run_count']} | " + f"{_pct(b['pass_rate'])} | {_pct(b['availability'])} | {_ms(b['avg_latency_ms'])} |" + ) + lines += [ + "", + "## 能力汇总", + "", + "| 场景 | 运行数 | 通过率 | 可用性 | 时延 |", + "|------|------|------|------|------|", + ] + for c in report["capability_summary"]: + lines.append( + f"| {c['scenario_name']} | {c['run_count']} | {_pct(c['pass_rate'])} | " + f"{_pct(c['availability'])} | {_ms(c['avg_latency_ms'])} |" + ) + return "\n".join(lines) + + def render_markdown_report(run_id: str, session=None) -> str: """Render a report as Markdown string.""" report = generate_report(run_id, session) diff --git a/backend/agenteval/web/routers/campaigns.py b/backend/agenteval/web/routers/campaigns.py index f9c2eca..fc034d8 100644 --- a/backend/agenteval/web/routers/campaigns.py +++ b/backend/agenteval/web/routers/campaigns.py @@ -7,11 +7,12 @@ the live window position and spawned/completed Run counts, and a campaign can be cancelled mid-flight. """ -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Response from pydantic import BaseModel, Field from sqlmodel import Session from agenteval.evaluation.campaign_runner import campaign_progress, request_cancel, start_campaign +from agenteval.evaluation.report import generate_campaign_report, render_campaign_markdown_report from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus from agenteval.storage.db import utc_now from agenteval.storage.repository import ( @@ -84,6 +85,31 @@ async def cancel_campaign(campaign_id: str, session: Session = Depends(get_db)) return campaign.model_dump() +@router.get("/{campaign_id}/report") +async def get_campaign_report(campaign_id: str, session: Session = Depends(get_db)) -> dict: + campaign = CampaignRepository(session).get(campaign_id) + if not campaign: + raise HTTPException(status_code=404, detail="campaign not found") + runs = RunRepository(session).list_by_campaign(campaign_id) + scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()} + return generate_campaign_report(campaign, runs, scenario_names=scenario_names) + + +@router.get("/{campaign_id}/report/markdown") +async def get_campaign_report_markdown(campaign_id: str, session: Session = Depends(get_db)) -> Response: + campaign = CampaignRepository(session).get(campaign_id) + if not campaign: + raise HTTPException(status_code=404, detail="campaign not found") + runs = RunRepository(session).list_by_campaign(campaign_id) + scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()} + md = render_campaign_markdown_report(campaign, runs, scenario_names=scenario_names) + return Response( + content=md, + media_type="text/markdown; charset=utf-8", + headers={"Content-Disposition": f'attachment; filename="campaign-report-{campaign_id}.md"'}, + ) + + @router.get("/{campaign_id}") async def get_campaign(campaign_id: str, session: Session = Depends(get_db)) -> dict: campaign = CampaignRepository(session).get(campaign_id) diff --git a/tests/integration/test_campaigns_api.py b/tests/integration/test_campaigns_api.py index 681efdc..f140261 100644 --- a/tests/integration/test_campaigns_api.py +++ b/tests/integration/test_campaigns_api.py @@ -216,3 +216,57 @@ async def test_detail_includes_progress_fields(client, seeded_db): assert progress["spawned_runs"] == 0 assert progress["completed_runs"] == 0 assert progress["current_offset_seconds"] == 0.0 + + +# ── campaign report endpoint (ticket 04) ───────────────────────────────────── + +async def test_campaign_report_structure_and_values(client, seeded_db): + from agenteval.models import EvalRun, RunStatus + + campaign_id = (await client.post("/api/campaigns", json=_valid_payload())).json()["id"] + + repo = RunRepository(seeded_db) + repo.create(EvalRun( + target_id="t-1", scenario_id="s-1", campaign_id=campaign_id, + status=RunStatus.COMPLETED, + summary={"total_cases": 1, "passed_cases": 1, "pass_rate": 1.0, "avg_latency_ms": 100}, + )) + repo.create(EvalRun( + target_id="t-1", scenario_id="s-1", campaign_id=campaign_id, + status=RunStatus.COMPLETED, + summary={"total_cases": 1, "passed_cases": 0, "pass_rate": 0.0, "avg_latency_ms": 200}, + )) + + report = (await client.get(f"/api/campaigns/{campaign_id}/report")).json() + assert report["campaign_id"] == campaign_id + assert "time_trend" in report and "capability_summary" in report + assert report["summary"]["total_runs"] == 2 + assert report["summary"]["completed_runs"] == 2 + assert report["summary"]["overall_pass_rate"] == 0.5 + cap = next(c for c in report["capability_summary"] if c["scenario_id"] == "s-1") + assert cap["run_count"] == 2 + assert cap["scenario_name"] == "mock-scenario" + + +async def test_campaign_report_missing_404(client, seeded_db): + resp = await client.get("/api/campaigns/nope/report") + assert resp.status_code == 404 + + +async def test_campaign_report_markdown_export(client, seeded_db): + from agenteval.models import EvalRun, RunStatus + + campaign_id = (await client.post("/api/campaigns", json=_valid_payload())).json()["id"] + RunRepository(seeded_db).create(EvalRun( + target_id="t-1", scenario_id="s-1", campaign_id=campaign_id, + status=RunStatus.COMPLETED, + summary={"total_cases": 1, "passed_cases": 1, "pass_rate": 1.0, "avg_latency_ms": 100}, + )) + + resp = await client.get(f"/api/campaigns/{campaign_id}/report/markdown") + assert resp.status_code == 200 + assert "text/markdown" in resp.headers["content-type"] + assert "attachment" in resp.headers["content-disposition"] + assert "# 活动周期报告" in resp.text + assert "## 时间趋势" in resp.text + assert "## 能力汇总" in resp.text diff --git a/tests/unit/test_campaign_report.py b/tests/unit/test_campaign_report.py new file mode 100644 index 0000000..e5f4342 --- /dev/null +++ b/tests/unit/test_campaign_report.py @@ -0,0 +1,153 @@ +"""Unit tests for generate_campaign_report (dual-axis campaign report).""" + +from datetime import datetime, timedelta, timezone + +import pytest +from sqlmodel import Session, SQLModel, create_engine + +from agenteval.evaluation.report import generate_campaign_report +from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, EvalRun, RunStatus +from agenteval.storage.repository import CampaignRepository, RunRepository + +T0 = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc) + + +@pytest.fixture() +def report_session(tmp_path): + from agenteval.storage.db import ( # noqa: F401 + CampaignDB, EvalResultDB, EvalRunDB, EvalTargetDB, FileCategoryDB, + FileRecordDB, ScenarioDB, TurnDB, + ) + engine = create_engine( + f"sqlite:///{tmp_path / 'campaign_report.db'}", + connect_args={"check_same_thread": False}, + ) + SQLModel.metadata.create_all(engine) + session = Session(engine) + try: + yield session + finally: + session.close() + engine.dispose() + + +def _seed_campaign(session, *, window_seconds=12, time_scale=1.0) -> Campaign: + campaign = CampaignRepository(session).create(Campaign( + name="cycle", target_id="t-1", window_seconds=window_seconds, time_scale=time_scale, + plan=[CampaignPlanEntry(scenario_id="s-a", offset_seconds=0, count=1)], + )) + campaign.status = CampaignStatus.RUNNING + campaign.started_at = T0 + return CampaignRepository(session).update(campaign) + + +def _seed_child(session, campaign_id, scenario_id, status, offset, *, pass_rate=None, latency=None) -> EvalRun: + summary = None + if status == RunStatus.COMPLETED: + summary = { + "total_cases": 1, + "passed_cases": 1 if (pass_rate or 0) >= 1 else 0, + "pass_rate": pass_rate, + "avg_latency_ms": latency, + } + return RunRepository(session).create(EvalRun( + target_id="t-1", scenario_id=scenario_id, campaign_id=campaign_id, + status=status, started_at=T0 + timedelta(seconds=offset), summary=summary, + )) + + +def _bucket(report, idx): + return report["time_trend"][idx] + + +def _cap(report, sid): + return next(c for c in report["capability_summary"] if c["scenario_id"] == sid) + + +# ── happy path: multi-bucket, multi-scenario, with a failure ──────────────── + +def test_dual_axis_values(report_session): + campaign = _seed_campaign(report_session) # window 12s, 12 buckets → 1s each + _seed_child(report_session, campaign.id, "s-a", RunStatus.COMPLETED, 0, pass_rate=1.0, latency=100) + _seed_child(report_session, campaign.id, "s-a", RunStatus.COMPLETED, 0, pass_rate=0.0, latency=200) + _seed_child(report_session, campaign.id, "s-b", RunStatus.COMPLETED, 6, pass_rate=0.5, latency=300) + _seed_child(report_session, campaign.id, "s-b", RunStatus.FAILED, 6) + + report = generate_campaign_report(campaign, RunRepository(report_session).list_by_campaign(campaign.id)) + + assert len(report["time_trend"]) == 12 + + b0 = _bucket(report, 0) + assert b0["run_count"] == 2 + assert b0["pass_rate"] == 0.5 # (1.0 + 0.0) / 2 + assert b0["availability"] == 1.0 + assert b0["avg_latency_ms"] == 150.0 # (100 + 200) / 2 + + b6 = _bucket(report, 6) + assert b6["run_count"] == 2 + assert b6["pass_rate"] == 0.25 # (0.5 + 0[failed]) / 2 — failure counts (ADR-0002) + assert b6["availability"] == 0.5 # 1 of 2 completed + assert b6["avg_latency_ms"] == 300.0 # failed run has no latency + + # empty bucket + assert _bucket(report, 3)["run_count"] == 0 + assert _bucket(report, 3)["pass_rate"] is None + + cap_a = _cap(report, "s-a") + assert cap_a["run_count"] == 2 and cap_a["pass_rate"] == 0.5 and cap_a["avg_latency_ms"] == 150.0 + cap_b = _cap(report, "s-b") + assert cap_b["run_count"] == 2 and cap_b["pass_rate"] == 0.25 and cap_b["availability"] == 0.5 + + s = report["summary"] + assert s["total_runs"] == 4 + assert s["completed_runs"] == 3 + assert s["overall_pass_rate"] == 0.375 # (1 + 0 + 0.5 + 0) / 4 + assert s["overall_availability"] == 0.75 + assert s["avg_latency_ms"] == 200.0 # (100 + 200 + 300) / 3 + + +def test_scenario_names_mapping(report_session): + campaign = _seed_campaign(report_session) + _seed_child(report_session, campaign.id, "s-a", RunStatus.COMPLETED, 0, pass_rate=1.0, latency=100) + report = generate_campaign_report( + campaign, RunRepository(report_session).list_by_campaign(campaign.id), + scenario_names={"s-a": "夜间问诊"}, + ) + assert _cap(report, "s-a")["scenario_name"] == "夜间问诊" + + +# ── boundaries: empty / partial ───────────────────────────────────────────── + +def test_empty_campaign_no_division_by_zero(report_session): + campaign = _seed_campaign(report_session) + report = generate_campaign_report(campaign, []) + assert len(report["time_trend"]) == 12 + assert all(b["run_count"] == 0 and b["pass_rate"] is None for b in report["time_trend"]) + assert report["capability_summary"] == [] + assert report["summary"]["total_runs"] == 0 + assert report["summary"]["overall_pass_rate"] is None + assert report["summary"]["avg_latency_ms"] is None + + +def test_only_in_flight_runs_partial_aggregate(report_session): + campaign = _seed_campaign(report_session) + _seed_child(report_session, campaign.id, "s-a", RunStatus.RUNNING, 0) + _seed_child(report_session, campaign.id, "s-a", RunStatus.PENDING, 0) + report = generate_campaign_report(campaign, RunRepository(report_session).list_by_campaign(campaign.id)) + s = report["summary"] + assert s["total_runs"] == 2 + assert s["completed_runs"] == 0 + assert s["overall_pass_rate"] == 0.0 # nothing completed yet, failures/incomplete count as 0 + assert s["overall_availability"] == 0.0 + assert s["avg_latency_ms"] is None + + +def test_time_scale_only_affects_bucketing_not_numbers(report_session): + # Compressed campaign: window 12s, scale 10 → a run 0.6s in maps to offset 6. + campaign = _seed_campaign(report_session, window_seconds=12, time_scale=10.0) + _seed_child(report_session, campaign.id, "s-a", RunStatus.COMPLETED, 0.6, pass_rate=0.8, latency=120) + report = generate_campaign_report(campaign, RunRepository(report_session).list_by_campaign(campaign.id)) + # Lands in bucket 6 (0.6s * 10 = 6.0), and the numbers are untouched by scale. + assert _bucket(report, 6)["run_count"] == 1 + assert _bucket(report, 6)["pass_rate"] == 0.8 + assert _bucket(report, 6)["avg_latency_ms"] == 120.0