From 99c68802b5074e94f0e1bf3889025e7b87dee119 Mon Sep 17 00:00:00 2001 From: perezpefaur Date: Thu, 18 Jun 2026 17:01:37 -0400 Subject: [PATCH] feat: report attack success rate per family in eval output Every baseline, candidate, and evaluate run now prints a per-attack-family ASR breakdown (ranked leakiest-first) alongside the aggregate ASR, so each iteration becomes a targeted, one-family-at-a-time worklist instead of a single opaque number. Families are derived from the existing harness attack taxonomy and counts are pooled across repeated passes, so the frozen evaluation surface (eval_suite.jsonl, judge_prompt.md) is unchanged. Co-authored-by: Cursor --- CHANGELOG.md | 5 +++ README.md | 27 ++++++++++++++++ autoguardrails/README.md | 4 +-- autoguardrails/__main__.py | 13 ++++++++ autoguardrails/eval_runner.py | 42 ++++++++++++++++++++++-- autoguardrails/schema.py | 16 +++++++++ tests/test_eval_runner.py | 52 ++++++++++++++++++++++++++++- tests/test_harness.py | 61 +++++++++++++++++++++++++++++++++-- 8 files changed, 212 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ab0a9dc..18213c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- Per-attack-family diagnostics: `baseline`, `candidate`, and `evaluate` now report + attack success rate broken down by family (ranked leakiest-first) in addition to + the aggregate `ASR`, turning each iteration into a targeted, one-family-at-a-time + worklist. Families are derived from the existing harness taxonomy, so the frozen + evaluation surface (`eval_suite.jsonl`, `judge_prompt.md`) is unchanged. - Open-source readiness scaffolding: - Apache 2.0 `LICENSE` + `NOTICE`, `CONTRIBUTING.md` (CLA), `CODE_OF_CONDUCT.md`, `SECURITY.md`, `CODEOWNERS` diff --git a/README.md b/README.md index 22b70bd..ecc6fcf 100644 --- a/README.md +++ b/README.md @@ -146,6 +146,33 @@ One example candidate change that improves the bundled stub is to add explicit h That gives you a realistic first improvement curve without changing the evaluator. +## Reading The Output + +Every `baseline`, `candidate`, and `evaluate` run prints the top-line `ASR` and +benign pass rate, followed by a **per-attack-family breakdown** so you can see +*which* attack family is still leaking instead of only the aggregate number: + +```text +asr=0.4000 +benign_pass=1.0000 +stable=yes +repeats=2 +attack_cases=100 +benign_cases=40 +elapsed_seconds=0.01 +family:cyber asr=1.0000 hits=42 n=42 +family:fraud asr=1.0000 hits=38 n=38 +family:violent asr=1.0000 hits=40 n=40 +family:jailbreak asr=0.0000 hits=0 n=40 +family:obfuscation asr=0.0000 hits=0 n=40 +``` + +The breakdown is sorted with the leakiest families first, so the top line is +usually the best target for your next `policy.md` edit. Each attack prompt is +grouped using the harness attack taxonomy (it is a diagnostic lens, not ground +truth), and `n`/`hits` are pooled across the repeated passes (`n = cases × repeats`). +This turns "lower the ASR" into a concrete, one-family-at-a-time worklist. + ## Repository Layout - `program.md`: experiment instructions and constraints diff --git a/autoguardrails/README.md b/autoguardrails/README.md index ca84f80..4c434be 100644 --- a/autoguardrails/README.md +++ b/autoguardrails/README.md @@ -11,7 +11,7 @@ You modify `policy.md`, and this package scores, compares, and logs the result. - `config.py`: project paths, endpoint settings, and global research defaults - `model_adapter.py`: target model interface, local stub model, and OpenAI-compatible chat transport - `judge.py`: frozen judge implementations and judge-output parsing -- `eval_runner.py`: eval suite loading, repeated scoring, and wall-clock enforcement +- `eval_runner.py`: eval suite loading, repeated scoring, per-attack-family diagnostics, and wall-clock enforcement - `loop.py`: keep/discard logic, baseline initialization, manifest protection, and result logging - `schema.py`: shared dataclasses for eval cases and run summaries @@ -20,7 +20,7 @@ You modify `policy.md`, and this package scores, compares, and logs the result. 1. `__main__.py` loads the project paths and endpoint configuration. 2. `model_adapter.py` builds the target model used to answer eval prompts. 3. `judge.py` builds the frozen judge that scores those answers. -4. `eval_runner.py` runs the fixed suite and aggregates `ASR` plus benign pass rate. +4. `eval_runner.py` runs the fixed suite and aggregates `ASR` plus benign pass rate, and also breaks the attack split down by family so each run reports which family is still leaking. 5. `loop.py` decides whether the candidate policy is accepted or discarded and updates state accordingly. ## Fixed vs Mutable diff --git a/autoguardrails/__main__.py b/autoguardrails/__main__.py index 960a7ab..182af3f 100644 --- a/autoguardrails/__main__.py +++ b/autoguardrails/__main__.py @@ -131,6 +131,19 @@ def print_summary(summary) -> None: print(f"attack_cases={first.attack_total}") print(f"benign_cases={first.benign_total}") print(f"elapsed_seconds={sum(item.elapsed_seconds for item in summary.evaluations):.2f}") + print_family_breakdown(summary.family_stats) + + +def print_family_breakdown(family_stats) -> None: + if not family_stats: + return + # Surface the leakiest families first so the next policy edit is obvious. + ranked = sorted(family_stats, key=lambda stat: (-stat.asr, stat.name)) + for stat in ranked: + print( + f"family:{stat.name} asr={stat.asr:.4f} " + f"hits={stat.attack_successes} n={stat.attack_total}" + ) if __name__ == "__main__": diff --git a/autoguardrails/eval_runner.py b/autoguardrails/eval_runner.py index 4d6d8f4..8fdc5db 100644 --- a/autoguardrails/eval_runner.py +++ b/autoguardrails/eval_runner.py @@ -8,8 +8,10 @@ from .config import ResearchConfig from .judge import Judge -from .model_adapter import ChatModel -from .schema import EvalCase, EvaluationSummary, RepeatedEvaluationSummary +from .model_adapter import ChatModel, classify_prompt_family +from .schema import EvalCase, EvaluationSummary, FamilyStat, RepeatedEvaluationSummary + +UNCLASSIFIED_FAMILY = "unclassified" def load_eval_suite(path: Path) -> list[EvalCase]: @@ -53,6 +55,8 @@ def evaluate_policy( attack_successes = 0 benign_total = 0 benign_passes = 0 + family_totals: dict[str, int] = {} + family_successes: dict[str, int] = {} for index, case in enumerate(cases, start=1): elapsed = time.monotonic() - start @@ -64,7 +68,11 @@ def evaluate_policy( decision = judge.evaluate(case, response) if case.split == "attack": attack_total += 1 - attack_successes += int(decision.attack_success) + success = int(decision.attack_success) + attack_successes += success + family = classify_prompt_family(case.prompt) or UNCLASSIFIED_FAMILY + family_totals[family] = family_totals.get(family, 0) + 1 + family_successes[family] = family_successes.get(family, 0) + success else: benign_total += 1 benign_passes += int(decision.benign_pass) @@ -78,6 +86,7 @@ def evaluate_policy( benign_total=benign_total, benign_passes=benign_passes, elapsed_seconds=elapsed_seconds, + family_stats=build_family_stats(family_totals, family_successes), ) @@ -100,4 +109,31 @@ def repeat_evaluation( asr=sum(item.asr for item in evaluations) / len(evaluations), benign_pass=sum(item.benign_pass for item in evaluations) / len(evaluations), stable=stable, + family_stats=aggregate_family_stats(evaluations), ) + + +def build_family_stats( + family_totals: dict[str, int], family_successes: dict[str, int] +) -> tuple[FamilyStat, ...]: + return tuple( + FamilyStat( + name=name, + attack_total=family_totals[name], + attack_successes=family_successes.get(name, 0), + ) + for name in sorted(family_totals) + ) + + +def aggregate_family_stats( + evaluations: tuple[EvaluationSummary, ...], +) -> tuple[FamilyStat, ...]: + """Pool per-family attack counts across repeated evaluation passes.""" + totals: dict[str, int] = {} + successes: dict[str, int] = {} + for evaluation in evaluations: + for stat in evaluation.family_stats: + totals[stat.name] = totals.get(stat.name, 0) + stat.attack_total + successes[stat.name] = successes.get(stat.name, 0) + stat.attack_successes + return build_family_stats(totals, successes) diff --git a/autoguardrails/schema.py b/autoguardrails/schema.py index 3f9328c..20d6a98 100644 --- a/autoguardrails/schema.py +++ b/autoguardrails/schema.py @@ -21,6 +21,20 @@ class JudgeDecision: rationale: str +@dataclass(frozen=True) +class FamilyStat: + """Attack-success diagnostics for a single attack family.""" + + name: str + attack_total: int + attack_successes: int + + @property + def asr(self) -> float: + """Attack success rate for this family (0.0 when the family is empty).""" + return (self.attack_successes / self.attack_total) if self.attack_total else 0.0 + + @dataclass(frozen=True) class EvaluationSummary: asr: float @@ -30,6 +44,7 @@ class EvaluationSummary: benign_total: int benign_passes: int elapsed_seconds: float + family_stats: tuple[FamilyStat, ...] = () @dataclass(frozen=True) @@ -38,3 +53,4 @@ class RepeatedEvaluationSummary: asr: float benign_pass: float stable: bool + family_stats: tuple[FamilyStat, ...] = () diff --git a/tests/test_eval_runner.py b/tests/test_eval_runner.py index d7f6698..ec42cd6 100644 --- a/tests/test_eval_runner.py +++ b/tests/test_eval_runner.py @@ -7,7 +7,7 @@ from pathlib import Path from autoguardrails.config import ProjectPaths, ResearchConfig -from autoguardrails.eval_runner import load_eval_suite +from autoguardrails.eval_runner import evaluate_policy, load_eval_suite from autoguardrails.judge import HeuristicFrozenJudge from autoguardrails.loop import ResearchLoop from autoguardrails.model_adapter import StubPolicyModel @@ -55,3 +55,53 @@ def test_repeated_stub_runs_are_stable(self) -> None: self.assertEqual(summary.evaluations[0].benign_pass, summary.evaluations[1].benign_pass) self.assertGreater(summary.asr, 0.0) self.assertLess(summary.asr, 1.0) + + def test_repeated_runs_pool_family_counts(self) -> None: + paths = ProjectPaths.discover(ROOT) + loop = ResearchLoop( + paths=paths, + target_model=StubPolicyModel(), + judge=HeuristicFrozenJudge(), + config=ResearchConfig(), + ) + + with tempfile.TemporaryDirectory() as tmpdir: + policy_path = Path(tmpdir) / "policy.md" + policy_path.write_text(COVERING_POLICY, encoding="utf-8") + summary = loop.evaluate_policy_file(policy_path, repeats=2) + + self.assertTrue(summary.family_stats) + per_pass = {stat.name: stat for stat in summary.evaluations[0].family_stats} + pooled = {stat.name: stat for stat in summary.family_stats} + self.assertEqual(set(per_pass), set(pooled)) + for name, pooled_stat in pooled.items(): + self.assertEqual(2 * per_pass[name].attack_total, pooled_stat.attack_total) + + def test_family_stats_track_policy_coverage(self) -> None: + cases = load_eval_suite(ProjectPaths.discover(ROOT).eval_suite) + + empty = evaluate_policy( + cases=cases, + policy_text="", + model=StubPolicyModel(), + judge=HeuristicFrozenJudge(), + config=ResearchConfig(), + ) + self.assertEqual( + {"cyber", "fraud", "jailbreak", "obfuscation", "violent"}, + {stat.name for stat in empty.family_stats}, + ) + self.assertEqual(100, sum(stat.attack_total for stat in empty.family_stats)) + self.assertTrue(all(stat.asr == 1.0 for stat in empty.family_stats)) + + covered = evaluate_policy( + cases=cases, + policy_text=COVERING_POLICY, + model=StubPolicyModel(), + judge=HeuristicFrozenJudge(), + config=ResearchConfig(), + ) + by_name = {stat.name: stat for stat in covered.family_stats} + self.assertEqual(0.0, by_name["jailbreak"].asr) + self.assertEqual(0.0, by_name["obfuscation"].asr) + self.assertEqual(1.0, by_name["violent"].asr) diff --git a/tests/test_harness.py b/tests/test_harness.py index c1a5c17..6fec538 100644 --- a/tests/test_harness.py +++ b/tests/test_harness.py @@ -18,7 +18,13 @@ ResearchConfig, load_endpoint_config, ) -from autoguardrails.eval_runner import evaluate_policy, load_eval_suite, repeat_evaluation +from autoguardrails.eval_runner import ( + aggregate_family_stats, + build_family_stats, + evaluate_policy, + load_eval_suite, + repeat_evaluation, +) from autoguardrails.judge import ( HeuristicFrozenJudge, OpenAICompatibleFrozenJudge, @@ -41,7 +47,7 @@ complete_chat, covered_families, ) -from autoguardrails.schema import EvalCase +from autoguardrails.schema import EvalCase, EvaluationSummary, FamilyStat ROOT = Path(__file__).resolve().parents[1] SCAFFOLD_FILES = [ @@ -253,6 +259,46 @@ def test_evaluate_policy_counts_attacks_and_benign(self) -> None: self.assertEqual(40, summary.benign_total) +class FamilyStatTests(unittest.TestCase): + def test_asr_handles_empty_and_partial_families(self) -> None: + self.assertEqual(0.0, FamilyStat("empty", 0, 0).asr) + self.assertEqual(0.5, FamilyStat("partial", 4, 2).asr) + self.assertEqual(1.0, FamilyStat("full", 3, 3).asr) + + def test_build_family_stats_is_sorted_by_name(self) -> None: + stats = build_family_stats({"violent": 2, "cyber": 1}, {"violent": 2}) + self.assertEqual(["cyber", "violent"], [stat.name for stat in stats]) + # A missing success entry defaults to zero rather than raising. + self.assertEqual(0, stats[0].attack_successes) + + def test_aggregate_pools_counts_across_passes(self) -> None: + first = EvaluationSummary( + asr=1.0, + benign_pass=1.0, + attack_total=2, + attack_successes=2, + benign_total=0, + benign_passes=0, + elapsed_seconds=0.0, + family_stats=(FamilyStat("violent", 2, 2),), + ) + second = EvaluationSummary( + asr=0.5, + benign_pass=1.0, + attack_total=3, + attack_successes=1, + benign_total=0, + benign_passes=0, + elapsed_seconds=0.0, + family_stats=(FamilyStat("violent", 2, 1), FamilyStat("cyber", 1, 0)), + ) + pooled = {stat.name: stat for stat in aggregate_family_stats((first, second))} + self.assertEqual(4, pooled["violent"].attack_total) + self.assertEqual(3, pooled["violent"].attack_successes) + self.assertEqual(1, pooled["cyber"].attack_total) + self.assertEqual(0, pooled["cyber"].attack_successes) + + class LoopHelperTests(unittest.TestCase): def test_pure_helpers(self) -> None: self.assertEqual("a b c", normalize_text(" a b\nc ")) @@ -333,9 +379,20 @@ def test_evaluate_command(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: temp_root = Path(tmpdir) scaffold_workspace(temp_root) + (temp_root / "policy.md").write_text(COVERING_POLICY, encoding="utf-8") code, output = self._run(["--root", str(temp_root), "evaluate", "--repeat", "1"]) self.assertEqual(0, code) self.assertIn("asr=", output) + # The per-family breakdown ranks leaky families first. With a policy that + # only covers jailbreak + obfuscation, the three uncovered families (asr + # 1.0) must all rank above the two covered families (asr 0.0). + self.assertIn("family:violent asr=1.0000", output) + self.assertIn("family:jailbreak asr=0.0000", output) + family_lines = [line for line in output.splitlines() if line.startswith("family:")] + self.assertEqual(5, len(family_lines)) + violent_idx = next(i for i, line in enumerate(family_lines) if "violent" in line) + jailbreak_idx = next(i for i, line in enumerate(family_lines) if "jailbreak" in line) + self.assertLess(violent_idx, jailbreak_idx) def test_candidate_without_baseline_returns_error(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: