File size: 6,409 Bytes
13fe504 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 | """Honest benchmark gate for the verified agent.
Two guarantees are enforced here, both grounded in reality rather than hope:
1. **Non-regression by construction (offline, CI-runnable).** The verified agent
seeds with the deterministic floor, so with the ``deterministic`` policy it
must produce *exactly* the legacy pipeline's verified fixes, and with any
policy it must produce a *superset*. :func:`compare_agent_vs_deterministic`
checks this directly against bundled fixtures — no model or network needed.
2. **Promotion bar (fed real eval numbers).** The agent may only become the
default repair path if a real evaluation shows it beats the deterministic
baseline F1 with zero safety regressions. :func:`agent_promotion_verdict`
encodes that decision as pure, testable logic.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
from dataforge.agent.controller import AgentRepairRequest, run_agent_repair
from dataforge.detectors.base import Schema
from dataforge.engine.repair import RepairPipelineRequest, run_repair_pipeline
__all__ = [
"AgentGateReport",
"FixtureParity",
"PromotionVerdict",
"agent_promotion_verdict",
"compare_agent_vs_deterministic",
"default_gate_fixtures",
]
@dataclass(frozen=True)
class FixtureParity:
"""Per-fixture comparison of the deterministic floor and the agent.
Args:
fixture: The fixture path that was evaluated.
floor_fix_count: Verified fixes from the legacy deterministic pipeline.
agent_fix_count: Verified fixes from the agent (deterministic policy).
agent_floor_count: The agent's own reported deterministic-floor count.
parity: Whether the agent reproduced the floor exactly (no regression).
"""
fixture: str
floor_fix_count: int
agent_fix_count: int
agent_floor_count: int
parity: bool
@dataclass(frozen=True)
class AgentGateReport:
"""Aggregate non-regression report across fixtures."""
fixtures: tuple[FixtureParity, ...] = field(default_factory=tuple)
@property
def all_parity(self) -> bool:
"""True when the agent matches the deterministic floor on every fixture."""
return all(item.parity for item in self.fixtures) and bool(self.fixtures)
def default_gate_fixtures() -> list[Path]:
"""Return offline, bundled CSV fixtures suitable for the parity gate."""
root = Path(__file__).resolve().parents[1]
candidates = [
root / "fixtures" / "hospital_10rows.csv",
root / "datasets" / "embedded" / "hospital" / "dirty.csv",
]
return [path for path in candidates if path.is_file()]
def compare_agent_vs_deterministic(
fixture_paths: list[Path] | None = None,
*,
schema: Schema | None = None,
) -> AgentGateReport:
"""Compare the agent (deterministic policy) to the legacy pipeline.
Args:
fixture_paths: CSV fixtures to evaluate. Defaults to bundled fixtures.
schema: Optional schema applied to both paths identically.
Returns:
An :class:`AgentGateReport`. ``all_parity`` is the CI assertion: the
verified agent never regresses the deterministic baseline.
"""
paths = fixture_paths if fixture_paths is not None else default_gate_fixtures()
results: list[FixtureParity] = []
for path in paths:
resolved = path.resolve()
legacy = run_repair_pipeline(
RepairPipelineRequest(source_path=resolved, mode="dry_run", schema=schema)
)
agent = run_agent_repair(
AgentRepairRequest(
source_path=resolved, mode="dry_run", schema=schema, policy="deterministic"
)
)
floor_count = len(legacy.fixes)
results.append(
FixtureParity(
fixture=str(resolved),
floor_fix_count=floor_count,
agent_fix_count=agent.fixes_count,
agent_floor_count=agent.floor_fix_count,
parity=agent.fixes_count == floor_count and agent.floor_fix_count == floor_count,
)
)
return AgentGateReport(fixtures=tuple(results))
@dataclass(frozen=True)
class PromotionVerdict:
"""Whether the agent may be promoted to the default repair path."""
promote: bool
reason: str
def agent_promotion_verdict(
*,
agent_f1: float,
baseline_f1: float,
safety_regressions: int,
parity_ok: bool,
min_margin: float = 0.0,
) -> PromotionVerdict:
"""Decide whether the agent may become the default, from real eval numbers.
The agent is promoted only when it (1) reproduces the deterministic floor
(no structural regression), (2) introduces zero safety regressions, and
(3) beats the deterministic baseline F1 by at least ``min_margin``.
Args:
agent_f1: Measured F1 of the agent policy on the benchmark.
baseline_f1: Measured F1 of the deterministic baseline.
safety_regressions: Count of adversarial/safety cases the agent newly
failed (must be zero).
parity_ok: Whether :func:`compare_agent_vs_deterministic` passed.
min_margin: Minimum F1 improvement required over the baseline.
Returns:
A :class:`PromotionVerdict` with a human-readable reason.
"""
if not parity_ok:
return PromotionVerdict(
promote=False,
reason="Agent does not reproduce the deterministic floor; structural regression.",
)
if safety_regressions > 0:
return PromotionVerdict(
promote=False,
reason=f"Agent introduced {safety_regressions} safety regression(s); blocked.",
)
if agent_f1 < baseline_f1 + min_margin:
return PromotionVerdict(
promote=False,
reason=(
f"Agent F1 {agent_f1:.4f} does not beat baseline {baseline_f1:.4f} "
f"by the required margin {min_margin:.4f}."
),
)
return PromotionVerdict(
promote=True,
reason=(
f"Agent F1 {agent_f1:.4f} beats baseline {baseline_f1:.4f} "
f"(margin >= {min_margin:.4f}) with no safety regression and floor parity."
),
)
|