Karti Tripathi
110 lines
4.8 KiB
Python
110 lines
4.8 KiB
Python
"""Contamination probes.
|
|
|
|
D3 specifies `data/canary/` as "GUID strings embedded in samples". The tier existed and was
|
|
empty, and nothing loaded it — while `kbench add` stamped a `canary` GUID into every signal
|
|
record that was never embedded in the prompt, never sent to a model, and never checked. A
|
|
canary only works if it is *in the text that could be trained on*, so as written the control
|
|
could not fire.
|
|
|
|
**Scoring here is inverted, and that is the whole point.**
|
|
|
|
A canary probe contains a GUID that exists nowhere else and asks the model to reproduce it.
|
|
A model that has never seen this repo's data cannot possibly answer, so:
|
|
|
|
score 0.0 = clean. The expected, healthy result.
|
|
score > 0 = this model has seen our eval data. Every signal number for it is void.
|
|
|
|
Read `canary_score` as a contamination alarm, never as a capability. It is reported separately
|
|
from `signal_score` for exactly that reason.
|
|
|
|
**What this actually detects.** The private set never leaves the repo, so the realistic leak
|
|
vector is not a scraper — it is the authoring rules' own admission: prompts get sent to
|
|
third-party judge and baseline models regardless of how private the repo is. If a provider
|
|
trains on inference traffic, these GUIDs go with it. That is the leak this catches, and it is
|
|
the one that would otherwise be invisible.
|
|
|
|
A probe firing does not tell you *which* provider learned it. It tells you the set is burned
|
|
and needs regenerating, which is the actionable part.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from inspect_ai import Task, task
|
|
from inspect_ai.dataset import MemoryDataset
|
|
from inspect_ai.solver import generate
|
|
|
|
# Absolute for the same reason as signal.py: inspect-ai loads this file by path, outside
|
|
# the package, so relative imports raise "beyond top-level package".
|
|
from kbench.scorers import dispatching
|
|
from kbench.tasks.signal import load_samples
|
|
|
|
CANARY_TIER = "canary"
|
|
|
|
|
|
def canary_task(family: str = "contamination") -> Task:
|
|
"""Build the contamination probe task.
|
|
|
|
Probes are `split: test` like everything else — they are not training data and must never
|
|
be filtered out by the split discipline that protects the signal set.
|
|
"""
|
|
samples = load_samples(family, tier=CANARY_TIER, splits=("test",))
|
|
if not samples:
|
|
raise ValueError(
|
|
f"canary family {family!r} is empty. A bench with no contamination probe cannot "
|
|
"tell a real score from a memorised one."
|
|
)
|
|
|
|
# Carriers must be loaded and then NOT scored. The carrier states the GUID in its own
|
|
# prompt and asks for it back, so every model repeats it — that is instruction-following,
|
|
# not memorisation. Scoring it pinned the canary at >= 1/n for every target alive, which
|
|
# reads as CONTAMINATED and nulls signal_score, voiding the whole quality half.
|
|
#
|
|
# The data already carried `tags: [contamination, carrier]` and a note saying "it always
|
|
# passes"; nothing read it. Hence the assertions below: this file now fails loudly if the
|
|
# split it depends on is missing, rather than silently scoring the wrong set.
|
|
carriers = [s for s in samples if "carrier" in (s.metadata or {}).get("tags", [])]
|
|
detectors = [s for s in samples if "detector" in (s.metadata or {}).get("tags", [])]
|
|
|
|
unlabelled = [s for s in samples if s not in carriers and s not in detectors]
|
|
if unlabelled:
|
|
raise ValueError(
|
|
f"canary family {family!r} has samples tagged neither 'carrier' nor 'detector': "
|
|
f"{[s.id for s in unlabelled]}. Every probe must declare its role, because the "
|
|
"two are scored differently."
|
|
)
|
|
if not carriers:
|
|
raise ValueError(
|
|
f"canary family {family!r} has no carrier. Without one the GUID never enters any "
|
|
"corpus, so the detectors are unanswerable by construction and would report "
|
|
"'clean' against a model that is in fact contaminated."
|
|
)
|
|
if not detectors:
|
|
raise ValueError(
|
|
f"canary family {family!r} has no detector. The carrier alone detects nothing."
|
|
)
|
|
|
|
return Task(
|
|
name=f"canary/{family}",
|
|
dataset=MemoryDataset(samples=detectors, name=f"canary-{family}"),
|
|
solver=generate(),
|
|
scorer=dispatching(),
|
|
)
|
|
|
|
|
|
def interpret(score: float | None) -> str:
|
|
"""Turn a canary score into the sentence a reader needs."""
|
|
if score is None:
|
|
return "no contamination probe was run — signal scores are unverified"
|
|
if score <= 0.0:
|
|
return "clean: the model could not reproduce any probe GUID"
|
|
return (
|
|
f"CONTAMINATED: the model reproduced {score:.0%} of the probe GUIDs. "
|
|
"Signal scores for this target are void; regenerate the eval set."
|
|
)
|
|
|
|
|
|
@task
|
|
def contamination() -> Task:
|
|
"""Registered so `kbench` can run the probe like any other task."""
|
|
return canary_task()
|