Compare commits

...

5 Commits

Author SHA1 Message Date
sinohqb
aa40c8e0d8 refactor(campaign): type Campaign.summary as CampaignSummary VO
Some checks failed
CI / test (push) Failing after 12s
Campaign.summary was a bare Optional[dict] while RunSummary is a typed VO —
scheduler state (spawned_indices/errors) flowed untyped through
set_/get_summary. Introduce CampaignSummary + SchedulerState (extra=allow,
validate_assignment), mirroring RunSummary; campaign_runner reads/writes the
VO. Also converge the ~10 repeated JSON column get/set pairs onto
_json_dumps/_json_loads helpers, unifying ensure_ascii=False and fixing the
set_modalities ensure_ascii=True trap.
2026-07-31 15:09:10 +08:00
sinohqb
ffc058f951 refactor(engine): thin _run_case and _save_rule_results
Three near-identical Turn(...)+save_turn blocks (send-fail / poll-except /
happy path) collapse into one _persist_turn helper differing only by the
optional fields set. The期望→隐式规则 translation becomes a pure,
unit-testable derive_implicit_rules seam (CONTEXT: 期望与规则叠加生效),
and the rule_type→ModelPurpose map is hoisted to a module constant.
2026-07-31 14:59:50 +08:00
sinohqb
d411572607 refactor(storage): converge CRUD skeleton into BaseRepository
Target/Scenario/Run/Campaign repositories repeated the same
__init__/list_all/get/create/delete skeleton (~120 lines). Extract a
generic BaseRepository[M, DB]: subclasses declare the table + ordering
column and implement instance-method _to_db/_from_db converters (so
Scenario's _from_db can reach self.session for model bindings). Bespoke
paths (Scenario create/delete, all update) stay per-subclass. Adds
test_repository.py locking the CRUD + JSON round-trip contract.
2026-07-31 14:30:11 +08:00
sinohqb
9c01afa79b refactor(engine): extract build_run_summary pure seam
Single-run summary口径 (pass_rate / judged_pass_rate / avg_latency /
connectivity split) was inlined in run(), reachable only by driving a
whole async run, and report.py recomputed judged_pass_rate independently.
Extract build_run_summary — a pure function parallel to aggregate_runs
(cross-run) and combine_case_outcome (case-level). run() now collects
material and delegates; judged_pass_rate is stored in RunSummary so the
report reads it instead of recomputing.
2026-07-31 14:20:51 +08:00
sinohqb
983a58d013 refactor(verdict): unify read path on authoritative case_outcomes
Read paths recomputed per-case pass/connectivity independently — report
generation, the logs endpoint, and the frontend each derived it, and the
frontend's every(passed) recompute ignored the engine's authoritative
verdict. Extract resolve_case_verdicts: a single pure seam that prefers
stored case_outcomes verbatim and approximates only for legacy runs. The
logs endpoint now surfaces case_verdicts so the frontend reads instead of
recomputing.
2026-07-31 14:11:58 +08:00
17 changed files with 988 additions and 350 deletions

View File

@ -29,7 +29,15 @@ from agenteval.evaluation.campaign_scheduler import (
resolve_finalize, resolve_finalize,
) )
from agenteval.evaluation.engine import EvalEngine from agenteval.evaluation.engine import EvalEngine
from agenteval.models import Campaign, CampaignStatus, EvalRun, RunStatus, RunTrigger from agenteval.models import (
Campaign,
CampaignStatus,
CampaignSummary,
EvalRun,
RunStatus,
RunTrigger,
SchedulerState,
)
from agenteval.storage.db import get_session, utc_now from agenteval.storage.db import get_session, utc_now
from agenteval.storage.repository import ( from agenteval.storage.repository import (
CampaignRepository, CampaignRepository,
@ -39,9 +47,6 @@ from agenteval.storage.repository import (
) )
from agenteval.task_registry import TaskRegistry from agenteval.task_registry import TaskRegistry
_SPAWNED_KEY = "spawned_indices"
_ERRORS_KEY = "errors"
# Real wall-clock seconds between scheduler ticks. time_scale compresses the # Real wall-clock seconds between scheduler ticks. time_scale compresses the
# *window*, not the tick cadence — production 24h campaigns still tick slowly. # *window*, not the tick cadence — production 24h campaigns still tick slowly.
DEFAULT_TICK_SECONDS = 1.0 DEFAULT_TICK_SECONDS = 1.0
@ -62,8 +67,9 @@ class AdvanceResult:
def _spawned_indices(campaign: Campaign) -> set[int]: def _spawned_indices(campaign: Campaign) -> set[int]:
scheduler = (campaign.summary or {}).get("scheduler", {}) if campaign.summary is None:
return set(scheduler.get(_SPAWNED_KEY, [])) return set()
return set(campaign.summary.scheduler.spawned_indices)
def current_window_offset(campaign: Campaign) -> float: def current_window_offset(campaign: Campaign) -> float:
@ -152,7 +158,7 @@ async def advance_campaign(
) )
result = AdvanceResult(finished=decision.finished) result = AdvanceResult(finished=decision.finished)
errors: list[dict] = list((campaign.summary or {}).get("scheduler", {}).get(_ERRORS_KEY, [])) errors: list[dict] = list(campaign.summary.scheduler.errors) if campaign.summary else []
for due in decision.due: for due in decision.due:
# Cancellation stops further spawning; runs already in flight finish. # Cancellation stops further spawning; runs already in flight finish.
if cancel_event is not None and cancel_event.is_set(): if cancel_event is not None and cancel_event.is_set():
@ -167,12 +173,10 @@ async def advance_campaign(
spawned.add(due.index) spawned.add(due.index)
# Persist progress per entry: a failure partway through a multi-entry # Persist progress per entry: a failure partway through a multi-entry
# advance must never lose which entries already spawned, since restart # advance must never lose which entries already spawned, since restart
# recovery reads this back from the DB. # recovery reads this back from the DB. Mutate the existing summary so
scheduler_state: dict = {_SPAWNED_KEY: sorted(spawned)} # any unknown top-level keys survive the read-modify-write.
if errors: summary = campaign.summary or CampaignSummary()
scheduler_state[_ERRORS_KEY] = errors summary.scheduler = SchedulerState(spawned_indices=sorted(spawned), errors=errors)
summary = dict(campaign.summary or {})
summary["scheduler"] = scheduler_state
campaign.summary = summary campaign.summary = summary
repo.update(campaign) repo.update(campaign)

View File

@ -0,0 +1,67 @@
"""Case-verdict read seam — the single place the read path derives pass/connectivity.
The engine is the authority: it runs ``judgement.combine_case_outcome`` once and
writes each case's verdict into ``summary.case_outcomes``. Every read surface
(report generation, the run-logs endpoint) must present *that* verdict, never
recompute it otherwise WEIGHTED/ANY logic and connectivity cases diverge from
what was judged.
This module is that single seam. It reads the authoritative ``case_outcomes``
when present, and only for older runs that predate it falls back to a documented
approximation from persisted turns/results. Pure no I/O; callers build the
per-case ``CaseEvidence`` from whatever they already have in hand.
"""
from dataclasses import dataclass
from agenteval.models import CaseOutcomeSummary
@dataclass(frozen=True)
class CaseEvidence:
"""What the legacy approximation needs about one case's persisted record.
``result_passes`` is the per-rule pass flags (empty means no judged rule
result exists for the case the connectivity-vs-fault fork).
"""
has_turns: bool
all_replied: bool
result_passes: tuple[bool, ...] = ()
def resolve_case_verdicts(
*,
case_outcomes: dict[str, CaseOutcomeSummary],
evidence: dict[str, CaseEvidence],
errored_case_ids: set[str],
) -> dict[str, CaseOutcomeSummary]:
"""Resolve every case in ``evidence`` to its authoritative-or-approximated verdict.
Authoritative ``case_outcomes`` win verbatim. For a case missing from it (an
older run), approximate per CONTEXT.md / ADR-0002: a case with no judged
results but turns that all replied and no case-level error is a *connectivity*
case (counts as passed); a case with results passes iff every rule passed;
anything else (a fault) fails.
"""
verdicts: dict[str, CaseOutcomeSummary] = {}
for case_id, ev in evidence.items():
authoritative = case_outcomes.get(case_id)
if authoritative is not None:
verdicts[case_id] = authoritative
continue
connectivity = (
not ev.result_passes
and ev.has_turns
and ev.all_replied
and case_id not in errored_case_ids
)
if connectivity:
passed = True
elif not ev.result_passes:
passed = False
else:
passed = all(ev.result_passes)
verdicts[case_id] = CaseOutcomeSummary(passed=passed, connectivity=connectivity)
return verdicts

View File

@ -8,13 +8,16 @@ via an ``asyncio.Event`` cancel token.
import asyncio import asyncio
import uuid import uuid
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime
from typing import Any, Callable, Optional from typing import Any, Callable, Optional
from agenteval.channels.base import EvalChannel from agenteval.channels.base import EvalChannel
from agenteval.channels.factory import ChannelFactory from agenteval.channels.factory import ChannelFactory
from agenteval.config import get_settings from agenteval.config import get_settings
from agenteval.evaluation.implicit_rules import derive_implicit_rules
from agenteval.evaluation.judgement import CaseOutcome, RuleOutcome, combine_case_outcome from agenteval.evaluation.judgement import CaseOutcome, RuleOutcome, combine_case_outcome
from agenteval.evaluation.rules import RuleResult, get_rule from agenteval.evaluation.rules import RuleResult, get_rule
from agenteval.evaluation.run_summary import build_run_summary
from agenteval.model_gateway import ModelGateway from agenteval.model_gateway import ModelGateway
from agenteval.models import ( from agenteval.models import (
Case, Case,
@ -58,6 +61,14 @@ def _build_send_message(content: str) -> dict[str, Any]:
} }
# 需要模型资源的规则类型 → 评测岗位ModelPurpose其余规则无需模型
RULE_PURPOSE = {
"llm_score": ModelPurpose.JUDGE,
"semantic_similarity": ModelPurpose.EMBEDDING,
"safety": ModelPurpose.MODERATION,
}
class EvalEngine: class EvalEngine:
"""Execute evaluation scenarios against targets. """Execute evaluation scenarios against targets.
@ -130,9 +141,7 @@ class EvalEngine:
try: try:
total_cases = len(self.scenario.cases) total_cases = len(self.scenario.cases)
passed_cases = 0 case_outcomes: dict[str, CaseOutcome] = {}
failed_cases = 0
case_outcomes: dict[str, dict[str, bool]] = {}
for idx, case in enumerate(self.scenario.cases, start=1): for idx, case in enumerate(self.scenario.cases, start=1):
self._check_cancel() self._check_cancel()
@ -151,11 +160,7 @@ class EvalEngine:
case, case,
progress_callback, progress_callback,
) )
case_outcomes[case.id] = {"passed": outcome.passed, "connectivity": outcome.connectivity} case_outcomes[case.id] = outcome
if outcome.passed:
passed_cases += 1
else:
failed_cases += 1
await self._emit( await self._emit(
progress_callback, progress_callback,
"case_end", "case_end",
@ -170,34 +175,18 @@ class EvalEngine:
) )
results = self.run_repo.get_results(run.id) results = self.run_repo.get_results(run.id)
total_rules = len(results)
passed_rules = sum(1 for r in results if r.passed)
turns = self.run_repo.get_turns(run.id) turns = self.run_repo.get_turns(run.id)
latencies = [t.latency_ms for t in turns if t.latency_ms is not None] summary = build_run_summary(
avg_latency_ms = round(sum(latencies) / len(latencies), 1) if latencies else None case_outcomes=case_outcomes,
latencies=[t.latency_ms for t in turns if t.latency_ms is not None],
summary = { rule_passes=[r.passed for r in results],
"total_cases": total_cases, case_errors=self._case_errors or None,
"passed_cases": passed_cases, model_configs=(
"failed_cases": failed_cases, {purpose.value: config.snapshot() for purpose, config in self._resolved_models.items()}
"total_rules": total_rules, if self._resolved_models
"passed_rules": passed_rules, else None
# 通过率是用例级口径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,
}
# Surface fatal case-level errors (e.g. dynamic generation failures)
# so the report / DB record shows *why* a run produced no results.
if self._case_errors:
summary["case_errors"] = self._case_errors
if self._resolved_models:
summary["model_configs"] = {
purpose.value: config.snapshot() for purpose, config in self._resolved_models.items()
}
run.status = RunStatus.COMPLETED run.status = RunStatus.COMPLETED
run.completed_at = utc_now() run.completed_at = utc_now()
run.summary = summary run.summary = summary
@ -206,7 +195,7 @@ class EvalEngine:
"run_completed", "run_completed",
{ {
"status": "completed", "status": "completed",
"summary": summary, "summary": summary.model_dump(),
}, },
) )
except CancelledError: except CancelledError:
@ -258,6 +247,36 @@ class EvalEngine:
# ── case / turn execution ───────────────────────────────────────── # ── case / turn execution ─────────────────────────────────────────
def _persist_turn(
self,
run: EvalRun,
case: Case,
round_index: int,
message: str,
sent_at: datetime,
*,
question_msg_id: Optional[str] = None,
reply: Optional[dict] = None,
received_at: Optional[datetime] = None,
latency_ms: Optional[int] = None,
) -> Turn:
"""Build, persist, and return a Turn. The three call sites (send-fail /
poll-except / happy path) differ only in which optional fields are set."""
turn = Turn(
id=str(uuid.uuid4()),
run_id=run.id,
case_id=case.id,
round_index=round_index,
sent_message=_build_send_message(message),
sent_at=sent_at,
question_msg_id=question_msg_id,
reply=reply,
received_at=received_at,
latency_ms=latency_ms,
)
self.result_repo.save_turn(turn)
return turn
async def _run_case( async def _run_case(
self, self,
run: EvalRun, run: EvalRun,
@ -298,15 +317,7 @@ class EvalEngine:
sent_at = utc_now() sent_at = utc_now()
send_result = await self.channel.send(message) send_result = await self.channel.send(message)
if not send_result.ok: if not send_result.ok:
turn = Turn( turn = self._persist_turn(run, case, round_index, message, sent_at)
id=str(uuid.uuid4()),
run_id=run.id,
case_id=case.id,
round_index=round_index,
sent_message=_build_send_message(message),
sent_at=sent_at,
)
self.result_repo.save_turn(turn)
await self._save_rule_results(run, case, turn, [], progress_callback) await self._save_rule_results(run, case, turn, [], progress_callback)
await self._emit( await self._emit(
progress_callback, progress_callback,
@ -326,17 +337,15 @@ class EvalEngine:
) )
except Exception as poll_exc: except Exception as poll_exc:
received_at = utc_now() received_at = utc_now()
turn = Turn( turn = self._persist_turn(
id=str(uuid.uuid4()), run,
run_id=run.id, case,
case_id=case.id, round_index,
round_index=round_index, message,
sent_message=_build_send_message(message), sent_at,
sent_at=sent_at,
question_msg_id=send_result.question_msg_id, question_msg_id=send_result.question_msg_id,
received_at=received_at, received_at=received_at,
) )
self.result_repo.save_turn(turn)
await self._emit( await self._emit(
progress_callback, progress_callback,
"turn_error", "turn_error",
@ -353,19 +362,17 @@ class EvalEngine:
if sent_at and received_at: if sent_at and received_at:
latency_ms = int((received_at - sent_at).total_seconds() * 1000) latency_ms = int((received_at - sent_at).total_seconds() * 1000)
turn = Turn( turn = self._persist_turn(
id=str(uuid.uuid4()), run,
run_id=run.id, case,
case_id=case.id, round_index,
round_index=round_index, message,
sent_message=_build_send_message(message), sent_at,
sent_at=sent_at,
question_msg_id=send_result.question_msg_id, question_msg_id=send_result.question_msg_id,
reply=reply.raw_message if reply else None, reply=reply.raw_message if reply else None,
received_at=received_at, received_at=received_at,
latency_ms=latency_ms, latency_ms=latency_ms,
) )
self.result_repo.save_turn(turn)
dialog.append(turn) dialog.append(turn)
await self._emit( await self._emit(
@ -401,27 +408,7 @@ class EvalEngine:
from agenteval.models import EvalRuleConfig from agenteval.models import EvalRuleConfig
rules_config: list[EvalRuleConfig] = list(case.eval_rules) rules_config: list[EvalRuleConfig] = list(case.eval_rules)
implicit_config = derive_implicit_rules(case.expectations)
implicit_config: list[EvalRuleConfig] = []
if case.expectations.response_time_max_ms:
implicit_config.append(
EvalRuleConfig(
type="response_time",
params={
"max_ms": case.expectations.response_time_max_ms,
},
)
)
if case.expectations.keywords_include or case.expectations.keywords_exclude:
implicit_config.append(
EvalRuleConfig(
type="keyword_match",
params={
"keywords": case.expectations.keywords_include,
"exclude_keywords": case.expectations.keywords_exclude,
},
)
)
all_replied = bool(dialog) and all(t.reply is not None for t in dialog) all_replied = bool(dialog) and all(t.reply is not None for t in dialog)
@ -436,11 +423,7 @@ class EvalEngine:
all_rules = [(cfg, False) for cfg in rules_config] + [(cfg, True) for cfg in implicit_config] all_rules = [(cfg, False) for cfg in rules_config] + [(cfg, True) for cfg in implicit_config]
for rule_config, is_implicit in all_rules: for rule_config, is_implicit in all_rules:
purpose = { purpose = RULE_PURPOSE.get(rule_config.type)
"llm_score": ModelPurpose.JUDGE,
"semantic_similarity": ModelPurpose.EMBEDDING,
"safety": ModelPurpose.MODERATION,
}.get(rule_config.type)
try: try:
model_config = self._resolve_model(purpose) if purpose else None model_config = self._resolve_model(purpose) if purpose else None
rule = get_rule( rule = get_rule(

View File

@ -0,0 +1,30 @@
"""Translate a case's期望 (Expectation) into implicit评估规则.
期望描述"想要什么"评估规则是"怎么判定"期望派生的隐式规则与显式规则
叠加生效CONTEXT.md此翻译是纯逻辑独立于规则执行与持久化便于单测
"""
from agenteval.models import EvalRuleConfig, Expectation
def derive_implicit_rules(expectation: Expectation) -> list[EvalRuleConfig]:
"""Build the implicit rule configs a case's expectation implies."""
rules: list[EvalRuleConfig] = []
if expectation.response_time_max_ms:
rules.append(
EvalRuleConfig(
type="response_time",
params={"max_ms": expectation.response_time_max_ms},
)
)
if expectation.keywords_include or expectation.keywords_exclude:
rules.append(
EvalRuleConfig(
type="keyword_match",
params={
"keywords": expectation.keywords_include,
"exclude_keywords": expectation.keywords_exclude,
},
)
)
return rules

View File

@ -9,6 +9,7 @@ from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
from typing import Any, Optional from typing import Any, Optional
from agenteval.evaluation.case_verdict import CaseEvidence, resolve_case_verdicts
from agenteval.evaluation.metrics import aggregate_runs from agenteval.evaluation.metrics import aggregate_runs
from agenteval.evaluation.report_render import render_html, render_json, render_markdown from agenteval.evaluation.report_render import render_html, render_json, render_markdown
from agenteval.models import Campaign, EvalRun, RunStatus, RunSummary from agenteval.models import Campaign, EvalRun, RunStatus, RunSummary
@ -68,37 +69,31 @@ def generate_report(run_id: str, session=None) -> dict[str, Any]:
summary = run.summary or RunSummary() summary = run.summary or RunSummary()
errored_case_ids = {e.get("case_id") for e in summary.case_errors} errored_case_ids = {e.get("case_id") for e in summary.case_errors}
# 权威判定:引擎经 judgement.combine_case_outcome 算一次写入 summary # 权威判定:引擎经 combine_case_outcome 算一次写入 summary读路径只读不重算。
# 老 run 没有该字段时退回从持久化结果反推WEIGHTED/ANY 只能近似)。 # resolve_case_verdicts 统一处理「权威优先、老 run 近似回退」(唯一落点)。
authoritative = summary.case_outcomes evidence = {
case_id: CaseEvidence(
has_turns=bool(item["turns"]),
all_replied=item["all_replied"],
result_passes=tuple(r["passed"] for r in item["results"]),
)
for case_id, item in case_map.items()
}
verdicts = resolve_case_verdicts(
case_outcomes=summary.case_outcomes,
evidence=evidence,
errored_case_ids=errored_case_ids,
)
cases = [] cases = []
for case_id in sorted(case_map.keys()): for case_id in sorted(case_map.keys()):
item = case_map[case_id] item = case_map[case_id]
if case_id in authoritative: verdict = verdicts[case_id]
outcome = authoritative[case_id]
connectivity = outcome.connectivity
passed = outcome.passed
else:
# 连通用例无任何判定结果且每轮都收到回复、无用例级错误CONTEXT.md
connectivity = (
not item["results"]
and bool(item["turns"])
and item["all_replied"]
and case_id not in errored_case_ids
)
if connectivity:
passed = True
elif not item["results"]:
# 故障用例无结果且非连通不通过ADR-0002
passed = False
else:
passed = all(r["passed"] for r in item["results"])
cases.append( cases.append(
{ {
"case_id": case_id, "case_id": case_id,
"passed": passed, "passed": verdict.passed,
"connectivity": connectivity, "connectivity": verdict.connectivity,
"turns": sorted(item["turns"], key=lambda x: x["round"]), "turns": sorted(item["turns"], key=lambda x: x["round"]),
"results": item["results"], "results": item["results"],
} }
@ -108,8 +103,10 @@ def generate_report(run_id: str, session=None) -> dict[str, Any]:
passed_cases = summary.passed_cases passed_cases = summary.passed_cases
connectivity_count = sum(1 for c in cases if c["connectivity"]) connectivity_count = sum(1 for c in cases if c["connectivity"])
judged_total = total_cases - connectivity_count judged_total = total_cases - connectivity_count
# 连通用例按引擎口径计通过,判定型通过数 = 总通过数 - 连通用例数 # 判定型通过率由 build_run_summary 入库,读路径只读;老 run 缺字段时按同一口径回退近似
judged_pass_rate = round((passed_cases - connectivity_count) / judged_total, 4) if judged_total > 0 else None 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 { return {
"run_id": run.id, "run_id": run.id,

View File

@ -0,0 +1,53 @@
"""Single-run summary aggregation — the pure落点 for one run's口径.
Parallel to ``metrics.aggregate_runs`` (cross-run) and
``judgement.combine_case_outcome`` (case-level): given the authoritative
per-case outcomes plus raw latency/rule material, compute the run's
summary口径 once. No IO the engine collects material and calls this; DB
writes and event emits stay in the caller. See CONTEXT.md (通过率) / ADR-0002.
"""
from typing import Any, Optional, Sequence
from agenteval.evaluation.judgement import CaseOutcome
from agenteval.models import CaseOutcomeSummary, RunSummary
def build_run_summary(
*,
case_outcomes: dict[str, CaseOutcome],
latencies: Sequence[float],
rule_passes: Sequence[bool],
case_errors: Optional[list[dict[str, str]]] = None,
model_configs: Optional[dict[str, Any]] = None,
) -> RunSummary:
"""Compute a run's summary口径 from its authoritative case outcomes."""
total_cases = len(case_outcomes)
passed_cases = sum(1 for o in case_outcomes.values() if o.passed)
connectivity_count = sum(1 for o in case_outcomes.values() if o.connectivity)
pass_rate = round(passed_cases / total_cases, 4) if total_cases else 0.0
# 连通用例按引擎口径计通过,判定型通过数 = 总通过数 - 连通用例数
judged_total = total_cases - connectivity_count
judged_pass_rate = (
round((passed_cases - connectivity_count) / judged_total, 4) if judged_total > 0 else None
)
avg_latency_ms = round(sum(latencies) / len(latencies), 1) if latencies else None
return RunSummary(
total_cases=total_cases,
passed_cases=passed_cases,
failed_cases=total_cases - passed_cases,
total_rules=len(rule_passes),
passed_rules=sum(1 for p in rule_passes if p),
pass_rate=pass_rate,
judged_pass_rate=judged_pass_rate,
avg_latency_ms=avg_latency_ms,
case_outcomes={
case_id: CaseOutcomeSummary(passed=o.passed, connectivity=o.connectivity)
for case_id, o in case_outcomes.items()
},
case_errors=case_errors or [],
model_configs=model_configs or {},
)

View File

@ -187,6 +187,8 @@ class RunSummary(BaseModel):
passed_rules: int = 0 passed_rules: int = 0
# 用例级通过率含执行失败ADR-0002失败/取消的 run 无此值 # 用例级通过率含执行失败ADR-0002失败/取消的 run 无此值
pass_rate: Optional[float] = None pass_rate: Optional[float] = None
# 判定型通过率:连通用例从分子分母双双剔除;无判定型用例时为空
judged_pass_rate: Optional[float] = None
avg_latency_ms: Optional[float] = None avg_latency_ms: Optional[float] = None
case_outcomes: dict[str, CaseOutcomeSummary] = Field(default_factory=dict) case_outcomes: dict[str, CaseOutcomeSummary] = Field(default_factory=dict)
case_errors: list[dict[str, str]] = Field(default_factory=list) case_errors: list[dict[str, str]] = Field(default_factory=list)
@ -244,10 +246,37 @@ class CampaignPlanEntry(BaseModel):
count: int = Field(default=1, ge=1) count: int = Field(default=1, ge=1)
class SchedulerState(BaseModel):
"""Durable scheduler progress for a campaign — restart-safe (ADR-0003).
``spawned_indices`` are the plan entries already派生 into child Runs;
``errors`` records entries whose spawn failed (marked spawned to avoid
infinite retry).
"""
spawned_indices: list[int] = Field(default_factory=list)
errors: list[dict[str, Any]] = Field(default_factory=list)
class CampaignSummary(BaseModel):
"""Typed value of ``Campaign.summary`` — mirrors RunSummary's treatment.
Unknown top-level keys are preserved (extra=allow) so summaries written by
older versions keep parsing and survive read-modify-write.
"""
model_config = {"extra": "allow"}
scheduler: SchedulerState = Field(default_factory=SchedulerState)
class Campaign(BaseModel): class Campaign(BaseModel):
"""An evaluation campaign: a service-cycle window over a single target, """An evaluation campaign: a service-cycle window over a single target,
driving many child Runs from a static plan (ADR-0003).""" driving many child Runs from a static plan (ADR-0003)."""
# summary 以属性赋值写入scheduler loop赋值时即校验成 CampaignSummary
model_config = {"validate_assignment": True}
id: Optional[str] = None id: Optional[str] = None
name: str name: str
target_id: str target_id: str
@ -258,7 +287,7 @@ class Campaign(BaseModel):
started_at: Optional[datetime] = None started_at: Optional[datetime] = None
completed_at: Optional[datetime] = None completed_at: Optional[datetime] = None
created_at: Optional[datetime] = None created_at: Optional[datetime] = None
summary: Optional[dict[str, Any]] = None summary: Optional[CampaignSummary] = None
class Turn(BaseModel): class Turn(BaseModel):

View File

@ -44,6 +44,16 @@ def new_uuid() -> str:
return str(uuid.uuid4()) return str(uuid.uuid4())
def _json_dumps(value: Any) -> str:
"""Serialize a JSON column value. ensure_ascii=False keeps CJK readable
in the stored text the single serialization口径 for all JSON columns."""
return json.dumps(value, ensure_ascii=False)
def _json_loads(raw: str) -> Any:
return json.loads(raw)
class EvalTargetDB(SQLModel, table=True): class EvalTargetDB(SQLModel, table=True):
"""Database table for evaluation targets.""" """Database table for evaluation targets."""
@ -65,10 +75,10 @@ class EvalTargetDB(SQLModel, table=True):
) )
def get_config(self) -> dict[str, Any]: def get_config(self) -> dict[str, Any]:
return json.loads(self.channel_config) return _json_loads(self.channel_config)
def set_config(self, config: dict[str, Any]) -> None: def set_config(self, config: dict[str, Any]) -> None:
self.channel_config = json.dumps(config, ensure_ascii=False) self.channel_config = _json_dumps(config)
class ScenarioDB(SQLModel, table=True): class ScenarioDB(SQLModel, table=True):
@ -92,22 +102,22 @@ class ScenarioDB(SQLModel, table=True):
) )
def get_tags(self) -> list[str]: def get_tags(self) -> list[str]:
return json.loads(self.tags) return _json_loads(self.tags)
def set_tags(self, tags: list[str]) -> None: def set_tags(self, tags: list[str]) -> None:
self.tags = json.dumps(tags, ensure_ascii=False) self.tags = _json_dumps(tags)
def get_cases(self) -> list[dict[str, Any]]: def get_cases(self) -> list[dict[str, Any]]:
return json.loads(self.cases) return _json_loads(self.cases)
def set_cases(self, cases: list[dict[str, Any]]) -> None: def set_cases(self, cases: list[dict[str, Any]]) -> None:
self.cases = json.dumps(cases, ensure_ascii=False) self.cases = _json_dumps(cases)
def get_llm_config(self) -> Optional[dict[str, Any]]: def get_llm_config(self) -> Optional[dict[str, Any]]:
return json.loads(self.llm_config) if self.llm_config else None return _json_loads(self.llm_config) if self.llm_config else None
def set_llm_config(self, config: Optional[dict[str, Any]]) -> None: def set_llm_config(self, config: Optional[dict[str, Any]]) -> None:
self.llm_config = json.dumps(config, ensure_ascii=False) if config else None self.llm_config = _json_dumps(config) if config else None
class ModelConfigDB(SQLModel, table=True): class ModelConfigDB(SQLModel, table=True):
@ -140,14 +150,14 @@ class ModelConfigDB(SQLModel, table=True):
updated_at: Optional[datetime] = Field(default_factory=utc_now) updated_at: Optional[datetime] = Field(default_factory=utc_now)
def get_input_modalities(self) -> list[str]: def get_input_modalities(self) -> list[str]:
return json.loads(self.input_modalities) return _json_loads(self.input_modalities)
def get_output_modalities(self) -> list[str]: def get_output_modalities(self) -> list[str]:
return json.loads(self.output_modalities) return _json_loads(self.output_modalities)
def set_modalities(self, input_modalities: list[str], output_modalities: list[str]) -> None: def set_modalities(self, input_modalities: list[str], output_modalities: list[str]) -> None:
self.input_modalities = json.dumps(input_modalities, ensure_ascii=True) self.input_modalities = _json_dumps(input_modalities)
self.output_modalities = json.dumps(output_modalities, ensure_ascii=True) self.output_modalities = _json_dumps(output_modalities)
class ScenarioModelBindingDB(SQLModel, table=True): class ScenarioModelBindingDB(SQLModel, table=True):
@ -178,16 +188,16 @@ class CampaignDB(SQLModel, table=True):
summary: Optional[str] = None summary: Optional[str] = None
def get_plan(self) -> list[dict[str, Any]]: def get_plan(self) -> list[dict[str, Any]]:
return json.loads(self.plan) return _json_loads(self.plan)
def set_plan(self, plan: list[dict[str, Any]]) -> None: def set_plan(self, plan: list[dict[str, Any]]) -> None:
self.plan = json.dumps(plan, ensure_ascii=False) self.plan = _json_dumps(plan)
def get_summary(self) -> Optional[dict[str, Any]]: def get_summary(self) -> Optional[dict[str, Any]]:
return json.loads(self.summary) if self.summary else None return _json_loads(self.summary) if self.summary else None
def set_summary(self, summary: dict[str, Any]) -> None: def set_summary(self, summary: dict[str, Any]) -> None:
self.summary = json.dumps(summary, ensure_ascii=False) self.summary = _json_dumps(summary)
class EvalRunDB(SQLModel, table=True): class EvalRunDB(SQLModel, table=True):
@ -218,10 +228,10 @@ class EvalRunDB(SQLModel, table=True):
) )
def get_summary(self) -> Optional[dict[str, Any]]: def get_summary(self) -> Optional[dict[str, Any]]:
return json.loads(self.summary) if self.summary else None return _json_loads(self.summary) if self.summary else None
def set_summary(self, summary: dict[str, Any]) -> None: def set_summary(self, summary: dict[str, Any]) -> None:
self.summary = json.dumps(summary, ensure_ascii=False) self.summary = _json_dumps(summary)
class TurnDB(SQLModel, table=True): class TurnDB(SQLModel, table=True):
@ -243,16 +253,16 @@ class TurnDB(SQLModel, table=True):
run: Optional[EvalRunDB] = Relationship(back_populates="turns") run: Optional[EvalRunDB] = Relationship(back_populates="turns")
def get_sent_message(self) -> dict[str, Any]: def get_sent_message(self) -> dict[str, Any]:
return json.loads(self.sent_message) return _json_loads(self.sent_message)
def set_sent_message(self, message: dict[str, Any]) -> None: def set_sent_message(self, message: dict[str, Any]) -> None:
self.sent_message = json.dumps(message, ensure_ascii=False) self.sent_message = _json_dumps(message)
def get_reply(self) -> Optional[dict[str, Any]]: def get_reply(self) -> Optional[dict[str, Any]]:
return json.loads(self.reply) if self.reply else None return _json_loads(self.reply) if self.reply else None
def set_reply(self, reply: Optional[dict[str, Any]]) -> None: def set_reply(self, reply: Optional[dict[str, Any]]) -> None:
self.reply = json.dumps(reply, ensure_ascii=False) if reply else None self.reply = _json_dumps(reply) if reply else None
class EvalResultDB(SQLModel, table=True): class EvalResultDB(SQLModel, table=True):

View File

@ -1,6 +1,6 @@
"""Repository layer for database access.""" """Repository layer for database access."""
from typing import Optional from typing import Generic, Optional, TypeVar
from sqlmodel import Session, select from sqlmodel import Session, select
@ -18,131 +18,56 @@ from agenteval.storage.db import (
) )
from agenteval.storage.model_config_repository import ScenarioModelBindingRepository from agenteval.storage.model_config_repository import ScenarioModelBindingRepository
M = TypeVar("M") # domain model
def _target_to_db(target: EvalTarget) -> EvalTargetDB: DB = TypeVar("DB") # persisted table row
db = EvalTargetDB(
id=target.id,
name=target.name,
description=target.description,
platform=target.platform.value,
channel_type=target.channel_type.value,
status=target.status.value,
created_at=target.created_at,
updated_at=target.updated_at or utc_now(),
)
db.set_config(target.channel_config)
return db
def _target_from_db(db: EvalTargetDB) -> EvalTarget: class BaseRepository(Generic[M, DB]):
return EvalTarget( """Shared CRUD skeleton for id-keyed entity repositories.
id=db.id,
name=db.name,
description=db.description,
platform=db.platform,
channel_type=db.channel_type,
channel_config=db.get_config(),
status=db.status,
created_at=db.created_at,
updated_at=db.updated_at,
)
Subclasses declare the table (``_table``) and the ``list_all`` ordering
column name (``_order_by``, newest-first), and implement the ``_to_db`` /
``_from_db`` converter pair. The converters are instance methods so a
subclass whose ``_from_db`` needs cross-table reads (e.g. Scenario's model
bindings) can reach ``self.session``. Entities with bespoke create/update
(binding validation, versioning) override just those methods.
"""
def _scenario_to_db(scenario: Scenario) -> ScenarioDB: _table: type
db = ScenarioDB( _order_by: str
id=scenario.id,
name=scenario.name,
description=scenario.description,
created_at=scenario.created_at,
updated_at=scenario.updated_at or utc_now(),
)
db.set_tags(scenario.tags)
# mode="json" 与 update() 的考纲比较保持同一序列化形态,避免假升版
db.set_cases([case.model_dump(mode="json") for case in scenario.cases])
db.set_llm_config(scenario.llm_config)
return db
def __init__(self, session: Optional[Session] = None):
self.session = session or get_session()
def _scenario_from_db(db: ScenarioDB, session: Session) -> Scenario: def _to_db(self, obj: M) -> DB:
bindings = ScenarioModelBindingRepository(session).get_for_scenario(db.id or "") raise NotImplementedError
return Scenario(
id=db.id,
name=db.name,
description=db.description,
tags=db.get_tags(),
cases=[Case(**case) for case in db.get_cases()],
model_bindings=bindings,
llm_config=db.get_llm_config(),
version=db.version or 1,
created_at=db.created_at,
updated_at=db.updated_at,
)
def _from_db(self, db: DB) -> M:
raise NotImplementedError
def _run_to_db(run: EvalRun) -> EvalRunDB: def list_all(self) -> list[M]:
db = EvalRunDB( column = getattr(self._table, self._order_by)
id=run.id, statement = select(self._table).order_by(column.desc())
target_id=run.target_id, return [self._from_db(r) for r in self.session.exec(statement).all()]
scenario_id=run.scenario_id,
scenario_version=run.scenario_version,
campaign_id=run.campaign_id,
status=run.status.value,
triggered_by=run.triggered_by.value,
started_at=run.started_at,
completed_at=run.completed_at,
)
if run.summary is not None:
db.set_summary(run.summary.model_dump(mode="json"))
return db
def get(self, entity_id: str) -> Optional[M]:
db = self.session.get(self._table, entity_id)
return self._from_db(db) if db else None
def _run_from_db(db: EvalRunDB) -> EvalRun: def create(self, obj: M) -> M:
return EvalRun( db = self._to_db(obj)
id=db.id, self.session.add(db)
target_id=db.target_id, self.session.commit()
scenario_id=db.scenario_id, self.session.refresh(db)
scenario_version=db.scenario_version or 1, return self._from_db(db)
campaign_id=db.campaign_id,
status=db.status,
triggered_by=db.triggered_by or "manual",
started_at=db.started_at,
completed_at=db.completed_at,
summary=db.get_summary(),
)
def delete(self, entity_id: str) -> bool:
def _campaign_to_db(campaign: Campaign) -> CampaignDB: db = self.session.get(self._table, entity_id)
db = CampaignDB( if not db:
id=campaign.id, return False
name=campaign.name, self.session.delete(db)
target_id=campaign.target_id, self.session.commit()
window_seconds=campaign.window_seconds, return True
time_scale=campaign.time_scale,
status=campaign.status.value,
started_at=campaign.started_at,
completed_at=campaign.completed_at,
created_at=campaign.created_at,
)
db.set_plan([entry.model_dump(mode="json") for entry in campaign.plan])
if campaign.summary:
db.set_summary(campaign.summary)
return db
def _campaign_from_db(db: CampaignDB) -> Campaign:
return Campaign(
id=db.id,
name=db.name,
target_id=db.target_id,
window_seconds=db.window_seconds,
time_scale=db.time_scale,
plan=db.get_plan(),
status=db.status,
started_at=db.started_at,
completed_at=db.completed_at,
created_at=db.created_at,
summary=db.get_summary(),
)
def _result_to_db(result: EvalResult) -> EvalResultDB: def _result_to_db(result: EvalResult) -> EvalResultDB:
@ -171,26 +96,38 @@ def _result_from_db(db: EvalResultDB) -> EvalResult:
) )
class TargetRepository: class TargetRepository(BaseRepository[EvalTarget, EvalTargetDB]):
"""Repository for evaluation targets.""" """Repository for evaluation targets."""
def __init__(self, session: Optional[Session] = None): _table = EvalTargetDB
self.session = session or get_session() _order_by = "created_at"
def list_all(self) -> list[EvalTarget]: def _to_db(self, target: EvalTarget) -> EvalTargetDB:
statement = select(EvalTargetDB).order_by(EvalTargetDB.created_at.desc()) db = EvalTargetDB(
return [_target_from_db(r) for r in self.session.exec(statement).all()] id=target.id,
name=target.name,
description=target.description,
platform=target.platform.value,
channel_type=target.channel_type.value,
status=target.status.value,
created_at=target.created_at,
updated_at=target.updated_at or utc_now(),
)
db.set_config(target.channel_config)
return db
def get(self, target_id: str) -> Optional[EvalTarget]: def _from_db(self, db: EvalTargetDB) -> EvalTarget:
db = self.session.get(EvalTargetDB, target_id) return EvalTarget(
return _target_from_db(db) if db else None id=db.id,
name=db.name,
def create(self, target: EvalTarget) -> EvalTarget: description=db.description,
db = _target_to_db(target) platform=db.platform,
self.session.add(db) channel_type=db.channel_type,
self.session.commit() channel_config=db.get_config(),
self.session.refresh(db) status=db.status,
return _target_from_db(db) created_at=db.created_at,
updated_at=db.updated_at,
)
def update(self, target: EvalTarget) -> Optional[EvalTarget]: def update(self, target: EvalTarget) -> Optional[EvalTarget]:
existing = self.session.get(EvalTargetDB, target.id) existing = self.session.get(EvalTargetDB, target.id)
@ -206,33 +143,46 @@ class TargetRepository:
self.session.add(existing) self.session.add(existing)
self.session.commit() self.session.commit()
self.session.refresh(existing) self.session.refresh(existing)
return _target_from_db(existing) return self._from_db(existing)
def delete(self, target_id: str) -> bool:
db = self.session.get(EvalTargetDB, target_id)
if not db:
return False
self.session.delete(db)
self.session.commit()
return True
class ScenarioRepository: class ScenarioRepository(BaseRepository[Scenario, ScenarioDB]):
"""Repository for evaluation scenarios.""" """Repository for evaluation scenarios."""
def __init__(self, session: Optional[Session] = None): _table = ScenarioDB
self.session = session or get_session() _order_by = "created_at"
def list_all(self) -> list[Scenario]: def _to_db(self, scenario: Scenario) -> ScenarioDB:
statement = select(ScenarioDB).order_by(ScenarioDB.created_at.desc()) db = ScenarioDB(
return [_scenario_from_db(r, self.session) for r in self.session.exec(statement).all()] id=scenario.id,
name=scenario.name,
description=scenario.description,
created_at=scenario.created_at,
updated_at=scenario.updated_at or utc_now(),
)
db.set_tags(scenario.tags)
# mode="json" 与 update() 的考纲比较保持同一序列化形态,避免假升版
db.set_cases([case.model_dump(mode="json") for case in scenario.cases])
db.set_llm_config(scenario.llm_config)
return db
def get(self, scenario_id: str) -> Optional[Scenario]: def _from_db(self, db: ScenarioDB) -> Scenario:
db = self.session.get(ScenarioDB, scenario_id) bindings = ScenarioModelBindingRepository(self.session).get_for_scenario(db.id or "")
return _scenario_from_db(db, self.session) if db else None return Scenario(
id=db.id,
name=db.name,
description=db.description,
tags=db.get_tags(),
cases=[Case(**case) for case in db.get_cases()],
model_bindings=bindings,
llm_config=db.get_llm_config(),
version=db.version or 1,
created_at=db.created_at,
updated_at=db.updated_at,
)
def create(self, scenario: Scenario) -> Scenario: def create(self, scenario: Scenario) -> Scenario:
db = _scenario_to_db(scenario) db = self._to_db(scenario)
bindings = {purpose.value: config_id for purpose, config_id in scenario.model_bindings.items()} bindings = {purpose.value: config_id for purpose, config_id in scenario.model_bindings.items()}
try: try:
ModelConfigService(self.session).validate_bindings(bindings) ModelConfigService(self.session).validate_bindings(bindings)
@ -244,7 +194,7 @@ class ScenarioRepository:
except Exception: except Exception:
self.session.rollback() self.session.rollback()
raise raise
return _scenario_from_db(db, self.session) return self._from_db(db)
def update(self, scenario: Scenario) -> Optional[Scenario]: def update(self, scenario: Scenario) -> Optional[Scenario]:
existing = self.session.get(ScenarioDB, scenario.id) existing = self.session.get(ScenarioDB, scenario.id)
@ -277,7 +227,7 @@ class ScenarioRepository:
except Exception: except Exception:
self.session.rollback() self.session.rollback()
raise raise
return _scenario_from_db(existing, self.session) return self._from_db(existing)
def delete(self, scenario_id: str) -> bool: def delete(self, scenario_id: str) -> bool:
db = self.session.get(ScenarioDB, scenario_id) db = self.session.get(ScenarioDB, scenario_id)
@ -293,15 +243,41 @@ class ScenarioRepository:
return True return True
class RunRepository: class RunRepository(BaseRepository[EvalRun, EvalRunDB]):
"""Repository for evaluation runs.""" """Repository for evaluation runs."""
def __init__(self, session: Optional[Session] = None): _table = EvalRunDB
self.session = session or get_session() _order_by = "started_at"
def list_all(self) -> list[EvalRun]: def _to_db(self, run: EvalRun) -> EvalRunDB:
statement = select(EvalRunDB).order_by(EvalRunDB.started_at.desc()) db = EvalRunDB(
return [_run_from_db(r) for r in self.session.exec(statement).all()] id=run.id,
target_id=run.target_id,
scenario_id=run.scenario_id,
scenario_version=run.scenario_version,
campaign_id=run.campaign_id,
status=run.status.value,
triggered_by=run.triggered_by.value,
started_at=run.started_at,
completed_at=run.completed_at,
)
if run.summary is not None:
db.set_summary(run.summary.model_dump(mode="json"))
return db
def _from_db(self, db: EvalRunDB) -> EvalRun:
return EvalRun(
id=db.id,
target_id=db.target_id,
scenario_id=db.scenario_id,
scenario_version=db.scenario_version or 1,
campaign_id=db.campaign_id,
status=db.status,
triggered_by=db.triggered_by or "manual",
started_at=db.started_at,
completed_at=db.completed_at,
summary=db.get_summary(),
)
def list_by_campaign(self, campaign_id: str) -> list[EvalRun]: def list_by_campaign(self, campaign_id: str) -> list[EvalRun]:
statement = ( statement = (
@ -309,18 +285,7 @@ class RunRepository:
.where(EvalRunDB.campaign_id == campaign_id) .where(EvalRunDB.campaign_id == campaign_id)
.order_by(EvalRunDB.started_at) .order_by(EvalRunDB.started_at)
) )
return [_run_from_db(r) for r in self.session.exec(statement).all()] return [self._from_db(r) for r in self.session.exec(statement).all()]
def get(self, run_id: str) -> Optional[EvalRun]:
db = self.session.get(EvalRunDB, run_id)
return _run_from_db(db) if db else None
def create(self, run: EvalRun) -> EvalRun:
db = _run_to_db(run)
self.session.add(db)
self.session.commit()
self.session.refresh(db)
return _run_from_db(db)
def mark_orphans_failed(self) -> int: def mark_orphans_failed(self) -> int:
"""服务启动时清理:把遗留的 running/pending 运行标记为 failed。 """服务启动时清理:把遗留的 running/pending 运行标记为 failed。
@ -357,7 +322,7 @@ class RunRepository:
self.session.add(existing) self.session.add(existing)
self.session.commit() self.session.commit()
self.session.refresh(existing) self.session.refresh(existing)
return _run_from_db(existing) return self._from_db(existing)
def get_turns(self, run_id: str) -> list[TurnDB]: def get_turns(self, run_id: str) -> list[TurnDB]:
statement = select(TurnDB).where(TurnDB.run_id == run_id).order_by(TurnDB.sent_at) statement = select(TurnDB).where(TurnDB.run_id == run_id).order_by(TurnDB.sent_at)
@ -367,36 +332,44 @@ class RunRepository:
statement = select(EvalResultDB).where(EvalResultDB.run_id == run_id) statement = select(EvalResultDB).where(EvalResultDB.run_id == run_id)
return [_result_from_db(r) for r in self.session.exec(statement).all()] return [_result_from_db(r) for r in self.session.exec(statement).all()]
def delete(self, run_id: str) -> bool:
"""Delete a run. ORM-level cascade removes associated turns/results."""
db = self.session.get(EvalRunDB, run_id)
if not db:
return False
self.session.delete(db)
self.session.commit()
return True
class CampaignRepository(BaseRepository[Campaign, CampaignDB]):
class CampaignRepository:
"""Repository for evaluation campaigns (评估活动).""" """Repository for evaluation campaigns (评估活动)."""
def __init__(self, session: Optional[Session] = None): _table = CampaignDB
self.session = session or get_session() _order_by = "created_at"
def list_all(self) -> list[Campaign]: def _to_db(self, campaign: Campaign) -> CampaignDB:
statement = select(CampaignDB).order_by(CampaignDB.created_at.desc()) db = CampaignDB(
return [_campaign_from_db(r) for r in self.session.exec(statement).all()] id=campaign.id,
name=campaign.name,
target_id=campaign.target_id,
window_seconds=campaign.window_seconds,
time_scale=campaign.time_scale,
status=campaign.status.value,
started_at=campaign.started_at,
completed_at=campaign.completed_at,
created_at=campaign.created_at,
)
db.set_plan([entry.model_dump(mode="json") for entry in campaign.plan])
if campaign.summary:
db.set_summary(campaign.summary.model_dump(mode="json"))
return db
def get(self, campaign_id: str) -> Optional[Campaign]: def _from_db(self, db: CampaignDB) -> Campaign:
db = self.session.get(CampaignDB, campaign_id) return Campaign(
return _campaign_from_db(db) if db else None id=db.id,
name=db.name,
def create(self, campaign: Campaign) -> Campaign: target_id=db.target_id,
db = _campaign_to_db(campaign) window_seconds=db.window_seconds,
self.session.add(db) time_scale=db.time_scale,
self.session.commit() plan=db.get_plan(),
self.session.refresh(db) status=db.status,
return _campaign_from_db(db) started_at=db.started_at,
completed_at=db.completed_at,
created_at=db.created_at,
summary=db.get_summary(),
)
def update(self, campaign: Campaign) -> Optional[Campaign]: def update(self, campaign: Campaign) -> Optional[Campaign]:
existing = self.session.get(CampaignDB, campaign.id) existing = self.session.get(CampaignDB, campaign.id)
@ -411,11 +384,11 @@ class CampaignRepository:
existing.started_at = campaign.started_at existing.started_at = campaign.started_at
existing.completed_at = campaign.completed_at existing.completed_at = campaign.completed_at
if campaign.summary is not None: if campaign.summary is not None:
existing.set_summary(campaign.summary) existing.set_summary(campaign.summary.model_dump(mode="json"))
self.session.add(existing) self.session.add(existing)
self.session.commit() self.session.commit()
self.session.refresh(existing) self.session.refresh(existing)
return _campaign_from_db(existing) return self._from_db(existing)
class ResultRepository: class ResultRepository:

View File

@ -6,8 +6,9 @@ from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel from pydantic import BaseModel
from sqlmodel import Session from sqlmodel import Session
from agenteval.evaluation.case_verdict import CaseEvidence, resolve_case_verdicts
from agenteval.evaluation.engine import EvalEngine from agenteval.evaluation.engine import EvalEngine
from agenteval.models import EvalRun, RunStatus, RunTrigger from agenteval.models import EvalRun, RunStatus, RunSummary, RunTrigger
from agenteval.storage.db import get_session, iso_utc from agenteval.storage.db import get_session, iso_utc
from agenteval.storage.repository import RunRepository, ScenarioRepository, TargetRepository from agenteval.storage.repository import RunRepository, ScenarioRepository, TargetRepository
from agenteval.task_registry import TaskRegistry from agenteval.task_registry import TaskRegistry
@ -169,6 +170,37 @@ async def get_run_logs(run_id: str, session: Session = Depends(get_db)) -> dict:
for r in results for r in results
] ]
# Authoritative per-case verdicts: resolve_case_verdicts prefers the engine's
# stored case_outcomes and approximates only for legacy runs (single seam).
evidence: dict[str, dict] = {}
for t in turns:
ev = evidence.get(t.case_id)
if ev is None:
ev = {"has_turns": True, "all_replied": True, "passes": []}
evidence[t.case_id] = ev
else:
ev["has_turns"] = True
if t.get_reply() is None:
ev["all_replied"] = False
for r in results:
ev = evidence.setdefault(r.case_id, {"has_turns": False, "all_replied": True, "passes": []})
ev["passes"].append(r.passed)
summary = run.summary or RunSummary()
errored_case_ids = {e.get("case_id") for e in summary.case_errors}
verdicts = resolve_case_verdicts(
case_outcomes=summary.case_outcomes,
evidence={
cid: CaseEvidence(
has_turns=ev["has_turns"],
all_replied=ev["all_replied"],
result_passes=tuple(ev["passes"]),
)
for cid, ev in evidence.items()
},
errored_case_ids=errored_case_ids,
)
case_verdicts = {cid: {"passed": v.passed, "connectivity": v.connectivity} for cid, v in verdicts.items()}
scenario_snapshot: dict = {} scenario_snapshot: dict = {}
scenario = ScenarioRepository(session).get(run.scenario_id) scenario = ScenarioRepository(session).get(run.scenario_id)
if scenario: if scenario:
@ -191,4 +223,9 @@ async def get_run_logs(run_id: str, session: Session = Depends(get_db)) -> dict:
"rule_pass_threshold": case.rule_pass_threshold, "rule_pass_threshold": case.rule_pass_threshold,
} }
return {"turns": turns_data, "results": results_data, "scenario_snapshot": scenario_snapshot} return {
"turns": turns_data,
"results": results_data,
"case_verdicts": case_verdicts,
"scenario_snapshot": scenario_snapshot,
}

View File

@ -278,6 +278,7 @@ export interface CaseSnapshot {
export interface RunLogsResponse { export interface RunLogsResponse {
turns: RunLogsTurn[] turns: RunLogsTurn[]
results: RunLogsResult[] results: RunLogsResult[]
case_verdicts: Record<string, { passed: boolean; connectivity: boolean }>
scenario_snapshot: Record<string, CaseSnapshot> scenario_snapshot: Record<string, CaseSnapshot>
} }

View File

@ -160,9 +160,10 @@ export function useRunSession(): RunSession {
reason: r.reason, reason: r.reason,
}) })
} }
const verdicts = data.case_verdicts ?? {}
for (const cs of caseMap.values()) { for (const cs of caseMap.values()) {
cs.turns.sort((a, b) => a.roundIndex - b.roundIndex) cs.turns.sort((a, b) => a.roundIndex - b.roundIndex)
cs.passed = cs.ruleResults.length > 0 && cs.ruleResults.every((r) => r.passed) cs.passed = verdicts[cs.caseId]?.passed ?? false
} }
const cases = Array.from(caseMap.values()) const cases = Array.from(caseMap.values())
const total = cases.length const total = cases.length

View File

@ -0,0 +1,116 @@
"""Unit tests for the pure single-run summary seam.
build_run_summary is the single reusable落点 for one run's口径 (pass_rate /
judged_pass_rate / avg_latency / connectivity split), parallel to
metrics.aggregate_runs (cross-run) and judgement.combine_case_outcome
(case-level). No IO fed constructed material, asserted directly.
"""
from agenteval.evaluation.judgement import CaseOutcome
from agenteval.evaluation.run_summary import build_run_summary
def _outcomes(**kw: CaseOutcome) -> dict[str, CaseOutcome]:
return dict(kw)
def test_all_judged_passing():
summary = build_run_summary(
case_outcomes=_outcomes(
a=CaseOutcome(passed=True, connectivity=False),
b=CaseOutcome(passed=True, connectivity=False),
),
latencies=[100.0, 300.0],
rule_passes=[True, True, True],
)
assert summary.total_cases == 2
assert summary.passed_cases == 2
assert summary.failed_cases == 0
assert summary.pass_rate == 1.0
assert summary.judged_pass_rate == 1.0
assert summary.total_rules == 3
assert summary.passed_rules == 3
assert summary.avg_latency_ms == 200.0
def test_mixed_pass_fail():
summary = build_run_summary(
case_outcomes=_outcomes(
a=CaseOutcome(passed=True, connectivity=False),
b=CaseOutcome(passed=False, connectivity=False),
c=CaseOutcome(passed=False, connectivity=False),
),
latencies=[],
rule_passes=[True, False],
)
assert summary.total_cases == 3
assert summary.passed_cases == 1
assert summary.failed_cases == 2
assert summary.pass_rate == 0.3333
assert summary.judged_pass_rate == 0.3333
assert summary.passed_rules == 1
assert summary.total_rules == 2
assert summary.avg_latency_ms is None
def test_connectivity_excluded_from_judged_rate():
# 连通用例计入总通过率但从判定型口径的分子分母双双剔除report.py 现口径)
summary = build_run_summary(
case_outcomes=_outcomes(
conn=CaseOutcome(passed=True, connectivity=True),
judged_pass=CaseOutcome(passed=True, connectivity=False),
judged_fail=CaseOutcome(passed=False, connectivity=False),
),
latencies=[50.0],
rule_passes=[True, False],
)
assert summary.total_cases == 3
assert summary.passed_cases == 2 # conn + judged_pass
assert summary.pass_rate == 0.6667
# judged: 分母 = 3 - 1 连通 = 2分子 = 2 通过 - 1 连通 = 1
assert summary.judged_pass_rate == 0.5
def test_all_connectivity_yields_no_judged_rate():
summary = build_run_summary(
case_outcomes=_outcomes(
a=CaseOutcome(passed=True, connectivity=True),
b=CaseOutcome(passed=True, connectivity=True),
),
latencies=[10.0],
rule_passes=[],
)
assert summary.pass_rate == 1.0
assert summary.judged_pass_rate is None # judged_total == 0
def test_empty_run():
summary = build_run_summary(case_outcomes={}, latencies=[], rule_passes=[])
assert summary.total_cases == 0
assert summary.pass_rate == 0.0
assert summary.judged_pass_rate is None
assert summary.avg_latency_ms is None
def test_case_errors_and_model_configs_passed_through():
errors = [{"case_id": "x", "message": "boom"}]
configs = {"judge": {"model": "gpt-4o"}}
summary = build_run_summary(
case_outcomes=_outcomes(x=CaseOutcome(passed=False, connectivity=False)),
latencies=[],
rule_passes=[],
case_errors=errors,
model_configs=configs,
)
assert summary.case_errors == errors
assert summary.model_configs == configs
def test_case_outcomes_persisted_as_verdict_snapshot():
summary = build_run_summary(
case_outcomes=_outcomes(a=CaseOutcome(passed=True, connectivity=True)),
latencies=[],
rule_passes=[],
)
assert summary.case_outcomes["a"].passed is True
assert summary.case_outcomes["a"].connectivity is True

View File

@ -0,0 +1,74 @@
"""CampaignSummary VO — typed campaign scheduler state, restart-safe (ADR-0003).
Mirrors RunSummary's typed treatment: Campaign.summary is no longer a bare
dict. Legacy dict summaries coerce; unknown top-level keys survive
(extra=allow) so older records keep parsing.
"""
from agenteval.models import (
Campaign,
CampaignPlanEntry,
CampaignSummary,
SchedulerState,
)
from agenteval.storage.repository import CampaignRepository, TargetRepository
from tests.unit.test_repository import _make_target
def _campaign(**summary_kw) -> Campaign:
kw = dict(
name="c",
target_id="t-1",
window_seconds=3600,
plan=[CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=1)],
)
kw.update(summary_kw)
return Campaign(**kw)
def test_summary_defaults_are_empty():
s = CampaignSummary()
assert s.scheduler.spawned_indices == []
assert s.scheduler.errors == []
def test_scheduler_state_holds_progress():
s = SchedulerState(spawned_indices=[0, 2], errors=[{"index": 1, "error": "boom"}])
assert s.spawned_indices == [0, 2]
assert s.errors[0]["error"] == "boom"
def test_legacy_dict_summary_coerces():
c = _campaign(summary={"scheduler": {"spawned_indices": [0, 1], "errors": [{"index": 2, "error": "x"}]}})
assert isinstance(c.summary, CampaignSummary)
assert c.summary.scheduler.spawned_indices == [0, 1]
assert c.summary.scheduler.errors[0]["error"] == "x"
def test_assigning_dict_coerces_to_vo():
c = _campaign()
c.summary = {"scheduler": {"spawned_indices": [3]}}
assert isinstance(c.summary, CampaignSummary)
assert c.summary.scheduler.spawned_indices == [3]
def test_unknown_top_level_keys_survive():
c = _campaign(summary={"scheduler": {}, "future_axis": {"availability": 0.9}})
assert isinstance(c.summary, CampaignSummary)
assert c.summary.model_extra["future_axis"] == {"availability": 0.9}
def test_summary_survives_repository_round_trip(db_session):
TargetRepository(db_session).create(_make_target())
repo = CampaignRepository(db_session)
repo.create(
_campaign(
id="cp-1",
summary=CampaignSummary(scheduler=SchedulerState(spawned_indices=[0, 1], errors=[{"index": 2}])),
)
)
fetched = repo.get("cp-1")
assert fetched is not None
assert isinstance(fetched.summary, CampaignSummary)
assert fetched.summary.scheduler.spawned_indices == [0, 1]
assert fetched.summary.scheduler.errors == [{"index": 2}]

View File

@ -0,0 +1,105 @@
"""Unit tests for resolve_case_verdicts — the single case-verdict read seam."""
from agenteval.evaluation.case_verdict import CaseEvidence, resolve_case_verdicts
from agenteval.models import CaseOutcomeSummary
def _ev(has_turns=True, all_replied=True, result_passes=()):
return CaseEvidence(has_turns=has_turns, all_replied=all_replied, result_passes=tuple(result_passes))
# ── authoritative wins ──────────────────────────────────────────────────────
def test_authoritative_outcome_used_verbatim_ignoring_evidence():
# Evidence would approximate passed=True (all rules pass), but the authoritative
# verdict says failed (e.g. WEIGHTED/ANY logic) — authority must win.
verdicts = resolve_case_verdicts(
case_outcomes={"c1": CaseOutcomeSummary(passed=False, connectivity=False)},
evidence={"c1": _ev(result_passes=(True, True))},
errored_case_ids=set(),
)
assert verdicts["c1"] == CaseOutcomeSummary(passed=False, connectivity=False)
def test_authoritative_connectivity_preserved():
verdicts = resolve_case_verdicts(
case_outcomes={"c1": CaseOutcomeSummary(passed=True, connectivity=True)},
evidence={"c1": _ev(result_passes=(False,))},
errored_case_ids=set(),
)
assert verdicts["c1"].connectivity is True
assert verdicts["c1"].passed is True
# ── legacy approximation (case_outcomes missing) ────────────────────────────
def test_legacy_connectivity_case_passes():
# No judged results, has turns, every turn replied, not errored → connectivity pass.
verdicts = resolve_case_verdicts(
case_outcomes={},
evidence={"c1": _ev(has_turns=True, all_replied=True, result_passes=())},
errored_case_ids=set(),
)
assert verdicts["c1"] == CaseOutcomeSummary(passed=True, connectivity=True)
def test_legacy_all_rules_pass():
verdicts = resolve_case_verdicts(
case_outcomes={},
evidence={"c1": _ev(result_passes=(True, True))},
errored_case_ids=set(),
)
assert verdicts["c1"] == CaseOutcomeSummary(passed=True, connectivity=False)
def test_legacy_some_rule_fails():
verdicts = resolve_case_verdicts(
case_outcomes={},
evidence={"c1": _ev(result_passes=(True, False))},
errored_case_ids=set(),
)
assert verdicts["c1"] == CaseOutcomeSummary(passed=False, connectivity=False)
def test_legacy_fault_no_results_not_all_replied():
# No results and a turn missing its reply → fault, not connectivity → fail (ADR-0002).
verdicts = resolve_case_verdicts(
case_outcomes={},
evidence={"c1": _ev(has_turns=True, all_replied=False, result_passes=())},
errored_case_ids=set(),
)
assert verdicts["c1"] == CaseOutcomeSummary(passed=False, connectivity=False)
def test_legacy_errored_case_not_connectivity():
# A case with a case-level error is not connectivity even if it replied.
verdicts = resolve_case_verdicts(
case_outcomes={},
evidence={"c1": _ev(has_turns=True, all_replied=True, result_passes=())},
errored_case_ids={"c1"},
)
assert verdicts["c1"] == CaseOutcomeSummary(passed=False, connectivity=False)
def test_legacy_no_turns_no_results_is_fault():
verdicts = resolve_case_verdicts(
case_outcomes={},
evidence={"c1": _ev(has_turns=False, all_replied=True, result_passes=())},
errored_case_ids=set(),
)
assert verdicts["c1"] == CaseOutcomeSummary(passed=False, connectivity=False)
# ── mixed authoritative + legacy in one call ────────────────────────────────
def test_mixed_authoritative_and_legacy():
verdicts = resolve_case_verdicts(
case_outcomes={"auth": CaseOutcomeSummary(passed=True, connectivity=False)},
evidence={
"auth": _ev(result_passes=(False,)), # authority overrides
"legacy": _ev(result_passes=(True,)), # approximated
},
errored_case_ids=set(),
)
assert verdicts["auth"] == CaseOutcomeSummary(passed=True, connectivity=False)
assert verdicts["legacy"] == CaseOutcomeSummary(passed=True, connectivity=False)

View File

@ -0,0 +1,45 @@
"""Unit tests for derive_implicit_rules — 期望→隐式规则的纯翻译."""
from agenteval.evaluation.implicit_rules import derive_implicit_rules
from agenteval.models import Expectation
def test_empty_expectation_yields_no_rules():
assert derive_implicit_rules(Expectation()) == []
def test_intent_and_coherence_alone_yield_no_rules():
# intent / coherence_min_score 不派生隐式规则(无对应规则类型)
exp = Expectation(intent="问诊分流", coherence_min_score=0.8)
assert derive_implicit_rules(exp) == []
def test_response_time_yields_response_time_rule():
rules = derive_implicit_rules(Expectation(response_time_max_ms=2000))
assert len(rules) == 1
assert rules[0].type == "response_time"
assert rules[0].params == {"max_ms": 2000}
def test_keywords_include_yields_keyword_rule():
rules = derive_implicit_rules(Expectation(keywords_include=["挂号", "门诊"]))
assert len(rules) == 1
assert rules[0].type == "keyword_match"
assert rules[0].params == {"keywords": ["挂号", "门诊"], "exclude_keywords": []}
def test_keywords_exclude_alone_yields_keyword_rule():
rules = derive_implicit_rules(Expectation(keywords_exclude=["投诉"]))
assert len(rules) == 1
assert rules[0].type == "keyword_match"
assert rules[0].params == {"keywords": [], "exclude_keywords": ["投诉"]}
def test_both_expectations_yield_two_rules_in_order():
exp = Expectation(
response_time_max_ms=1500,
keywords_include=["预约"],
keywords_exclude=["取消"],
)
rules = derive_implicit_rules(exp)
assert [r.type for r in rules] == ["response_time", "keyword_match"]

View File

@ -0,0 +1,113 @@
"""CRUD + serialization round-trip characterization for the repository layer.
These lock the create get list_all delete contract shared by the
Target / Run / Campaign repositories (the BaseRepository skeleton), plus the
summary/plan JSON round-trip, so the generic-base refactor stays behaviour-
preserving. Scenario's bespoke create/update (binding validation, versioning)
is covered by its own tests.
"""
from datetime import datetime, timezone
from agenteval.models import (
Campaign,
CampaignPlanEntry,
ChannelType,
EvalRun,
EvalTarget,
PlatformType,
RunStatus,
RunSummary,
TargetStatus,
)
from agenteval.storage.repository import (
CampaignRepository,
RunRepository,
TargetRepository,
)
def _make_target(tid: str = "t-1") -> EvalTarget:
return EvalTarget(
id=tid,
name="target",
platform=PlatformType.AI_DIGITAL_EMPLOYEE,
channel_type=ChannelType.TUTU_API,
channel_config={
"base_url": "x",
"token": "x",
"tenant": "x",
"chat_channel_id": "x",
"chat_contact_id": "x",
},
)
def test_target_crud_round_trip(db_session):
repo = TargetRepository(db_session)
repo.create(_make_target())
fetched = repo.get("t-1")
assert fetched is not None
assert fetched.name == "target"
assert fetched.channel_config["base_url"] == "x"
assert fetched.status == TargetStatus.ACTIVE or fetched.status is not None
assert [t.id for t in repo.list_all()] == ["t-1"]
assert repo.delete("t-1") is True
assert repo.get("t-1") is None
assert repo.delete("t-1") is False
def test_target_list_all_newest_first(db_session):
repo = TargetRepository(db_session)
older = _make_target("t-old")
older.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
newer = _make_target("t-new")
newer.created_at = datetime(2026, 6, 1, tzinfo=timezone.utc)
repo.create(older)
repo.create(newer)
assert [t.id for t in repo.list_all()] == ["t-new", "t-old"]
def test_run_summary_json_round_trip(db_session):
TargetRepository(db_session).create(_make_target())
repo = RunRepository(db_session)
repo.create(
EvalRun(
id="r-1",
target_id="t-1",
scenario_id="s-1",
status=RunStatus.COMPLETED,
summary=RunSummary(total_cases=2, passed_cases=1, pass_rate=0.5),
)
)
fetched = repo.get("r-1")
assert fetched is not None
assert fetched.summary is not None
assert fetched.summary.total_cases == 2
assert fetched.summary.pass_rate == 0.5
def test_campaign_crud_and_plan_round_trip(db_session):
TargetRepository(db_session).create(_make_target())
repo = CampaignRepository(db_session)
repo.create(
Campaign(
id="cp-1",
name="campaign",
target_id="t-1",
window_seconds=3600,
plan=[CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=2)],
)
)
fetched = repo.get("cp-1")
assert fetched is not None
assert fetched.plan[0].scenario_id == "s-1"
assert fetched.plan[0].count == 2
assert [c.id for c in repo.list_all()] == ["cp-1"]
# Campaign inherits the shared delete() from the base repository.
assert repo.delete("cp-1") is True
assert repo.get("cp-1") is None