Wiring an Eval Pipeline to the Claude API

Wiring an Eval Pipeline to the Claude API

When output quality has no regex — e.g., “did it stick to the source?” — a second model grades against a rubric. This post grades groundedness: every factual claim in an answer traceable to retrieved context.

“Reply with JSON only” is a request, not a guarantee. A judge becomes a component when its verdict is a typed object the harness can fail on. For pipeline stages, test sets, and judge failure modes, see the previous post.

1 Structured outputs

The Claude API can constrain generation to a JSON schema. In Python, define the shape with a Pydantic model; messages.parse() returns a validated instance.

from typing import Literal

from anthropic import Anthropic
from pydantic import BaseModel, ConfigDict, ValidationError

client = Anthropic()

RUBRIC = """You are checking whether an ANSWER is fully supported by the CONTEXT.

CONTEXT:
{context}

ANSWER:
{answer}

Decompose the ANSWER into individual factual claims. A claim is supported only
if it is stated in or directly entailed by the CONTEXT. General knowledge does
not count as support. The verdict is "fail" if any claim is unsupported."""


class Verdict(BaseModel):
    """One judge's reading of one answer."""

    model_config = ConfigDict(extra="forbid")

    unsupported_claims: list[str]
    verdict: Literal["pass", "fail"]


def judge(context: str, answer: str) -> Verdict:
    """Score one answer for groundedness against its retrieved context."""
    try:
        response = client.messages.parse(
            model="claude-opus-5",
            max_tokens=16000,
            messages=[
                {"role": "user", "content": RUBRIC.format(context=context, answer=answer)}
            ],
            output_format=Verdict,
        )
    except ValidationError as exc:  # budget ran out mid-JSON: parse() raises
        raise RuntimeError(f"unparseable verdict: {exc}") from exc
    if response.parsed_output is None:  # refusal, or budget spent inside thinking
        raise RuntimeError(f"no verdict: {response.stop_reason}")
    return response.parsed_output

• Field order matters: unsupported_claims before verdict forces evidence before the label.
• max_tokens caps reasoning and reply from one budget. Budget exhaustion yields parsed_output is None or a ValidationError mid-JSON.

Example verdict:

Verdict(unsupported_claims=['Refunds can also be issued as store credit.'],
        verdict='fail')

2 Test set

Data: synthetic — six hand-written (context, answer) pairs standing in for a support assistant over help-centre articles.

• context: passage a retriever pulled from articles
• answer: assistant reply
• label: human verdict

Three pairs are grounded; three contain exactly one fabricated claim in the documentation’s vocabulary (e.g., an invented processing-time SLA). That failure is costly for customers and invisible to string matching.

Synthetic data fits because the target metric is judge–human agreement, which requires labelled cases with known verdicts.

CASES = [
    {
        "id": "refund-window-grounded",
        "context": "Refunds are available within 30 days of purchase. Once "
        "approved, the amount returns to the original payment method.",
        "answer": "You can request a refund within 30 days, and it goes back "
        "to the card you paid with.",
        "label": "pass",
    },
    {
        "id": "refund-window-invented-sla",
        "context": "Refunds are available within 30 days of purchase. Once "
        "approved, the amount returns to the original payment method.",
        "answer": "You can request a refund within 30 days. Refunds are "
        "processed within 3 business days.",
        "label": "fail",
    },
    # ...four more, three grounded and three ungrounded in total.
]

3 Constraints

• Six cases demonstrate machinery; calibration sets run to 50–200 cases and require periodic re-measurement as models update.
• This pipeline scores answers against context, not retrieval quality. An answer grounded in the wrong passage still passes; retrieval failure needs a separate scorer.

Mechanics are identical at six cases or two hundred; at scale, batch execution matters.

4 Batches API

Each judged case is an API call. The Batches API submits all cases as one job, polls until completion, and bills at half per-token price.

Each request carries a custom_id tying results back to cases:

import time

from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request

batch = client.messages.batches.create(
    requests=[
        Request(
            custom_id=case["id"],
            params=MessageCreateParamsNonStreaming(
                model="claude-opus-5",
                max_tokens=16000,
                messages=[
                    {
                        "role": "user",
                        "content": RUBRIC.format(
                            context=case["context"], answer=case["answer"]
                        ),
                    }
                ],
                output_config={
                    "format": {
                        "type": "json_schema",
                        "schema": Verdict.model_json_schema(),
                    }
                },
            ),
        )
        for case in CASES
    ]
)

while client.messages.batches.retrieve(batch.id).processing_status != "ended":
    time.sleep(30)

Batch path differences:

• Uses output_config.format with raw JSON schema (Verdict.model_json_schema()), not parse()’s output_format=.
• extra="forbid" on the Pydantic model sets additionalProperties: false, required by the API’s strict schema subset.

Results return in completion order, not submission order. Look up by custom_id:

verdicts = {}
for result in client.messages.batches.results(batch.id):
    if result.result.type == "errored":
        raise RuntimeError(f"{result.custom_id}: {result.result.error}")
    if result.result.type != "succeeded":
        raise RuntimeError(f"{result.custom_id}: {result.result.type}")
    message = result.result.message
    if message.stop_reason != "end_turn":
        raise RuntimeError(f"{result.custom_id}: stopped on {message.stop_reason}")
    text = next((b.text for b in message.content if b.type == "text"), None)
    if text is None:
        raise RuntimeError(f"{result.custom_id}: completed turn with no text block")
    verdicts[result.custom_id] = Verdict.model_validate_json(text)

labels = {case["id"]: case["label"] for case in CASES}
pass_rate = sum(v.verdict == "pass" for v in verdicts.values()) / len(verdicts)
agreement = sum(verdicts[i].verdict == labels[i] for i in labels) / len(labels)

succeeded means transport success, not usable content. Check stop_reason == "end_turn"; refusals and truncated replies can still arrive under succeeded.

Plausible output on six cases:

pass rate                   0.67
agreement with labels       0.83  (5/6)
disagreement                refund-window-invented-sla: judged pass, labelled fail

5 Metrics

Two ratios:

• Pass rate — fraction of answers the judge passed (about the system under test).
• Agreement with labels — fraction where judge verdict matched human label (about the judge).

Agreement determines whether pass rate is meaningful. A judge wrong one time in five can skew reported pass rate by up to twenty points. Disagreement on the invented processing-time case shows the pipeline has not yet caught vocabulary-matched hallucinations.

Typed verdicts make failures visible: refusals, truncated replies, and malformed JSON stop the run and name the case. Batch submission and the two ratios are ordinary code once the verdict has a type.

6 References

• How Prompt Evaluation Pipelines Work — criteria, test sets, judge calibration