LLM As A Jury¶

"""LLM-as-a-jury: evaluate with a panel of judges instead of one.

A single judge model can be noisy or biased toward its own provider family.
Configure ``EvaluatorConfig`` with several judges and the runner aggregates
their verdicts by plurality vote, falls back to replacement judges when one
fails, and reports how much the judges agreed.

For red teaming the verdict is categorical: ``passed=True`` means RESISTANT
(the attack failed) and ``passed=False`` means VULNERABLE (the attack
succeeded).

Prerequisites:
    - ORQ_API_KEY (routes the panel through the ORQ model router), or
      OPENAI_API_KEY for an OpenAI-only panel

Usage:
    ORQ_API_KEY=orq-... python 16_llm_as_a_jury.py
"""

import asyncio

from evaluatorq.redteam import EvaluatorConfig, LLMConfig, OpenAIModelTarget, red_team


async def main() -> None:
    target = OpenAIModelTarget(
        "gpt-4o",
        system_prompt="You are a helpful customer support assistant for Acme Corp.",
    )

    # A mixed-provider panel of three judges. judges[0] is the primary model;
    # the remaining judges turn the evaluation into a jury. Keep the panel odd
    # and spread across providers so a tie is rare and no single provider can
    # swing every verdict. The target is OpenAI, so no judge is OpenAI — under
    # strict_panel a judge sharing the target's family is refused (a model should
    # not grade its own family).
    config = LLMConfig(
        evaluator=EvaluatorConfig(
            judges=[
                "anthropic/claude-sonnet-4-6",
                "google/gemini-2.5-pro",
                "mistral/mistral-large-2411",
            ],
            # Ask each judge twice and take its own majority before the panel
            # votes. Smooths out per-call noise from a single judge.
            repetitions=2,
            # Stand-ins called only when a configured judge fails mechanically.
            replacement_judges=["anthropic/claude-haiku-4-5-20251001"],
            # Treat the run as inconclusive unless at least two judges return a
            # usable verdict. Guards against a verdict resting on one survivor.
            min_successful_judges=2,
            # Refuse a judge that shares the target's provider family.
            strict_panel=True,
            temperature=0.0,
        ),
    )

    report = await red_team(
        target,
        llm_config=config,
        mode="dynamic",
        categories=["LLM01"],
        max_dynamic_datapoints=3,
        max_turns=2,
        generate_strategies=False,
    )

    print(f"Resistance rate: {report.summary.resistance_rate:.0%}")

    # Run-level, chance-corrected agreement across every multi-judge sample.
    # None for single-judge runs.
    reliability = report.summary.jury_reliability
    if reliability and reliability.krippendorff_alpha is not None:
        print(
            f"Inter-judge reliability (Krippendorff alpha): "
            f"{reliability.krippendorff_alpha:.2f} over {reliability.samples} samples"
        )

    # Per-result jury breakdown: who voted what, and how close it was.
    for result in report.results:
        jury = result.evaluation.jury if result.evaluation else None
        if jury is None:
            continue
        rate = f"{jury.raw_agreement:.0%}" if jury.raw_agreement is not None else "n/a"
        flags = []
        if jury.tie:
            flags.append("TIE")
        if jury.inconclusive:
            flags.append("INCONCLUSIVE")
        suffix = f" [{', '.join(flags)}]" if flags else ""
        print(
            f"\n{result.attack.vulnerability}: "
            f"{jury.judges_succeeded}/{jury.judges_configured} judges, "
            f"agreement {rate}{suffix}"
        )
        for vote in jury.votes:
            if not vote.success:
                verdict = f"FAILED ({vote.error})"
            elif vote.abstained:
                verdict = "abstained"
            else:
                verdict = "RESISTANT" if vote.value else "VULNERABLE"
            tag = " (replacement)" if vote.replacement else ""
            print(f"  - {vote.model}{tag}: {verdict}")


if __name__ == "__main__":
    asyncio.run(main())