Evaluation Prompt Engineering: Designing Reliable LLM Judges › Code Implementation

import os

# When ANTHROPIC_API_KEY is not set, the code falls back to simulated
# responses so the notebook executes cleanly in CI and offline environments.
_API_KEY_AVAILABLE = bool(os.environ.get("ANTHROPIC_API_KEY"))

# Simulated judge responses keyed by question fragment for reproducibility
_MOCK_RESPONSES = {
    "capital of Australia": (
        4,
        "Canberra is the correct capital of Australia.",
    ),
    "capital": (1, "Sydney is not the capital of Australia; Canberra is."),
    "Hamlet": (4, "William Shakespeare wrote Hamlet in the early 1600s."),
    "bones": (2, "The adult human body has 206 bones, not 250."),
    "atomic number 79": (4, "Gold (Au) has atomic number 79."),
    "Sun to Earth": (
        4,
        "Light takes approximately 8 minutes 20 seconds to travel from the Sun to Earth.",
    ),
    "carbon-14": (
        4,
        "Carbon-14 has a half-life of 5,730 years, widely used in radiocarbon dating.",
    ),
}

from dataclasses import dataclass, field


@dataclass
class EvalCriterion:
    name: str
    description: str
    scale_anchors: dict[int, str]  # e.g. {1: "Wrong", 4: "Correct"}


@dataclass
class FewShotExample:
    question: str
    response: str
    reasoning: str
    score: int


@dataclass
class EvalPromptTemplate:
    role_description: str
    criterion: EvalCriterion
    examples: list[FewShotExample] = field(default_factory=list)
    use_chain_of_thought: bool = True

    def format_scale(self) -> str:
        lines = []
        for score, label in sorted(self.criterion.scale_anchors.items()):
            lines.append(f"  {score}: {label}")
        return "\n".join(lines)

    def format_example(self, ex: FewShotExample) -> str:
        return (
            f"Question: {ex.question}\n"
            f"Response: {ex.response}\n"
            f"Reasoning: {ex.reasoning}\n"
            f"SCORE: {ex.score}"
        )

    def assemble(self, question: str, response: str) -> str:
        scale_text = self.format_scale()
        cot_instruction = (
            "Before giving your score, write a one-sentence reasoning trace.\n"
            if self.use_chain_of_thought
            else ""
        )
        system = (
            f"{self.role_description}\n\n"
            f"You will evaluate: {self.criterion.name}\n"
            f"Definition: {self.criterion.description}\n\n"
            f"Scale:\n{scale_text}\n\n"
            f"{cot_instruction}"
            f"Format your output as:\nReasoning: [one sentence]\nSCORE: [number]"
        )
        user_parts = []
        for ex in self.examples:
            user_parts.append(f"--- EXAMPLE ---\n{self.format_example(ex)}\n")
        user_parts.append(
            f"--- EVALUATE ---\nQuestion: {question}\nResponse: {response}"
        )
        return system, "\n".join(user_parts)

import random
import re
from dataclasses import dataclass
from typing import Optional


@dataclass
class JudgeVerdict:
    score: Optional[int]
    reasoning: str
    raw_output: str
    parse_success: bool


def parse_verdict(raw: str) -> JudgeVerdict:
    """Extract reasoning and score from chain-of-thought output."""
    reasoning_match = re.search(
        r"Reasoning:\s*(.+?)(?=SCORE:|$)", raw, re.DOTALL
    )
    score_match = re.search(r"SCORE:\s*(\d+)", raw)

    reasoning = reasoning_match.group(1).strip() if reasoning_match else ""
    score = int(score_match.group(1)) if score_match else None
    parse_success = score is not None

    return JudgeVerdict(
        score=score,
        reasoning=reasoning,
        raw_output=raw,
        parse_success=parse_success,
    )


def _mock_verdict(question: str, temperature: float) -> JudgeVerdict:
    """Return a plausible mock verdict for offline / CI execution."""
    for key, (score, reasoning) in _MOCK_RESPONSES.items():
        if key.lower() in question.lower():
            # At temperature > 0, introduce slight variance around the correct score
            if temperature > 0 and random.random() < 0.25:
                score = max(1, min(4, score + random.choice([-1, 1])))
            raw = f"Reasoning: {reasoning}\nSCORE: {score}"
            return parse_verdict(raw)
    # Default: score 3 with generic reasoning
    raw = "Reasoning: The response is mostly correct.\nSCORE: 3"
    return parse_verdict(raw)


def run_judge(
    template: EvalPromptTemplate,
    question: str,
    response: str,
    model: str = "claude-haiku-4-5-20251001",
    temperature: float = 0.0,
    n_samples: int = 1,
) -> list[JudgeVerdict]:
    if not _API_KEY_AVAILABLE:
        return [_mock_verdict(question, temperature) for _ in range(n_samples)]

    client = anthropic.Anthropic()
    system_prompt, user_message = template.assemble(question, response)

    verdicts = []
    for _ in range(n_samples):
        result = client.messages.create(
            model=model,
            max_tokens=256,
            temperature=temperature,
            system=system_prompt,
            messages=[{"role": "user", "content": user_message}],
        )
        raw = result.content[0].text
        verdicts.append(parse_verdict(raw))
    return verdicts

factual_accuracy_criterion = EvalCriterion(
    name="Factual Accuracy",
    description=(
        "Does the response answer the question correctly? "
        "Evaluate only factual correctness, not fluency or tone."
    ),
    scale_anchors={
        1: "Wrong. The response is factually incorrect or fabricated.",
        2: "Partially correct. A significant error affects the core answer.",
        3: "Mostly correct. Minor imprecision but core answer is right.",
        4: "Correct. The response answers the question accurately.",
    },
)

examples = [
    FewShotExample(
        question="In what year did the Berlin Wall fall?",
        response="The Berlin Wall fell in 1989.",
        reasoning="The response correctly identifies 1989 as the year of the fall.",
        score=4,
    ),
    FewShotExample(
        question="What is the speed of light in a vacuum?",
        response="The speed of light is approximately 300,000 miles per second.",
        reasoning=(
            "The speed of light is ~300,000 km/s, not miles/s. "
            "The magnitude is correct but the unit is wrong."
        ),
        score=2,
    ),
]

template = EvalPromptTemplate(
    role_description=(
        "You are an expert fact-checker evaluating language model responses "
        "to factual knowledge questions."
    ),
    criterion=factual_accuracy_criterion,
    examples=examples,
    use_chain_of_thought=True,
)

test_cases = [
    {
        "question": "What is the capital of Australia?",
        "response": "The capital of Australia is Canberra.",
        "expected": 4,
    },
    {
        "question": "What is the capital of Australia?",
        "response": "The capital of Australia is Sydney.",
        "expected": 1,
    },
    {
        "question": "Who wrote Hamlet?",
        "response": "Hamlet was written by William Shakespeare in the early 1600s.",
        "expected": 4,
    },
    {
        "question": "How many bones are in the adult human body?",
        "response": "The adult human body has approximately 250 bones.",
        "expected": 2,
    },
    {
        "question": "What element has atomic number 79?",
        "response": "Atomic number 79 corresponds to gold, symbol Au.",
        "expected": 4,
    },
]

results = []
for case in test_cases:
    verdicts = run_judge(
        template=template,
        question=case["question"],
        response=case["response"],
        temperature=0.0,
        n_samples=1,
    )
    v = verdicts[0]
    results.append(
        {
            "question": case["question"],
            "response": case["response"],
            "expected": case["expected"],
            "predicted": v.score,
            "reasoning": v.reasoning,
            "parse_success": v.parse_success,
        }
    )

# Borderline case: response is mostly correct but includes an imprecision
borderline_case = {
    "question": "How long does light take to travel from the Sun to Earth?",
    "response": "Light from the Sun takes about 8 minutes to reach Earth.",
}

# Run 5 times to measure consistency
n_samples = 5
consistency_verdicts = run_judge(
    template=template,
    question=borderline_case["question"],
    response=borderline_case["response"],
    temperature=0.2,  # Non-zero to observe variance
    n_samples=n_samples,
)

scores = [v.score for v in consistency_verdicts if v.score is not None]

def audit_prompt_sensitivity(
    base_template: EvalPromptTemplate,
    paraphrase_role_descriptions: list[str],
    question: str,
    response: str,
    model: str = "claude-haiku-4-5-20251001",
) -> dict:
    """
    Run the same question/response pair through multiple prompt variants
    and return agreement statistics.
    """
    all_scores = []
    for role_desc in paraphrase_role_descriptions:
        variant = EvalPromptTemplate(
            role_description=role_desc,
            criterion=base_template.criterion,
            examples=base_template.examples,
            use_chain_of_thought=base_template.use_chain_of_thought,
        )
        verdicts = run_judge(variant, question, response, model=model)
        if verdicts[0].parse_success:
            all_scores.append(verdicts[0].score)

    if len(all_scores) < 2:
        return {"scores": all_scores, "agreement": None, "max_range": None}

    all_agree = len(set(all_scores)) == 1
    score_range = max(all_scores) - min(all_scores)
    return {
        "scores": all_scores,
        "all_agree": all_agree,
        "max_range": score_range,
        "verdict": "ROBUST" if score_range <= 1 else "SENSITIVE",
    }


# Example sensitivity audit with three paraphrase variants
paraphrase_variants = [
    "You are an expert fact-checker evaluating language model responses to factual knowledge questions.",
    "You are a domain expert assessing the accuracy of AI-generated answers to knowledge questions.",
    "You are reviewing AI responses for factual correctness.",
]

audit_question = "What is the half-life of carbon-14?"
audit_response = "Carbon-14 has a half-life of approximately 5,730 years, which is why it is used for radiocarbon dating."