refactor(metrics): extract dashboard aggregation to compute_dashboard

仪表盘聚合逻辑从 stats.py router 下沉到 metrics.py 的 compute_dashboard
纯函数。_settled 重命名为 settled_runs 并公开,_ts 重命名为 _sortable_ts。
router 从 40 行聚合逻辑缩到 5 行,只负责数据获取和序列化。

- 新增 compute_dashboard(runs, scenario_names, target_names) -> dict
- 新增 settled_runs(runs) 公开接口(原 _settled)
- trend 端点同步迁移到 settled_runs
- 5 个新测试覆盖 dashboard 聚合逻辑
This commit is contained in:
sinohqb 2026-08-04 11:36:55 +08:00
parent 42be31dd1f
commit c24998c762
3 changed files with 196 additions and 76 deletions

View File

@ -12,6 +12,8 @@ Rules (ADR-0004, extending ADR-0002's service perspective across runs):
- ``run_count`` still reports everything that happened, cancelled included. - ``run_count`` still reports everything that happened, cancelled included.
""" """
from collections import defaultdict
from datetime import datetime, timezone
from typing import Any from typing import Any
from agenteval.models import EvalRun, RunStatus from agenteval.models import EvalRun, RunStatus
@ -38,6 +40,87 @@ def aggregate_runs(runs: list[EvalRun]) -> dict[str, Any]:
} }
def settled_runs(runs: list[EvalRun]) -> list[EvalRun]:
"""Runs with an outcome — in-flight runs are not results yet.
Aggregation itself (fault=0.0, cancelled excluded) is ADR-0004's concern
and lives in ``aggregate_runs``; callers only choose *which* runs count.
"""
return [r for r in runs if r.status in (RunStatus.COMPLETED, RunStatus.FAILED)]
def _sortable_ts(dt: datetime | None) -> float:
"""Sortable timestamp tolerant of naive/aware mixes in legacy rows."""
if dt is None:
return 0.0
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.timestamp()
def compute_dashboard(
runs: list[EvalRun],
scenario_names: dict[str, str],
target_names: dict[str, str],
) -> dict[str, Any]:
"""Dashboard aggregation: counts, pass rate, per-scenario stats, recent runs.
Pure function no I/O. Caller fetches runs and name maps, passes them in.
"""
settled = settled_runs(runs)
overall_pass_rate = aggregate_runs(settled)["pass_rate"]
today = datetime.now(timezone.utc).date()
today_runs = 0
running_count = 0
trigger_breakdown: dict[str, int] = defaultdict(int)
for r in runs:
if r.started_at:
started = r.started_at
if started.tzinfo is None:
started = started.replace(tzinfo=timezone.utc)
if started.date() == today:
today_runs += 1
if r.status in (RunStatus.RUNNING, RunStatus.PENDING):
running_count += 1
trigger_breakdown[r.triggered_by.value] += 1
by_scenario: dict[str, list] = defaultdict(list)
for r in settled:
by_scenario[r.scenario_id].append(r)
scenario_stats = []
for sid, sruns in by_scenario.items():
agg = aggregate_runs(sruns)
last_run = max(sruns, key=lambda r: _sortable_ts(r.started_at))
scenario_stats.append({
"scenario_id": sid,
"scenario_name": scenario_names.get(sid, sid[:8]),
"run_count": agg["run_count"],
"avg_pass_rate": agg["pass_rate"],
"last_run_at": last_run.started_at.isoformat() if last_run.started_at else None,
})
scenario_stats.sort(key=lambda s: s["run_count"], reverse=True)
recent_runs = sorted(runs, key=lambda r: _sortable_ts(r.started_at), reverse=True)[:10]
return {
"runs_count": len(runs),
"today_runs": today_runs,
"running_count": running_count,
"overall_pass_rate": overall_pass_rate,
"trigger_breakdown": dict(trigger_breakdown),
"scenario_stats": scenario_stats,
"recent_runs": [
{
**r.model_dump(),
"scenario_name": scenario_names.get(r.scenario_id),
"target_name": target_names.get(r.target_id),
}
for r in recent_runs
],
}
def _completed_pass_rate(run: EvalRun) -> float: def _completed_pass_rate(run: EvalRun) -> float:
if run.summary is None or run.summary.pass_rate is None: if run.summary is None or run.summary.pass_rate is None:
return 0.0 return 0.0

View File

@ -1,13 +1,9 @@
"""API routes for statistics and dashboard data.""" """API routes for statistics and dashboard data."""
from collections import defaultdict
from datetime import datetime, timezone
from fastapi import APIRouter, Depends from fastapi import APIRouter, Depends
from sqlmodel import Session from sqlmodel import Session
from agenteval.evaluation.metrics import aggregate_runs from agenteval.evaluation.metrics import compute_dashboard, settled_runs
from agenteval.models import EvalRun, RunStatus
from agenteval.storage.model_config_repository import ModelConfigRepository from agenteval.storage.model_config_repository import ModelConfigRepository
from agenteval.storage.repository import RunRepository, ScenarioRepository, TargetRepository from agenteval.storage.repository import RunRepository, ScenarioRepository, TargetRepository
from agenteval.web.deps import get_db from agenteval.web.deps import get_db
@ -15,24 +11,6 @@ from agenteval.web.deps import get_db
router = APIRouter() router = APIRouter()
def _ts(dt: datetime | None) -> float:
"""Sortable timestamp tolerant of naive/aware mixes in legacy rows."""
if dt is None:
return 0.0
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.timestamp()
def _settled(runs: list[EvalRun]) -> list[EvalRun]:
"""Runs with an outcome — in-flight runs are not results yet.
Aggregation itself (fault=0.0, cancelled excluded) is ADR-0004's concern
and lives in ``aggregate_runs``; callers only choose *which* runs count.
"""
return [r for r in runs if r.status in (RunStatus.COMPLETED, RunStatus.FAILED)]
@router.get("/dashboard") @router.get("/dashboard")
def dashboard(session: Session = Depends(get_db)) -> dict: def dashboard(session: Session = Depends(get_db)) -> dict:
targets = TargetRepository(session).list_all() targets = TargetRepository(session).list_all()
@ -42,70 +20,25 @@ def dashboard(session: Session = Depends(get_db)) -> dict:
target_names = {t.id: t.name for t in targets} target_names = {t.id: t.name for t in targets}
settled_runs = _settled(runs) dashboard_data = compute_dashboard(runs, scenario_names, target_names)
overall_pass_rate = aggregate_runs(settled_runs)["pass_rate"]
today = datetime.now(timezone.utc).date()
today_runs = 0
running_count = 0
trigger_breakdown: dict[str, int] = defaultdict(int)
for r in runs:
if r.started_at:
started = r.started_at
if started.tzinfo is None:
started = started.replace(tzinfo=timezone.utc)
if started.date() == today:
today_runs += 1
if r.status in ("running", "pending"):
running_count += 1
trigger_breakdown[r.triggered_by.value] += 1
# Per-scenario aggregation over settled runs (ADR-0004 via aggregate_runs).
by_scenario: dict[str, list] = defaultdict(list)
for r in settled_runs:
by_scenario[r.scenario_id].append(r)
scenario_stats = []
for sid, sruns in by_scenario.items():
agg = aggregate_runs(sruns)
last_run = max(sruns, key=lambda r: _ts(r.started_at))
scenario_stats.append({
"scenario_id": sid,
"scenario_name": scenario_names.get(sid, sid[:8]),
"run_count": agg["run_count"],
"avg_pass_rate": agg["pass_rate"],
"last_run_at": last_run.started_at.isoformat() if last_run.started_at else None,
})
scenario_stats.sort(key=lambda s: s["run_count"], reverse=True)
recent_runs = sorted(runs, key=lambda r: _ts(r.started_at), reverse=True)[:10]
return { return {
"targets_count": len(targets), "targets_count": len(targets),
"scenarios_count": len(scenario_names), "scenarios_count": len(scenario_names),
"runs_count": len(runs),
"model_configs_count": len(model_configs), "model_configs_count": len(model_configs),
"today_runs": today_runs, **dashboard_data,
"running_count": running_count,
"overall_pass_rate": overall_pass_rate,
"trigger_breakdown": dict(trigger_breakdown),
"scenario_stats": scenario_stats,
"recent_runs": [
{
**r.model_dump(),
"scenario_name": scenario_names.get(r.scenario_id),
"target_name": target_names.get(r.target_id),
}
for r in recent_runs
],
} }
@router.get("/trend") @router.get("/trend")
def trend(days: int = 30, session: Session = Depends(get_db)) -> list[dict]: def trend(days: int = 30, session: Session = Depends(get_db)) -> list[dict]:
from collections import defaultdict
from agenteval.evaluation.metrics import aggregate_runs
runs = RunRepository(session).list_all() runs = RunRepository(session).list_all()
daily: dict[str, list[EvalRun]] = defaultdict(list) daily: dict[str, list] = defaultdict(list)
for run in _settled(runs): for run in settled_runs(runs):
if run.started_at: if run.started_at:
daily[run.started_at.strftime("%Y-%m-%d")].append(run) daily[run.started_at.strftime("%Y-%m-%d")].append(run)

View File

@ -0,0 +1,104 @@
"""Tests for compute_dashboard — the dashboard aggregation logic."""
from datetime import datetime, timedelta, timezone
from agenteval.evaluation.metrics import compute_dashboard, settled_runs
from agenteval.models import EvalRun, RunStatus, RunTrigger
def _make_run(
run_id: str,
status: RunStatus = RunStatus.COMPLETED,
pass_rate: float = 1.0,
started_at: datetime | None = None,
scenario_id: str = "s1",
triggered_by: RunTrigger = RunTrigger.MANUAL,
) -> EvalRun:
return EvalRun(
id=run_id,
target_id="t1",
scenario_id=scenario_id,
status=status,
triggered_by=triggered_by,
started_at=started_at or datetime.now(timezone.utc),
completed_at=datetime.now(timezone.utc),
summary={"pass_rate": pass_rate, "total_cases": 1, "passed_cases": int(pass_rate)},
)
def test_settled_filters_to_completed_and_failed():
"""settled_runs excludes running/pending/cancelled."""
runs = [
_make_run("r1", RunStatus.COMPLETED),
_make_run("r2", RunStatus.FAILED),
_make_run("r3", RunStatus.RUNNING),
_make_run("r4", RunStatus.PENDING),
]
settled = settled_runs(runs)
assert len(settled) == 2
assert {r.id for r in settled} == {"r1", "r2"}
def test_dashboard_counts_and_pass_rate():
"""Dashboard returns correct counts and overall pass rate."""
now = datetime.now(timezone.utc)
runs = [
_make_run("r1", pass_rate=1.0, started_at=now),
_make_run("r2", pass_rate=0.5, started_at=now),
_make_run("r3", RunStatus.RUNNING, started_at=now),
]
result = compute_dashboard(runs, scenario_names={"s1": "Scenario 1"}, target_names={"t1": "Target 1"})
assert result["runs_count"] == 3
assert result["running_count"] == 1
assert result["today_runs"] == 3 # all runs started today (including running)
assert result["overall_pass_rate"] == 0.75 # (1.0 + 0.5) / 2
def test_dashboard_trigger_breakdown():
"""Dashboard groups runs by trigger source."""
now = datetime.now(timezone.utc)
runs = [
_make_run("r1", triggered_by=RunTrigger.MANUAL, started_at=now),
_make_run("r2", triggered_by=RunTrigger.MANUAL, started_at=now),
_make_run("r3", triggered_by=RunTrigger.CAMPAIGN, started_at=now),
]
result = compute_dashboard(runs, scenario_names={}, target_names={})
assert result["trigger_breakdown"] == {"manual": 2, "campaign": 1}
def test_dashboard_scenario_stats():
"""Dashboard aggregates per-scenario stats."""
now = datetime.now(timezone.utc)
runs = [
_make_run("r1", scenario_id="s1", pass_rate=1.0, started_at=now),
_make_run("r2", scenario_id="s1", pass_rate=0.5, started_at=now - timedelta(hours=1)),
_make_run("r3", scenario_id="s2", pass_rate=0.8, started_at=now),
]
result = compute_dashboard(
runs,
scenario_names={"s1": "Scenario 1", "s2": "Scenario 2"},
target_names={},
)
stats = result["scenario_stats"]
assert len(stats) == 2
# s1 has 2 runs, s2 has 1 run — sorted by run_count desc
assert stats[0]["scenario_id"] == "s1"
assert stats[0]["run_count"] == 2
assert stats[0]["avg_pass_rate"] == 0.75
assert stats[1]["scenario_id"] == "s2"
assert stats[1]["run_count"] == 1
def test_dashboard_recent_runs_sorted_and_limited():
"""Dashboard returns 10 most recent runs, sorted by started_at desc."""
now = datetime.now(timezone.utc)
runs = [
_make_run(f"r{i}", started_at=now - timedelta(hours=i))
for i in range(15)
]
result = compute_dashboard(runs, scenario_names={}, target_names={})
recent = result["recent_runs"]
assert len(recent) == 10
# Most recent first
assert recent[0]["id"] == "r0"
assert recent[9]["id"] == "r9"