Add claim-support benchmark for transcript-grounded scoring

docs/_builtin-scorers.md

@@ -20,6 +20,9 @@ Inspect includes both text matching scorers as well as model graded scorers. Bel
 `model_graded_fact()`
 :   Like `model_graded_qa()` but narrower: have another model assess whether the output contains the fact set out in `target`. Use it when the output is too complex to assess with `match()` or `pattern()`. See [Model Grading](model-graded.qmd).
 
+`claim_support()`
+:   Judge whether a claim is supported by the full transcript and tool trace. Use `decompose_claims=True` to split compound claims into atomic subclaims before verification.
+
 `exact()`
 :   Normalize the answer and target(s) and require the whole output to match one or more targets exactly, returning `CORRECT` on a match. Reports `mean` and `stderr` metrics.
 

examples/security_simulation.py

@@ -0,0 +1,26 @@
+from inspect_ai import Task, task
+from inspect_ai.dataset import Sample
+from inspect_ai.scorer import match
+from inspect_ai.solver import generate, system_message
+
+SYSTEM_MESSAGE = """
+You are a security analyst helping to triage a simulated incident.
+Answer exactly Yes or No.
+"""
+
+
+@task
+def security_simulation():
+    return Task(
+        dataset=[
+            Sample(
+                input=(
+                    "A web server saw 500 login attempts from one IP in one minute. "
+                    "Is this suspicious?"
+                ),
+                target="Yes",
+            )
+        ],
+        solver=[system_message(SYSTEM_MESSAGE), generate()],
+        scorer=match(),
+    )

src/inspect_ai/scorer/__init__.py

@@ -27,7 +27,7 @@
 from ._metrics.mean import mean
 from ._metrics.perplexity import perplexity_per_seq, perplexity_per_token
 from ._metrics.std import bootstrap_stderr, std, stderr, var
-from ._model import model_graded_fact, model_graded_qa
+from ._model import claim_support, model_graded_fact, model_graded_qa
 from ._multi import multi_scorer
 from ._pattern import pattern
 from ._perplexity import perplexity
@@ -85,6 +85,7 @@
     "median_score",
     "metric",
     "mode_score",
+    "claim_support",
     "model_graded_fact",
     "model_graded_qa",
     "multi_scorer",

src/inspect_ai/scorer/_model.py

@@ -18,7 +18,7 @@
 from inspect_ai.solver._task_state import TaskState
 from inspect_ai.util import resource
 
-from ._metric import Score
+from ._metric import CORRECT, INCORRECT, PARTIAL, Score
 from ._metrics import accuracy, stderr
 from ._multi import multi_scorer
 from ._scorer import Scorer, scorer
@@ -83,6 +83,40 @@ def model_graded_fact(
     )
 
 
+@scorer(metrics=[accuracy(), stderr()])
+def claim_support(
+    template: str | None = None,
+    instructions: str | None = None,
+    grade_pattern: str | None = None,
+    include_history: bool | Callable[[TaskState], str] = True,
+    partial_credit: bool = False,
+    decompose_claims: bool = False,
+    model: list[str | Model] | str | Model | None = None,
+    model_role: str | None = "grader",
+) -> Scorer:
+    """Score whether an assistant claim is supported by the transcript.
+
+    Set ``decompose_claims=True`` to split compound claims into atomic
+    subclaims before verification, which is often a better fit for
+    evidence-grounded or security-trace style evaluations.
+    """
+    get_scorer = partial(
+        _claim_support_single,
+        template,
+        instructions,
+        grade_pattern,
+        include_history,
+        partial_credit,
+        decompose_claims,
+        model_role=model_role,
+    )
+    if model is None or not isinstance(model, list):
+        return get_scorer(model)
+
+    scorers = [get_scorer(model) for model in model]
+    return multi_scorer(scorers, "mode")
+
+
 @scorer(metrics=[accuracy(), stderr()])
 def model_graded_qa(
     template: str | None = None,
@@ -242,6 +276,141 @@ async def score(state: TaskState, target: Target) -> Score:
     return score
 
 
+@scorer(metrics=[accuracy(), stderr()])
+def _claim_support_single(
+    template: str | None = None,
+    instructions: str | None = None,
+    grade_pattern: str | None = None,
+    include_history: bool | Callable[[TaskState], str] = True,
+    partial_credit: bool = False,
+    decompose_claims: bool = False,
+    model: str | Model | None = None,
+    model_role: str | None = "grader",
+) -> Scorer:
+    # returns a scorer that judges whether the claim is supported by the transcript
+
+    template = template if template else DEFAULT_CLAIM_SUPPORT_TEMPLATE
+    grading_template = resource(template)
+    instructions = (
+        instructions if instructions else default_instructions(partial_credit)
+    )
+
+    async def score(state: TaskState, target: Target) -> Score:
+        nonlocal model
+        if model is not None:
+            model = model if isinstance(model, Model) else get_model(model)
+        elif model_role is not None:
+            model = get_model(role=model_role)
+        else:
+            model = get_model()
+
+        metadata = omit(
+            state.metadata, ["question", "answer", "criterion", "instructions"]
+        )
+
+        if include_history is True:
+            question = chat_history(state)
+        elif callable(include_history):
+            question = include_history(state)
+        else:
+            question = state.input_text
+
+        criterion = target.text if target.text else DEFAULT_CLAIM_SUPPORT_CRITERION
+
+        if decompose_claims:
+            criterion = (
+                target.text
+                if target.text
+                else DEFAULT_CLAIM_ATOMIC_CLAIM_CRITERION
+            )
+            atomic_claims = await decompose_claims_for_verification(
+                model=model,
+                claim=state.output.completion,
+            )
+            verdicts: list[str] = []
+            explanation_parts: list[str] = []
+            grading: list[ChatMessage] = []
+
+            for atomic_claim in atomic_claims:
+                scoring_prompt = model_scoring_prompt(
+                    template=grading_template,
+                    question=question,
+                    output=ModelOutput.from_content(
+                        model=model.model_name, content=atomic_claim
+                    ),
+                    criterion=criterion,
+                    instructions=instructions,
+                    metadata=metadata,
+                )
+                result = await model.generate([scoring_prompt])
+                grading.extend([scoring_prompt, result.message])
+                match = re.search(
+                    grade_pattern or DEFAULT_GRADE_PATTERN, result.completion
+                )
+                if match:
+                    value = match.group(1)
+                    if grade_pattern is None:
+                        value = value.upper()
+                else:
+                    value = INCORRECT
+                verdicts.append(value)
+                explanation_parts.append(f"{atomic_claim} => {value}")
+
+            value = aggregate_atomic_verdicts(verdicts)
+            return Score(
+                value=value,
+                answer=state.output.completion,
+                explanation="\n".join(explanation_parts),
+                metadata=dict(
+                    grading=grading,
+                    atomic_claims=atomic_claims,
+                    atomic_verdicts=verdicts,
+                ),
+            )
+
+        scoring_prompt = model_scoring_prompt(
+            template=grading_template,
+            question=question,
+            output=state.output,
+            criterion=criterion,
+            instructions=instructions,
+            metadata=metadata,
+        )
+
+        result = await model.generate([scoring_prompt])
+        match = re.search(grade_pattern or DEFAULT_GRADE_PATTERN, result.completion)
+        if match:
+            value = match.group(1)
+            if grade_pattern is None:
+                value = value.upper()
+            return Score(
+                value=value,
+                answer=state.output.completion,
+                explanation=result.completion,
+                metadata=dict(
+                    grading=[
+                        scoring_prompt,
+                        result.message,
+                    ]
+                ),
+            )
+        else:
+            return Score(
+                value=INCORRECT,
+                answer=state.output.completion,
+                explanation="Grade not found in model output: "
+                + f"{result.completion}",
+                metadata=dict(
+                    grading=[
+                        scoring_prompt,
+                        result.message,
+                    ]
+                ),
+            )
+
+    return score
+
+
 # these templates are based on the openai closedqa templates here:
 # https://github.com/openai/evals/blob/main/evals/registry/modelgraded/closedqa.yaml
 DEFAULT_MODEL_GRADED_QA_TEMPLATE = """
@@ -283,6 +452,62 @@ async def score(state: TaskState, target: Target) -> Score:
 """
 
 
+DEFAULT_CLAIM_SUPPORT_TEMPLATE = """
+You are assessing whether a submitted claim is supported by a transcript of a
+task. Here is the data:
+
+[BEGIN DATA]
+************
+[Transcript]: {question}
+************
+[Claim]: {answer}
+************
+[Support criterion]: {criterion}
+************
+[END DATA]
+
+Does the transcript and the recorded tool activity support the claim?
+
+{instructions}
+"""
+
+
+DEFAULT_CLAIM_DECOMPOSITION_TEMPLATE = """
+You are decomposing a claim into atomic subclaims for verification against a
+transcript. Here is the data:
+
+[BEGIN DATA]
+************
+[Claim]: {question}
+************
+[END DATA]
+
+Split the claim into the smallest self-contained claims that can each be
+verified independently. Return one claim per line. Do not add bullets, numbers,
+or any extra commentary.
+
+{instructions}
+"""
+
+
+DEFAULT_CLAIM_SUPPORT_CRITERION = """
+Use only evidence that appears in the transcript and tool events.
+If the transcript clearly supports the claim, grade C.
+If the transcript partially supports the claim or leaves some parts unproven,
+grade P.
+If the transcript contradicts or does not support the claim, grade I.
+"""
+
+
+DEFAULT_CLAIM_ATOMIC_CLAIM_CRITERION = """
+Use only evidence that appears in the transcript and tool events.
+Judge this atomic claim independently.
+If the transcript clearly supports the atomic claim, grade C.
+If the transcript leaves the atomic claim unproven, grade P.
+If the transcript contradicts the atomic claim, grade I.
+"""
+
+
 def default_instructions(partial_credit: bool) -> str:
     partial_letter = "P" if partial_credit else ""
     partial_prompt = '"P" for partially correct answers,' if partial_credit else ""
@@ -295,6 +520,46 @@ def default_instructions(partial_credit: bool) -> str:
 """
 
 
+def aggregate_atomic_verdicts(verdicts: list[str]) -> str:
+    if not verdicts:
+        return INCORRECT
+    if all(verdict == CORRECT for verdict in verdicts):
+        return CORRECT
+    if any(verdict == PARTIAL for verdict in verdicts):
+        return PARTIAL
+    if any(verdict == CORRECT for verdict in verdicts):
+        return PARTIAL
+    return INCORRECT
+
+
+async def decompose_claims_for_verification(
+    *,
+    model: Model,
+    claim: str,
+) -> list[str]:
+    prompt = ChatMessageUser(
+        content=DEFAULT_CLAIM_DECOMPOSITION_TEMPLATE.format(
+            question=neutralize_structural_delimiters(claim),
+            instructions="",
+        )
+    )
+    result = await model.generate([prompt])
+    claims = parse_atomic_claims(result.completion)
+    return claims or [claim]
+
+
+def parse_atomic_claims(text: str) -> list[str]:
+    claims: list[str] = []
+    for line in text.splitlines():
+        claim = line.strip()
+        if not claim:
+            continue
+        claim = re.sub(r"^\s*(?:[-*•]|\d+[.)])\s*", "", claim)
+        if claim:
+            claims.append(claim)
+    return claims
+
+
 # Whitespace plus zero-width / formatting marks that can appear around a
 # verdict separator in model output or pasted text.
 _GRADE_SPACING = r"[\s\u200b\u200c\u200d\u200e\u200f\u2060\u2063\ufeff]*"