In [ ]:

# Copyright (c) 2025 Microsoft Corporation.
import os
from pathlib import Path
from typing import cast

import pandas as pd
from pydantic import SecretStr
from rich import print as rich_print

from benchmark_qed.autoe.assertion import (
    HierarchicalMode,
    load_and_normalize_hierarchical_assertions,
    run_assertion_evaluation,
    run_hierarchical_assertion_evaluation,
)
from benchmark_qed.autoe.pairwise import analyze_criteria, get_pairwise_scores
from benchmark_qed.autoe.reference import (
    get_reference_scores,
    summarize_reference_scores,
)
from benchmark_qed.cli.utils import print_df
from benchmark_qed.config.llm_config import (
    LLMConfig,
    LLMProvider,
)
from benchmark_qed.config.model.score import (
    pairwise_scores_criteria,
    reference_scores_criteria,
)
from benchmark_qed.llm.factory import ModelFactory

# Copyright (c) 2025 Microsoft Corporation. import os from pathlib import Path from typing import cast import pandas as pd from pydantic import SecretStr from rich import print as rich_print from benchmark_qed.autoe.assertion import ( HierarchicalMode, load_and_normalize_hierarchical_assertions, run_assertion_evaluation, run_hierarchical_assertion_evaluation, ) from benchmark_qed.autoe.pairwise import analyze_criteria, get_pairwise_scores from benchmark_qed.autoe.reference import ( get_reference_scores, summarize_reference_scores, ) from benchmark_qed.cli.utils import print_df from benchmark_qed.config.llm_config import ( LLMConfig, LLMProvider, ) from benchmark_qed.config.model.score import ( pairwise_scores_criteria, reference_scores_criteria, ) from benchmark_qed.llm.factory import ModelFactory

AutoE¶

In [ ]:

import nest_asyncio

nest_asyncio.apply()

import nest_asyncio nest_asyncio.apply()

In [ ]:

%load_ext dotenv
%dotenv

%load_ext dotenv %dotenv

Pairwise Comparisons of RAG Methods¶

The AutoE component automates relative comparisons of RAG methods using the LLM-as-a-judge approach. It presents an LLM with pairs of answers, along with the query and target metric, in a counterbalanced order. The model then judges whether the first answer wins, loses, or ties with the second. Aggregating these judgments across multiple queries and trials yields win rates for each method.

In the example below, we compare Vector RAG with short context (retrieves 50 text chunks) against Vector RAG with long context (retrieves 200 text chunks). We use synthetic questions generated from AP News health-related articles using AutoQ, covering data-global, data-local, and data-linked question types. Each query is evaluated in 4 counterbalanced trials across four default metrics (comprehensiveness, diversity, empowerment, and relevance), using GPT-4.1 as the judge.

Choosing the right LLM judge is critical: less capable models may introduce biases and yield unreliable results. A useful first step in validating a judge model is to run an A/A test — comparing a RAG method against itself. This should result in a ~0.5 win rate with no statistically significant differences.

In [ ]:

# Config LLM model to be used as judge
llm_config = LLMConfig(
    model="gpt-5.2",
    api_key=SecretStr(os.environ["OPENAI_API_KEY"]),
    llm_provider=LLMProvider.OpenAIChat,
    concurrent_requests=32,
    call_args={"temperature": 0.0, "seed": 42},
)
llm_client = ModelFactory.create_chat_model(llm_config)

# Config LLM model to be used as judge llm_config = LLMConfig( model="gpt-5.2", api_key=SecretStr(os.environ["OPENAI_API_KEY"]), llm_provider=LLMProvider.OpenAIChat, concurrent_requests=32, call_args={"temperature": 0.0, "seed": 42}, ) llm_client = ModelFactory.create_chat_model(llm_config)

In [ ]:

# Config conditions for comparison
base = "vector_rag_short_context"
others = ["vector_rag_long_context"]
question_sets = ["data_global", "data_local", "data_linked"]
trials = 4  # number of trials to run for each combination of [query, base, other]. Trials must be an even number to support counterbalancing.
alpha = 0.05  # significance level used for statistical tests

input_dir = "./example_answers"
output_dir = Path("./output/win_rates")
if not output_dir.exists():
    output_dir.mkdir(parents=True)

# load default criteria. You can also define your own criteria as a list Criteria objects
criteria = pairwise_scores_criteria()

# Config conditions for comparison base = "vector_rag_short_context" others = ["vector_rag_long_context"] question_sets = ["data_global", "data_local", "data_linked"] trials = 4 # number of trials to run for each combination of [query, base, other]. Trials must be an even number to support counterbalancing. alpha = 0.05 # significance level used for statistical tests input_dir = "./example_answers" output_dir = Path("./output/win_rates") if not output_dir.exists(): output_dir.mkdir(parents=True) # load default criteria. You can also define your own criteria as a list Criteria objects criteria = pairwise_scores_criteria()

In [ ]:

# run pairwise comparisons for each question set and each pair of [base, other].
all_results = []
for question_set in question_sets:
    for other in others:
        rich_print(f"Processing {base} vs {other} for question set: {question_set}")
        result = get_pairwise_scores(
            llm_client=llm_client,
            llm_config=llm_config,
            base_name=base,
            other_name=other,
            base_answers=pd.read_json(
                f"{input_dir}/{base}/{question_set}_answers.json"
            ),
            other_answers=pd.read_json(
                f"{input_dir}/{other}/{question_set}_answers.json"
            ),
            criteria=criteria,
            trials=trials,
            include_score_id_in_prompt=True,
            question_id_key="question_id",
            question_text_key="question",  # Column name in the answer files
        )
        result["question_set"] = question_set
        all_results.append(result)

        # save pairwise results for each question set and pair of [base, other]
        result.to_csv(
            output_dir / f"{question_set}_{base}--{other}.csv",
            index=False,
        )

# save all pairwise results in a single file
all_results_df = pd.concat(all_results, ignore_index=True)
all_results_df.to_csv(output_dir / "win_rates.csv", index=False)

# perform significance testing on the results
significance_test_results = analyze_criteria(
    all_results_df,
    alpha=alpha,
)
significance_test_results.to_csv(output_dir / "winrates_sig_tests.csv", index=False)

print_df(
    cast(
        pd.DataFrame,
        significance_test_results[
            [
                "question_set",
                "criteria",
                "base_name",
                "other_name",
                "base_mean",
                "other_mean",
                "formatted_corrected_p_value",
            ]
        ],
    ),
    "Win Rates Summary",
)

# run pairwise comparisons for each question set and each pair of [base, other]. all_results = [] for question_set in question_sets: for other in others: rich_print(f"Processing {base} vs {other} for question set: {question_set}") result = get_pairwise_scores( llm_client=llm_client, llm_config=llm_config, base_name=base, other_name=other, base_answers=pd.read_json( f"{input_dir}/{base}/{question_set}_answers.json" ), other_answers=pd.read_json( f"{input_dir}/{other}/{question_set}_answers.json" ), criteria=criteria, trials=trials, include_score_id_in_prompt=True, question_id_key="question_id", question_text_key="question", # Column name in the answer files ) result["question_set"] = question_set all_results.append(result) # save pairwise results for each question set and pair of [base, other] result.to_csv( output_dir / f"{question_set}_{base}--{other}.csv", index=False, ) # save all pairwise results in a single file all_results_df = pd.concat(all_results, ignore_index=True) all_results_df.to_csv(output_dir / "win_rates.csv", index=False) # perform significance testing on the results significance_test_results = analyze_criteria( all_results_df, alpha=alpha, ) significance_test_results.to_csv(output_dir / "winrates_sig_tests.csv", index=False) print_df( cast( pd.DataFrame, significance_test_results[ [ "question_set", "criteria", "base_name", "other_name", "base_mean", "other_mean", "formatted_corrected_p_value", ] ], ), "Win Rates Summary", )

In [ ]:

rich_print("Model usage statistics:")
rich_print(llm_client.metrics_store.get_metrics())

rich_print("Model usage statistics:") rich_print(llm_client.metrics_store.get_metrics())

Reference-based Scoring¶

When reference answers (such as ground truth or "gold standard" responses) are available, AutoE can evaluate RAG-generated answers against these references using metrics like correctness, completeness, or other user-defined criteria on a customizable scoring scale.

In the example below, we use the long context version as the reference method. Note that this is not ground truth—we use it here purely to demonstrate the reference-based scoring workflow. While more context can sometimes lead to more complete answers, this relationship is not universal: it depends on the model, the specific context window, and the nature of the questions. We then score the answers from short context against those from the long context version using the default metrics (correctness and completeness) on a scale from 1 to 10.

In [ ]:

# Config LLM model to be used as judge
llm_config = LLMConfig(
    model="gpt-5.2",
    api_key=SecretStr(os.environ["OPENAI_API_KEY"]),
    llm_provider=LLMProvider.OpenAIChat,
    concurrent_requests=32,
    call_args={"temperature": 0.0, "seed": 42},
)
llm_client = ModelFactory.create_chat_model(llm_config)

# Config LLM model to be used as judge llm_config = LLMConfig( model="gpt-5.2", api_key=SecretStr(os.environ["OPENAI_API_KEY"]), llm_provider=LLMProvider.OpenAIChat, concurrent_requests=32, call_args={"temperature": 0.0, "seed": 42}, ) llm_client = ModelFactory.create_chat_model(llm_config)

In [ ]:

# Config conditions for comparison
reference = (
    "vector_rag_long_context"  # long context as reference (more complete answers)
)
generated_rags = ["vector_rag_short_context"]  # short context to evaluate
question_sets = ["data_global", "data_local", "data_linked"]
trials = 4  # number of trials must be an even number to support counterbalancing

input_dir = "./example_answers"
output_dir = Path("./output/reference_scores")
if not output_dir.exists():
    output_dir.mkdir(parents=True)

# load default criteria (correctness and completeness). You can also define your own criteria as a list Criteria objects
criteria = reference_scores_criteria()

# Config conditions for comparison reference = ( "vector_rag_long_context" # long context as reference (more complete answers) ) generated_rags = ["vector_rag_short_context"] # short context to evaluate question_sets = ["data_global", "data_local", "data_linked"] trials = 4 # number of trials must be an even number to support counterbalancing input_dir = "./example_answers" output_dir = Path("./output/reference_scores") if not output_dir.exists(): output_dir.mkdir(parents=True) # load default criteria (correctness and completeness). You can also define your own criteria as a list Criteria objects criteria = reference_scores_criteria()

In [ ]:

# run comparisons for each question set and each pair of [generated, reference].
all_results = []
all_summaries = []
question_set = ""
for question_set in question_sets:
    for generated in generated_rags:
        rich_print(
            f"Comparing {generated} vs. {reference} for question set: {question_set}"
        )
        result = get_reference_scores(
            llm_client=llm_client,
            llm_config=llm_config,
            reference_answers=pd.read_json(
                f"{input_dir}/{reference}/{question_set}_answers.json"
            ),
            generated_answers=pd.read_json(
                f"{input_dir}/{generated}/{question_set}_answers.json"
            ),
            criteria=criteria,
            trials=trials,
            score_min=1,
            score_max=10,
            include_score_id_in_prompt=True,
            question_id_key="question_id",
            question_text_key="question",  # Column name in the answer files
        )

        all_results.append(result)
        result.to_csv(
            output_dir / f"{question_set}_{reference}--{generated}.csv",
            index=False,
        )
        summary_df = summarize_reference_scores(result)
        summary_df["question_set"] = question_set
        summary_df["reference"] = reference
        summary_df["generated"] = generated
        all_summaries.append(summary_df)

# save all results into a single file
all_results_df = pd.concat(all_results, ignore_index=True)
all_results_df.to_csv(output_dir / "reference_scores.csv", index=False)

all_summary_df = pd.concat(all_summaries, ignore_index=True)
print_df(
    cast(
        pd.DataFrame,
        all_summary_df[
            ["question_set", "criteria", "reference", "generated", "mean", "std"]
        ].reset_index(drop=True),
    ),
    "Reference Scores Summary",
)
all_summary_df.to_csv(output_dir / "reference_scores_summary.csv", index=False)

# run comparisons for each question set and each pair of [generated, reference]. all_results = [] all_summaries = [] question_set = "" for question_set in question_sets: for generated in generated_rags: rich_print( f"Comparing {generated} vs. {reference} for question set: {question_set}" ) result = get_reference_scores( llm_client=llm_client, llm_config=llm_config, reference_answers=pd.read_json( f"{input_dir}/{reference}/{question_set}_answers.json" ), generated_answers=pd.read_json( f"{input_dir}/{generated}/{question_set}_answers.json" ), criteria=criteria, trials=trials, score_min=1, score_max=10, include_score_id_in_prompt=True, question_id_key="question_id", question_text_key="question", # Column name in the answer files ) all_results.append(result) result.to_csv( output_dir / f"{question_set}_{reference}--{generated}.csv", index=False, ) summary_df = summarize_reference_scores(result) summary_df["question_set"] = question_set summary_df["reference"] = reference summary_df["generated"] = generated all_summaries.append(summary_df) # save all results into a single file all_results_df = pd.concat(all_results, ignore_index=True) all_results_df.to_csv(output_dir / "reference_scores.csv", index=False) all_summary_df = pd.concat(all_summaries, ignore_index=True) print_df( cast( pd.DataFrame, all_summary_df[ ["question_set", "criteria", "reference", "generated", "mean", "std"] ].reset_index(drop=True), ), "Reference Scores Summary", ) all_summary_df.to_csv(output_dir / "reference_scores_summary.csv", index=False)

Assertion-based Scoring¶

Assertion-based scoring evaluates RAG-generated answers by checking whether they contain specific factual assertions or claims that should be present according to a reference or gold standard. This approach is especially useful for tasks where the presence or absence of key facts is more important than holistic correctness or completeness.

Standard Assertions (for data-local and data-linked questions)¶

For data-local and data-linked questions, we use standard (flat) assertions. These are straightforward factual claims that can be independently verified against the generated answer. The LLM judge checks whether each assertion is supported by the answer.

In [ ]:

# Config LLM model to be used as judge
llm_config = LLMConfig(
    model="gpt-5.2",
    api_key=SecretStr(os.environ["OPENAI_API_KEY"]),
    llm_provider=LLMProvider.OpenAIChat,
    concurrent_requests=100,
    call_args={"temperature": 0.0, "seed": 42},
)
llm_client = ModelFactory.create_chat_model(llm_config)

# Config LLM model to be used as judge llm_config = LLMConfig( model="gpt-5.2", api_key=SecretStr(os.environ["OPENAI_API_KEY"]), llm_provider=LLMProvider.OpenAIChat, concurrent_requests=100, call_args={"temperature": 0.0, "seed": 42}, ) llm_client = ModelFactory.create_chat_model(llm_config)

In [ ]:

# Config for standard assertion scoring (data-local and data-linked)
generated_rags = ["vector_rag_short_context", "vector_rag_long_context"]
question_sets = [
    "data_local",
    "data_linked",
]  # Standard assertions for these question types
pass_threshold = 0.5
trials = 2

input_dir = Path("./example_answers")
output_dir = Path("./output/assertion_scores")
if not output_dir.exists():
    output_dir.mkdir(parents=True)

# Config for standard assertion scoring (data-local and data-linked) generated_rags = ["vector_rag_short_context", "vector_rag_long_context"] question_sets = [ "data_local", "data_linked", ] # Standard assertions for these question types pass_threshold = 0.5 trials = 2 input_dir = Path("./example_answers") output_dir = Path("./output/assertion_scores") if not output_dir.exists(): output_dir.mkdir(parents=True)

In [ ]:

# Run standard assertion scoring for multiple RAG methods
# This uses run_assertion_evaluation which handles multiple RAGs and runs significance tests

results_df = run_assertion_evaluation(
    llm_client=llm_client,
    llm_config=llm_config,
    question_sets=question_sets,
    generated_rags=generated_rags,
    input_dir=str(input_dir),
    output_dir=output_dir,
    trials=trials,
    top_k_assertions=None,  # Use all assertions
    pass_threshold=pass_threshold,
    # Assertions are in input_dir (not in RAG subdirs)
    assertions_filename_template=f"{question_set}_assertions.json",
    # Answers are in RAG subdirs with question_set name
    answers_path_template="{input_dir}/{generated_rag}/{question_set}_answers.json",
    run_significance_test=True,  # Run Friedman/Wilcoxon tests
    significance_alpha=0.05,
    significance_correction="holm",
    question_text_key="question",  # Column name in the answer files
    answer_text_key="answer",
)

print_df(results_df, "Assertion Scoring Results Summary")

# Run standard assertion scoring for multiple RAG methods # This uses run_assertion_evaluation which handles multiple RAGs and runs significance tests results_df = run_assertion_evaluation( llm_client=llm_client, llm_config=llm_config, question_sets=question_sets, generated_rags=generated_rags, input_dir=str(input_dir), output_dir=output_dir, trials=trials, top_k_assertions=None, # Use all assertions pass_threshold=pass_threshold, # Assertions are in input_dir (not in RAG subdirs) assertions_filename_template=f"{question_set}_assertions.json", # Answers are in RAG subdirs with question_set name answers_path_template="{input_dir}/{generated_rag}/{question_set}_answers.json", run_significance_test=True, # Run Friedman/Wilcoxon tests significance_alpha=0.05, significance_correction="holm", question_text_key="question", # Column name in the answer files answer_text_key="answer", ) print_df(results_df, "Assertion Scoring Results Summary")

In [ ]:

rich_print("\nModel usage statistics:")
rich_print(llm_client.metrics_store.get_metrics())

rich_print("\nModel usage statistics:") rich_print(llm_client.metrics_store.get_metrics())

Hierarchical Assertions (for data-global questions)¶

For data-global questions, we use hierarchical assertions. These have a global assertion with supporting (local) assertions, providing deeper insight into answer quality:

1. Global assertion pass/fail: Whether the main assertion is satisfied
2. Support coverage: What fraction of supporting assertions are satisfied
3. Discovery detection: Whether the answer contains relevant information beyond what's covered by the supporting assertions

This is particularly useful for global questions where answers may partially satisfy complex requirements.

In [ ]:

# Config LLM model to be used as judge
llm_config = LLMConfig(
    model="gpt-5.2",
    api_key=SecretStr(os.environ["OPENAI_API_KEY"]),
    llm_provider=LLMProvider.OpenAIChat,
    concurrent_requests=100,
    call_args={"temperature": 0.0, "seed": 42},
)
llm_client = ModelFactory.create_chat_model(llm_config)

# Config LLM model to be used as judge llm_config = LLMConfig( model="gpt-5.2", api_key=SecretStr(os.environ["OPENAI_API_KEY"]), llm_provider=LLMProvider.OpenAIChat, concurrent_requests=100, call_args={"temperature": 0.0, "seed": 42}, ) llm_client = ModelFactory.create_chat_model(llm_config)

In [ ]:

# Config for hierarchical assertion scoring (data-global only)
hierarchical_assertions_file = (
    "data_global_assertions.json"  # Contains assertions with supporting_assertions
)
generated_rags = ["vector_rag_short_context", "vector_rag_long_context"]
pass_threshold = 0.5
trials = 2  # number of trials for each assertion

# Evaluation mode: JOINT or STAGED
# - JOINT: Evaluate global and supporting assertions together in one LLM call. Cheaper but may be less accurate than the STAGED mode.
# - STAGED: Evaluate global assertions first, then supporting assertions only for passed globals.
#           Ensures global pass rate matches standard scoring.
hierarchical_mode = HierarchicalMode.STAGED

input_dir = Path("./example_answers")
output_dir = Path("./output/hierarchical_assertion_scores")
if not output_dir.exists():
    output_dir.mkdir(parents=True)

# Config for hierarchical assertion scoring (data-global only) hierarchical_assertions_file = ( "data_global_assertions.json" # Contains assertions with supporting_assertions ) generated_rags = ["vector_rag_short_context", "vector_rag_long_context"] pass_threshold = 0.5 trials = 2 # number of trials for each assertion # Evaluation mode: JOINT or STAGED # - JOINT: Evaluate global and supporting assertions together in one LLM call. Cheaper but may be less accurate than the STAGED mode. # - STAGED: Evaluate global assertions first, then supporting assertions only for passed globals. # Ensures global pass rate matches standard scoring. hierarchical_mode = HierarchicalMode.STAGED input_dir = Path("./example_answers") output_dir = Path("./output/hierarchical_assertion_scores") if not output_dir.exists(): output_dir.mkdir(parents=True)

In [ ]:

# Load hierarchical assertions for data-global questions
# The file structure has assertions with a "supporting_assertions" field
assertions = load_and_normalize_hierarchical_assertions(
    input_dir / hierarchical_assertions_file,
)
rich_print(
    f"Loaded {len(assertions)} hierarchical assertions for data-global questions"
)

# Load hierarchical assertions for data-global questions # The file structure has assertions with a "supporting_assertions" field assertions = load_and_normalize_hierarchical_assertions( input_dir / hierarchical_assertions_file, ) rich_print( f"Loaded {len(assertions)} hierarchical assertions for data-global questions" )

In [ ]:

# Run hierarchical assertion evaluation for all RAG methods
# This handles scoring, aggregation, comparison, and significance tests

comparison_df = run_hierarchical_assertion_evaluation(
    llm_client=llm_client,
    llm_config=llm_config,
    generated_rags=generated_rags,
    assertions=assertions,
    input_dir=str(input_dir),
    output_dir=output_dir,
    trials=trials,
    pass_threshold=pass_threshold,
    mode=hierarchical_mode,
    answers_path_template="{input_dir}/{generated_rag}/data_global_answers.json",
    run_significance_test=True,
    significance_alpha=0.05,
    significance_correction="holm",
    # Optional: run assertion-level clustered permutation tests
    # as a secondary analysis that accounts for within-question correlation
    run_clustered_permutation=True,
    n_permutations=10_000,
    permutation_seed=42,
    question_id_key="question_id",
    question_text_key="question",  # Column name in the answer files
    answer_text_key="answer",
    supporting_assertions_key="supporting_assertions",
)

# Run hierarchical assertion evaluation for all RAG methods # This handles scoring, aggregation, comparison, and significance tests comparison_df = run_hierarchical_assertion_evaluation( llm_client=llm_client, llm_config=llm_config, generated_rags=generated_rags, assertions=assertions, input_dir=str(input_dir), output_dir=output_dir, trials=trials, pass_threshold=pass_threshold, mode=hierarchical_mode, answers_path_template="{input_dir}/{generated_rag}/data_global_answers.json", run_significance_test=True, significance_alpha=0.05, significance_correction="holm", # Optional: run assertion-level clustered permutation tests # as a secondary analysis that accounts for within-question correlation run_clustered_permutation=True, n_permutations=10_000, permutation_seed=42, question_id_key="question_id", question_text_key="question", # Column name in the answer files answer_text_key="answer", supporting_assertions_key="supporting_assertions", )

In [ ]:

# The pipeline automatically prints and saves a significance summary table.
# You can also load it from CSV for further analysis:
sig_summary = pd.read_csv(output_dir / "significance_summary.csv")
print_df(sig_summary, "Significance Test Summary")

# The pipeline automatically prints and saves a significance summary table. # You can also load it from CSV for further analysis: sig_summary = pd.read_csv(output_dir / "significance_summary.csv") print_df(sig_summary, "Significance Test Summary")

In [ ]:

# Alternatively, you can run significance tests separately and build the summary
# yourself using summarize_significance_results(). This is useful when you have
# pre-computed aggregated scores and want to re-run tests with different parameters.
# See the compare_hierarchical_assertion_scores_significance and
# summarize_significance_results functions in benchmark_qed.autoe.assertion.

# Alternatively, you can run significance tests separately and build the summary # yourself using summarize_significance_results(). This is useful when you have # pre-computed aggregated scores and want to re-run tests with different parameters. # See the compare_hierarchical_assertion_scores_significance and # summarize_significance_results functions in benchmark_qed.autoe.assertion.

Chunk-level Assertion Scoring¶

Instead of scoring a synthesized answer, chunk-level assertion scoring evaluates each retrieved chunk directly against the per-question assertions and reports coverage at each k. This isolates retrieval quality from generation quality.

The input is a standard retrieval-results file using the same data_*_retrieval_results.json schema produced for the retrieval metrics, so no conversion is required:

• Each record has a question_id, a text question field, and a context list.
• Each context item has a chunk_id, text, and an optional rank. When rank is present on every item, chunks are evaluated in rank order (top-k semantics); otherwise the chunks are assumed to be already pre-sorted in decreasing order of relevance.

Three metrics are reported at each k:

• Coverage: macro-averaged pass rate (full + partial support)
• Strict Coverage: macro-averaged full-support rate only
• Coverage Strength: mean score across assertions

In [ ]:

import asyncio
import json

from graphrag_storage.file_storage import FileStorage

from benchmark_qed.autoe.chunk_assertion import run_assertion_eval_chunk_mode
from benchmark_qed.autoe.data_model.retrieval_result import (
    load_retrieval_results_from_dicts,
)
from benchmark_qed.autoe.prompts import assertion as chunk_assertion_prompts

# Retrieval results: standard schema {question_id, text, context: [{chunk_id, text, rank?}]}.
# rank is optional; when absent the chunks are assumed to be pre-sorted by relevance.
generated_rag = "vector_rag_short_context"
retrieval_path = input_dir / generated_rag / "data_global_retrieval_results.json"
retrieval_records = json.loads(retrieval_path.read_text(encoding="utf-8"))
eval_results = load_retrieval_results_from_dicts(
    retrieval_records,
    context_id_key="chunk_id",
    context_text_key="text",
    question_text_key="text",
)

# Assertions: each record carries an "assertions" list of {"statement": ...}.
assertions_records = json.loads(
    (input_dir / "data_global_assertions.json").read_text(encoding="utf-8")
)
question_set = {"assertions": assertions_records}

# Default chunk-assertion prompts shipped with the package.
prompts_dir = Path(chunk_assertion_prompts.__file__).parent
system_prompt = (prompts_dir / "chunk_assertion_system_prompt.txt").read_text(
    encoding="utf-8"
)
user_prompt = (prompts_dir / "chunk_assertion_user_prompt.txt").read_text(
    encoding="utf-8"
)

chunk_output_dir = Path("./output/chunk_assertion_scores")
chunk_output_dir.mkdir(parents=True, exist_ok=True)
chunk_output_storage = FileStorage(str(chunk_output_dir))

summaries = asyncio.run(
    run_assertion_eval_chunk_mode(
        eval_results,
        question_set,
        llm_client=llm_client,
        llm_config=llm_config,
        output_storage=chunk_output_storage,
        pass_threshold=0.5,
        k_list=[5, 10, 20, 50],
        system_prompt=system_prompt,
        user_prompt=user_prompt,
    )
)

import asyncio import json from graphrag_storage.file_storage import FileStorage from benchmark_qed.autoe.chunk_assertion import run_assertion_eval_chunk_mode from benchmark_qed.autoe.data_model.retrieval_result import ( load_retrieval_results_from_dicts, ) from benchmark_qed.autoe.prompts import assertion as chunk_assertion_prompts # Retrieval results: standard schema {question_id, text, context: [{chunk_id, text, rank?}]}. # rank is optional; when absent the chunks are assumed to be pre-sorted by relevance. generated_rag = "vector_rag_short_context" retrieval_path = input_dir / generated_rag / "data_global_retrieval_results.json" retrieval_records = json.loads(retrieval_path.read_text(encoding="utf-8")) eval_results = load_retrieval_results_from_dicts( retrieval_records, context_id_key="chunk_id", context_text_key="text", question_text_key="text", ) # Assertions: each record carries an "assertions" list of {"statement": ...}. assertions_records = json.loads( (input_dir / "data_global_assertions.json").read_text(encoding="utf-8") ) question_set = {"assertions": assertions_records} # Default chunk-assertion prompts shipped with the package. prompts_dir = Path(chunk_assertion_prompts.__file__).parent system_prompt = (prompts_dir / "chunk_assertion_system_prompt.txt").read_text( encoding="utf-8" ) user_prompt = (prompts_dir / "chunk_assertion_user_prompt.txt").read_text( encoding="utf-8" ) chunk_output_dir = Path("./output/chunk_assertion_scores") chunk_output_dir.mkdir(parents=True, exist_ok=True) chunk_output_storage = FileStorage(str(chunk_output_dir)) summaries = asyncio.run( run_assertion_eval_chunk_mode( eval_results, question_set, llm_client=llm_client, llm_config=llm_config, output_storage=chunk_output_storage, pass_threshold=0.5, k_list=[5, 10, 20, 50], system_prompt=system_prompt, user_prompt=user_prompt, ) )

In [ ]:

# Summarize coverage metrics at each k


# Sort by numeric k (labels look like "k5", "k10", ...), keeping the "all" entry last.
def _k_sort_key(item: tuple[str, object]) -> tuple[int, int]:
    label = item[0]
    if label.startswith("k") and label[1:].isdigit():
        return (0, int(label[1:]))
    return (1, 0)


chunk_results_df = pd.DataFrame([
    {
        "k": label,
        "coverage": summary.coverage,
        "strict_coverage": summary.strict_coverage,
        "coverage_strength": summary.mean_score,
        "mean_chunks": summary.mean_retrieved_chunks,
    }
    for label, summary in sorted(summaries.items(), key=_k_sort_key)
])
print_df(chunk_results_df, "Chunk-level Assertion Scoring Results")

# Summarize coverage metrics at each k # Sort by numeric k (labels look like "k5", "k10", ...), keeping the "all" entry last. def _k_sort_key(item: tuple[str, object]) -> tuple[int, int]: label = item[0] if label.startswith("k") and label[1:].isdigit(): return (0, int(label[1:])) return (1, 0) chunk_results_df = pd.DataFrame([ { "k": label, "coverage": summary.coverage, "strict_coverage": summary.strict_coverage, "coverage_strength": summary.mean_score, "mean_chunks": summary.mean_retrieved_chunks, } for label, summary in sorted(summaries.items(), key=_k_sort_key) ]) print_df(chunk_results_df, "Chunk-level Assertion Scoring Results")

Original vs. Differential Pairwise Scoring¶

Standard ("original") pairwise judging compares the full answers, which makes the verdict sensitive to confounds such as answer length and formatting. Differential pairwise judging (the extract-common-and-unique method) reduces that bias in two steps:

1. Extract: for each question, isolate the content that is common to both answers and the content that is unique to each.
2. Judge: score only the unique content, evaluating relevance, diversity, and comprehensiveness together in a single call.

Because the differential judge always scores exactly these criteria, the cells below restrict the original judge to the same criteria and run both methods on the same answer pairs. We then compare their win rates side-by-side to see how much the judging method changes the assessment scores.

In [ ]:

from benchmark_qed.autoe.pairwise import get_differential_pairwise_scores

# Config LLM model to be used as judge
llm_config = LLMConfig(
    model="gpt-5.2",
    api_key=SecretStr(os.environ["OPENAI_API_KEY"]),
    llm_provider=LLMProvider.OpenAIChat,
    concurrent_requests=32,
    call_args={"temperature": 0.0, "seed": 42},
)
llm_client = ModelFactory.create_chat_model(llm_config)

# Compare the original vs. differential judge on the SAME answer pairs.
diff_base = "vector_rag_short_context"
diff_others = ["vector_rag_long_context"]
diff_question_sets = ["data_global", "data_local", "data_linked"]
diff_trials = 4  # must be even to support counterbalancing
alpha = 0.05

diff_input_dir = "./example_answers"
diff_output_dir = Path("./output/original_vs_differential")
diff_output_dir.mkdir(parents=True, exist_ok=True)

# The differential judge now accepts any criteria (defaults or user-defined). For this
# experiment we score these three; the original judge is restricted to the same set so
# the two methods are directly comparable.
DIFFERENTIAL_CRITERIA = ("relevance", "diversity", "comprehensiveness")
comparable_criteria = [
    c for c in pairwise_scores_criteria() if c.name in DIFFERENTIAL_CRITERIA
]

from benchmark_qed.autoe.pairwise import get_differential_pairwise_scores # Config LLM model to be used as judge llm_config = LLMConfig( model="gpt-5.2", api_key=SecretStr(os.environ["OPENAI_API_KEY"]), llm_provider=LLMProvider.OpenAIChat, concurrent_requests=32, call_args={"temperature": 0.0, "seed": 42}, ) llm_client = ModelFactory.create_chat_model(llm_config) # Compare the original vs. differential judge on the SAME answer pairs. diff_base = "vector_rag_short_context" diff_others = ["vector_rag_long_context"] diff_question_sets = ["data_global", "data_local", "data_linked"] diff_trials = 4 # must be even to support counterbalancing alpha = 0.05 diff_input_dir = "./example_answers" diff_output_dir = Path("./output/original_vs_differential") diff_output_dir.mkdir(parents=True, exist_ok=True) # The differential judge now accepts any criteria (defaults or user-defined). For this # experiment we score these three; the original judge is restricted to the same set so # the two methods are directly comparable. DIFFERENTIAL_CRITERIA = ("relevance", "diversity", "comprehensiveness") comparable_criteria = [ c for c in pairwise_scores_criteria() if c.name in DIFFERENTIAL_CRITERIA ]

In [ ]:

# Run BOTH the original and the differential judge on the same answer pairs.
original_results = []
differential_results = []
for question_set in diff_question_sets:
    for other in diff_others:
        base_answers = pd.read_json(
            f"{diff_input_dir}/{diff_base}/{question_set}_answers.json"
        )
        other_answers = pd.read_json(
            f"{diff_input_dir}/{other}/{question_set}_answers.json"
        )

        rich_print(f"[Original] {diff_base} vs {other} — {question_set}")
        original = get_pairwise_scores(
            llm_client=llm_client,
            llm_config=llm_config,
            base_name=diff_base,
            other_name=other,
            base_answers=base_answers,
            other_answers=other_answers,
            criteria=comparable_criteria,
            trials=diff_trials,
            include_score_id_in_prompt=True,
            question_id_key="question_id",
            question_text_key="question",
        )
        original["question_set"] = question_set
        original_results.append(original)

        rich_print(f"[Differential] {diff_base} vs {other} — {question_set}")
        differential = get_differential_pairwise_scores(
            llm_client=llm_client,
            llm_config=llm_config,
            base_name=diff_base,
            other_name=other,
            base_answers=base_answers,
            other_answers=other_answers,
            criteria=comparable_criteria,
            trials=diff_trials,
            include_score_id_in_prompt=True,
            question_id_key="question_id",
            question_text_key="question",
        )
        differential["question_set"] = question_set
        differential_results.append(differential)

original_df = pd.concat(original_results, ignore_index=True)
differential_df = pd.concat(differential_results, ignore_index=True)
original_df.to_csv(diff_output_dir / "original_pairwise_scores.csv", index=False)
differential_df.to_csv(
    diff_output_dir / "differential_pairwise_scores.csv", index=False
)

# Run BOTH the original and the differential judge on the same answer pairs. original_results = [] differential_results = [] for question_set in diff_question_sets: for other in diff_others: base_answers = pd.read_json( f"{diff_input_dir}/{diff_base}/{question_set}_answers.json" ) other_answers = pd.read_json( f"{diff_input_dir}/{other}/{question_set}_answers.json" ) rich_print(f"[Original] {diff_base} vs {other} — {question_set}") original = get_pairwise_scores( llm_client=llm_client, llm_config=llm_config, base_name=diff_base, other_name=other, base_answers=base_answers, other_answers=other_answers, criteria=comparable_criteria, trials=diff_trials, include_score_id_in_prompt=True, question_id_key="question_id", question_text_key="question", ) original["question_set"] = question_set original_results.append(original) rich_print(f"[Differential] {diff_base} vs {other} — {question_set}") differential = get_differential_pairwise_scores( llm_client=llm_client, llm_config=llm_config, base_name=diff_base, other_name=other, base_answers=base_answers, other_answers=other_answers, criteria=comparable_criteria, trials=diff_trials, include_score_id_in_prompt=True, question_id_key="question_id", question_text_key="question", ) differential["question_set"] = question_set differential_results.append(differential) original_df = pd.concat(original_results, ignore_index=True) differential_df = pd.concat(differential_results, ignore_index=True) original_df.to_csv(diff_output_dir / "original_pairwise_scores.csv", index=False) differential_df.to_csv( diff_output_dir / "differential_pairwise_scores.csv", index=False )

In [ ]:

# Compute win rates for each method and compare the assessment scores side-by-side.
# analyze_criteria returns the "other" method win rate (other_mean) per
# question_set & criterion, along with Holm-corrected significance.
original_summary = analyze_criteria(original_df, alpha=alpha)
differential_summary = analyze_criteria(differential_df, alpha=alpha)

merge_keys = ["question_set", "criteria", "base_name", "other_name"]
comparison = original_summary.merge(
    differential_summary,
    on=merge_keys,
    suffixes=("_original", "_differential"),
)

# Difference in the "other" method win rate between the two judging methods.
comparison["winrate_delta"] = (
    comparison["other_mean_differential"] - comparison["other_mean_original"]
)
comparison.to_csv(
    diff_output_dir / "original_vs_differential_comparison.csv", index=False
)

# Build an unambiguous display table.
# Win rate always refers to the "other" method (long context) beating the "base"
# method (short context); the base win rate is simply 1 - other. Showing both sides
# for BOTH judges, and naming columns after the actual methods, avoids misreading.
base_label = comparison["base_name"].iloc[0]
other_label = comparison["other_name"].iloc[0]


def _distinguishing_labels(a: str, b: str) -> tuple[str, str]:
    """Strip the shared prefix/suffix so labels keep only the differing part.

    e.g. ("vector_rag_short_context", "vector_rag_long_context") -> ("short", "long").
    Falls back to the full names if they share no distinguishing middle.
    """
    prefix_len = 0
    for ca, cb in zip(a, b, strict=False):
        if ca != cb:
            break
        prefix_len += 1
    suffix_len = 0
    for ca, cb in zip(a[prefix_len:][::-1], b[prefix_len:][::-1], strict=False):
        if ca != cb:
            break
        suffix_len += 1
    short_a = a[prefix_len : len(a) - suffix_len].strip("_")
    short_b = b[prefix_len : len(b) - suffix_len].strip("_")
    return (short_a or a, short_b or b)


base_short, other_short = _distinguishing_labels(base_label, other_label)

display_df = pd.DataFrame({
    "question set": comparison["question_set"],
    "criteria": comparison["criteria"],
    # Original judge: both sides + significance
    f"{base_short} WR (orig)": 1 - comparison["other_mean_original"],
    f"{other_short} WR (orig)": comparison["other_mean_original"],
    "p (orig)": comparison["formatted_corrected_p_value_original"],
    # Differential judge: both sides + significance
    f"{base_short} WR (diff)": 1 - comparison["other_mean_differential"],
    f"{other_short} WR (diff)": comparison["other_mean_differential"],
    "p (diff)": comparison["formatted_corrected_p_value_differential"],
    # How much the "other" method win rate shifted after bias correction
    f"{other_short} WR Δ (diff-orig)": comparison["winrate_delta"],
})

print_df(
    cast(pd.DataFrame, display_df),
    f"Win rate of '{other_label}' vs '{base_label}'  "
    f"(columns show '{other_short}'/'{base_short}'; base WR = 1 - other WR; "
    f"orig=original judge, diff=differential judge)",
)

# Compute win rates for each method and compare the assessment scores side-by-side. # analyze_criteria returns the "other" method win rate (other_mean) per # question_set & criterion, along with Holm-corrected significance. original_summary = analyze_criteria(original_df, alpha=alpha) differential_summary = analyze_criteria(differential_df, alpha=alpha) merge_keys = ["question_set", "criteria", "base_name", "other_name"] comparison = original_summary.merge( differential_summary, on=merge_keys, suffixes=("_original", "_differential"), ) # Difference in the "other" method win rate between the two judging methods. comparison["winrate_delta"] = ( comparison["other_mean_differential"] - comparison["other_mean_original"] ) comparison.to_csv( diff_output_dir / "original_vs_differential_comparison.csv", index=False ) # Build an unambiguous display table. # Win rate always refers to the "other" method (long context) beating the "base" # method (short context); the base win rate is simply 1 - other. Showing both sides # for BOTH judges, and naming columns after the actual methods, avoids misreading. base_label = comparison["base_name"].iloc[0] other_label = comparison["other_name"].iloc[0] def _distinguishing_labels(a: str, b: str) -> tuple[str, str]: """Strip the shared prefix/suffix so labels keep only the differing part. e.g. ("vector_rag_short_context", "vector_rag_long_context") -> ("short", "long"). Falls back to the full names if they share no distinguishing middle. """ prefix_len = 0 for ca, cb in zip(a, b, strict=False): if ca != cb: break prefix_len += 1 suffix_len = 0 for ca, cb in zip(a[prefix_len:][::-1], b[prefix_len:][::-1], strict=False): if ca != cb: break suffix_len += 1 short_a = a[prefix_len : len(a) - suffix_len].strip("_") short_b = b[prefix_len : len(b) - suffix_len].strip("_") return (short_a or a, short_b or b) base_short, other_short = _distinguishing_labels(base_label, other_label) display_df = pd.DataFrame({ "question set": comparison["question_set"], "criteria": comparison["criteria"], # Original judge: both sides + significance f"{base_short} WR (orig)": 1 - comparison["other_mean_original"], f"{other_short} WR (orig)": comparison["other_mean_original"], "p (orig)": comparison["formatted_corrected_p_value_original"], # Differential judge: both sides + significance f"{base_short} WR (diff)": 1 - comparison["other_mean_differential"], f"{other_short} WR (diff)": comparison["other_mean_differential"], "p (diff)": comparison["formatted_corrected_p_value_differential"], # How much the "other" method win rate shifted after bias correction f"{other_short} WR Δ (diff-orig)": comparison["winrate_delta"], }) print_df( cast(pd.DataFrame, display_df), f"Win rate of '{other_label}' vs '{base_label}' " f"(columns show '{other_short}'/'{base_short}'; base WR = 1 - other WR; " f"orig=original judge, diff=differential judge)", )

Experiment: joint vs. separate criterion judging¶

The differential judge scores all criteria (relevance, diversity, comprehensiveness) together in a single LLM call. A natural question is whether judging each criterion in its own call would change the verdicts. Historically (on older models such as GPT-4) scoring metrics separately sometimes gave different results, which is why the standard pairwise judge evaluates one criterion per call.

This experiment tests whether that still matters for the differential judge:

• Joint — the shipped get_differential_pairwise_scores (one judge call scores all criteria). We reuse the differential_df computed above.
• Separate — the same extraction step, but one judge call per criterion (each call sees only that criterion's definition).

Both variants use the identical extraction step and the same answer pairs, so the only difference is joint-vs-separate judging. We then compare (a) the aggregate win rates and (b) the per-(question, criterion, trial) verdict agreement rate between the two variants. If agreement is high and win rates match, joint judging is safe; if not, separate judging is warranted.

Note: this reuses differential_df, llm_client, llm_config, and the diff_* config from the differential section above, so run that section first.

In [ ]:

# Separate-call variant of the differential judge: identical extraction, but one
# judge call per criterion (instead of all criteria in a single call).
import asyncio
import itertools
import uuid
from typing import Any

from graphrag_llm.completion import LLMCompletion

from benchmark_qed.autoe.data_model import ConditionPair, PairwiseExtractionLLMResponse
from benchmark_qed.autoe.data_model.pairwise import CriterionVerdict
from benchmark_qed.autoe.pairwise import SCORE_MAPPING
from benchmark_qed.autoe.pairwise.differential import PAIRWISE_PROMPTS_PATH
from benchmark_qed.config.utils import load_template_file
from benchmark_qed.llm import chat

# Criterion definitions kept in sync with the shipped joint judge system prompt.
JUDGE_CRITERIA_DESCRIPTIONS = {
    "relevance": (
        "Which unique content is more directly useful for answering the question — "
        "factually grounded, with specific data points, numbers, or verifiable claims "
        "that address the question's core intent?"
    ),
    "diversity": (
        "Which unique content covers more distinct perspectives, aspects, or dimensions "
        "of the question, offering a wider range of analytical angles to more "
        "comprehensively address the question's core intent?"
    ),
    "comprehensiveness": (
        "Which unique content provides more thorough and complete coverage of the "
        "aspects and details the question calls for, without being redundant or "
        "irrelevant?"
    ),
}


def _single_criterion_judge_system(criterion: str) -> str:
    return f"""---Role---
You are an impartial judge comparing two answers to a question based only on the content that is UNIQUE to each answer.

---Goal---
You are given a question, a summary of the content BOTH answers cover (Common Content), and the UNIQUE content of each answer. Compare the unique content of Answer 1 versus Answer 2 on a SINGLE criterion and declare a tie or pick a winner.

The null assumption is that there is no material difference between the two unique contents. For an answer to win, its unique content must demonstrate clear superiority on this criterion and add meaningful value over the Common Content. Trivial differences result in a tie. If both unique contents are "No unique part" (or trivially empty), declare a tie.

---Criterion---
{criterion.capitalize()}: {JUDGE_CRITERIA_DESCRIPTIONS[criterion]}

---Important Guidelines---
- No position biases: the order of the unique contents should NOT influence your judgment.
- Ignore length: do NOT let the length of the unique content affect your evaluation.
- Ignore formatting style: focus only on substantive content.

---Output Format---
Use these values for the winner: 1 (Answer 1 better), 2 (Answer 2 better), 0 (tie).
Format your response as a JSON object with the structure:
{{
    "reasoning": "A detailed explanation of your assessment.",
    "winner": 1 | 2 | 0
}}
"""


def _single_criterion_judge_user(
    criterion: str,
    score_id_text: str,
    question: str,
    common: str,
    unique1: str,
    unique2: str,
) -> str:
    return f"""{score_id_text}
---Question---
{question}

---Common Content (shared by both answers)---
{common}

---Start of Unique Content of Answer 1---
{unique1}
---End of Unique Content of Answer 1---

---Start of Unique Content of Answer 2---
{unique2}
---End of Unique Content of Answer 2---

Compare the unique content of Answer 1 and Answer 2 on {criterion}, relative to the question. Remember, the null assumption is that there is no material difference.

Format your response as a JSON object with the structure:
{{
    "reasoning": "A detailed explanation of your assessment.",
    "winner": 1 | 2 | 0
}}
"""


async def get_separate_differential_score(
    llm: LLMCompletion,
    *,
    question: str,
    answer_1_name: str,
    answer_1: str,
    answer_2_name: str,
    answer_2: str,
    trial: int = 0,
    include_score_id_in_prompt: bool = True,
    additional_call_args: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    """Extract common/unique once, then judge each criterion in its own call."""
    extract_system = load_template_file(
        PAIRWISE_PROMPTS_PATH / "pairwise_extract_system_prompt.txt"
    )
    extract_user = load_template_file(
        PAIRWISE_PROMPTS_PATH / "pairwise_extract_user_prompt.txt"
    )

    answers_text = {answer_1_name: answer_1, answer_2_name: answer_2}
    # Counterbalance the presentation order across trials (matches the joint judge).
    answers_order = (
        list(answers_text) if trial % 2 == 0 else list(reversed(answers_text))
    )
    score_id = uuid.uuid4().hex
    score_id_text = f"Score ID: {score_id}\n" if include_score_id_in_prompt else ""

    extract_user_text = extract_user.substitute(
        score_id=score_id_text,
        question=question,
        answer1=answers_text[answers_order[0]],
        answer2=answers_text[answers_order[1]],
    ).strip()
    extraction_response = await chat(
        llm,
        messages=[
            {"role": "system", "content": extract_system.template},
            {"role": "user", "content": extract_user_text},
        ],
        response_format=PairwiseExtractionLLMResponse,
        **(additional_call_args or {}),
    )
    extraction = extraction_response.formatted_response
    if extraction is None:
        msg = "LLM did not return a structured PairwiseExtractionLLMResponse."
        raise RuntimeError(msg)

    rows: list[dict[str, Any]] = []
    for criterion in DIFFERENTIAL_CRITERIA:
        judge_user = _single_criterion_judge_user(
            criterion,
            score_id_text,
            question,
            extraction.common,
            extraction.unique_answer_1,
            extraction.unique_answer_2,
        )
        verdict_response = await chat(
            llm,
            messages=[
                {
                    "role": "system",
                    "content": _single_criterion_judge_system(criterion),
                },
                {"role": "user", "content": judge_user},
            ],
            response_format=CriterionVerdict,
            **(additional_call_args or {}),
        )
        verdict = verdict_response.formatted_response
        if verdict is None:
            msg = "LLM did not return a structured CriterionVerdict."
            raise RuntimeError(msg)
        rows.append({
            "score_id": score_id,
            "question": question,
            "criteria": criterion,
            f"{answers_order[0]}_score": SCORE_MAPPING[verdict.winner],
            f"{answers_order[1]}_score": 1 - SCORE_MAPPING[verdict.winner],
            "reasoning": verdict.reasoning,
            "trial": trial,
        })
    return rows


def get_separate_differential_scores(
    *,
    llm_client: LLMCompletion,
    llm_config: LLMConfig,
    base_name: str,
    other_name: str,
    base_answers: pd.DataFrame,
    other_answers: pd.DataFrame,
    trials: int,
    include_score_id_in_prompt: bool = True,
    question_id_key: str = "question_id",
    question_text_key: str = "question_text",
) -> pd.DataFrame:
    """Separate-call counterpart of get_differential_pairwise_scores (same schema)."""
    pairs = (
        base_answers
        .merge(
            other_answers,
            how="inner",
            on=[question_id_key],
            suffixes=("_base", "_other"),
        )
        .drop(columns=[f"{question_text_key}_other"])
        .rename(
            columns={
                question_id_key: "question_id",
                f"{question_text_key}_base": "question_text",
            }
        )
    )
    pairs = pairs[["question_id", "question_text", "answer_base", "answer_other"]]

    tasks = [
        get_separate_differential_score(
            llm_client,
            question=pair.question_text,
            answer_1_name=base_name,
            answer_1=pair.answer_base,
            answer_2_name=other_name,
            answer_2=pair.answer_other,
            trial=n,
            include_score_id_in_prompt=include_score_id_in_prompt,
            additional_call_args=llm_config.call_args,
        )
        for pair in itertools.starmap(ConditionPair, pairs.itertuples(index=False))
        for n in range(trials)
    ]

    async def _run_tasks() -> list[list[dict[str, Any]]]:
        return await asyncio.gather(*tasks)

    nested_results = asyncio.run(_run_tasks())
    flat_rows = [row for group in nested_results for row in group]
    result = pd.DataFrame(flat_rows)
    result["base_name"] = base_name
    result["other_name"] = other_name
    return result

# Separate-call variant of the differential judge: identical extraction, but one # judge call per criterion (instead of all criteria in a single call). import asyncio import itertools import uuid from typing import Any from graphrag_llm.completion import LLMCompletion from benchmark_qed.autoe.data_model import ConditionPair, PairwiseExtractionLLMResponse from benchmark_qed.autoe.data_model.pairwise import CriterionVerdict from benchmark_qed.autoe.pairwise import SCORE_MAPPING from benchmark_qed.autoe.pairwise.differential import PAIRWISE_PROMPTS_PATH from benchmark_qed.config.utils import load_template_file from benchmark_qed.llm import chat # Criterion definitions kept in sync with the shipped joint judge system prompt. JUDGE_CRITERIA_DESCRIPTIONS = { "relevance": ( "Which unique content is more directly useful for answering the question — " "factually grounded, with specific data points, numbers, or verifiable claims " "that address the question's core intent?" ), "diversity": ( "Which unique content covers more distinct perspectives, aspects, or dimensions " "of the question, offering a wider range of analytical angles to more " "comprehensively address the question's core intent?" ), "comprehensiveness": ( "Which unique content provides more thorough and complete coverage of the " "aspects and details the question calls for, without being redundant or " "irrelevant?" ), } def _single_criterion_judge_system(criterion: str) -> str: return f"""---Role--- You are an impartial judge comparing two answers to a question based only on the content that is UNIQUE to each answer. ---Goal--- You are given a question, a summary of the content BOTH answers cover (Common Content), and the UNIQUE content of each answer. Compare the unique content of Answer 1 versus Answer 2 on a SINGLE criterion and declare a tie or pick a winner. The null assumption is that there is no material difference between the two unique contents. For an answer to win, its unique content must demonstrate clear superiority on this criterion and add meaningful value over the Common Content. Trivial differences result in a tie. If both unique contents are "No unique part" (or trivially empty), declare a tie. ---Criterion--- {criterion.capitalize()}: {JUDGE_CRITERIA_DESCRIPTIONS[criterion]} ---Important Guidelines--- - No position biases: the order of the unique contents should NOT influence your judgment. - Ignore length: do NOT let the length of the unique content affect your evaluation. - Ignore formatting style: focus only on substantive content. ---Output Format--- Use these values for the winner: 1 (Answer 1 better), 2 (Answer 2 better), 0 (tie). Format your response as a JSON object with the structure: {{ "reasoning": "A detailed explanation of your assessment.", "winner": 1 | 2 | 0 }} """ def _single_criterion_judge_user( criterion: str, score_id_text: str, question: str, common: str, unique1: str, unique2: str, ) -> str: return f"""{score_id_text} ---Question--- {question} ---Common Content (shared by both answers)--- {common} ---Start of Unique Content of Answer 1--- {unique1} ---End of Unique Content of Answer 1--- ---Start of Unique Content of Answer 2--- {unique2} ---End of Unique Content of Answer 2--- Compare the unique content of Answer 1 and Answer 2 on {criterion}, relative to the question. Remember, the null assumption is that there is no material difference. Format your response as a JSON object with the structure: {{ "reasoning": "A detailed explanation of your assessment.", "winner": 1 | 2 | 0 }} """ async def get_separate_differential_score( llm: LLMCompletion, *, question: str, answer_1_name: str, answer_1: str, answer_2_name: str, answer_2: str, trial: int = 0, include_score_id_in_prompt: bool = True, additional_call_args: dict[str, Any] | None = None, ) -> list[dict[str, Any]]: """Extract common/unique once, then judge each criterion in its own call.""" extract_system = load_template_file( PAIRWISE_PROMPTS_PATH / "pairwise_extract_system_prompt.txt" ) extract_user = load_template_file( PAIRWISE_PROMPTS_PATH / "pairwise_extract_user_prompt.txt" ) answers_text = {answer_1_name: answer_1, answer_2_name: answer_2} # Counterbalance the presentation order across trials (matches the joint judge). answers_order = ( list(answers_text) if trial % 2 == 0 else list(reversed(answers_text)) ) score_id = uuid.uuid4().hex score_id_text = f"Score ID: {score_id}\n" if include_score_id_in_prompt else "" extract_user_text = extract_user.substitute( score_id=score_id_text, question=question, answer1=answers_text[answers_order[0]], answer2=answers_text[answers_order[1]], ).strip() extraction_response = await chat( llm, messages=[ {"role": "system", "content": extract_system.template}, {"role": "user", "content": extract_user_text}, ], response_format=PairwiseExtractionLLMResponse, **(additional_call_args or {}), ) extraction = extraction_response.formatted_response if extraction is None: msg = "LLM did not return a structured PairwiseExtractionLLMResponse." raise RuntimeError(msg) rows: list[dict[str, Any]] = [] for criterion in DIFFERENTIAL_CRITERIA: judge_user = _single_criterion_judge_user( criterion, score_id_text, question, extraction.common, extraction.unique_answer_1, extraction.unique_answer_2, ) verdict_response = await chat( llm, messages=[ { "role": "system", "content": _single_criterion_judge_system(criterion), }, {"role": "user", "content": judge_user}, ], response_format=CriterionVerdict, **(additional_call_args or {}), ) verdict = verdict_response.formatted_response if verdict is None: msg = "LLM did not return a structured CriterionVerdict." raise RuntimeError(msg) rows.append({ "score_id": score_id, "question": question, "criteria": criterion, f"{answers_order[0]}_score": SCORE_MAPPING[verdict.winner], f"{answers_order[1]}_score": 1 - SCORE_MAPPING[verdict.winner], "reasoning": verdict.reasoning, "trial": trial, }) return rows def get_separate_differential_scores( *, llm_client: LLMCompletion, llm_config: LLMConfig, base_name: str, other_name: str, base_answers: pd.DataFrame, other_answers: pd.DataFrame, trials: int, include_score_id_in_prompt: bool = True, question_id_key: str = "question_id", question_text_key: str = "question_text", ) -> pd.DataFrame: """Separate-call counterpart of get_differential_pairwise_scores (same schema).""" pairs = ( base_answers .merge( other_answers, how="inner", on=[question_id_key], suffixes=("_base", "_other"), ) .drop(columns=[f"{question_text_key}_other"]) .rename( columns={ question_id_key: "question_id", f"{question_text_key}_base": "question_text", } ) ) pairs = pairs[["question_id", "question_text", "answer_base", "answer_other"]] tasks = [ get_separate_differential_score( llm_client, question=pair.question_text, answer_1_name=base_name, answer_1=pair.answer_base, answer_2_name=other_name, answer_2=pair.answer_other, trial=n, include_score_id_in_prompt=include_score_id_in_prompt, additional_call_args=llm_config.call_args, ) for pair in itertools.starmap(ConditionPair, pairs.itertuples(index=False)) for n in range(trials) ] async def _run_tasks() -> list[list[dict[str, Any]]]: return await asyncio.gather(*tasks) nested_results = asyncio.run(_run_tasks()) flat_rows = [row for group in nested_results for row in group] result = pd.DataFrame(flat_rows) result["base_name"] = base_name result["other_name"] = other_name return result

In [ ]:

# Run the SEPARATE-call variant on the same answer pairs as the joint differential run.
separate_results = []
for question_set in diff_question_sets:
    for other in diff_others:
        rich_print(f"[Separate] {diff_base} vs {other} — {question_set}")
        separate = get_separate_differential_scores(
            llm_client=llm_client,
            llm_config=llm_config,
            base_name=diff_base,
            other_name=other,
            base_answers=pd.read_json(
                f"{diff_input_dir}/{diff_base}/{question_set}_answers.json"
            ),
            other_answers=pd.read_json(
                f"{diff_input_dir}/{other}/{question_set}_answers.json"
            ),
            trials=diff_trials,
            include_score_id_in_prompt=True,
            question_id_key="question_id",
            question_text_key="question",
        )
        separate["question_set"] = question_set
        separate_results.append(separate)

separate_df = pd.concat(separate_results, ignore_index=True)
separate_df.to_csv(diff_output_dir / "separate_differential_scores.csv", index=False)

# --- (a) Aggregate win-rate comparison: joint vs. separate ---
joint_summary = analyze_criteria(differential_df, alpha=alpha)
separate_summary = analyze_criteria(separate_df, alpha=alpha)

js_keys = ["question_set", "criteria", "base_name", "other_name"]
js = joint_summary.merge(separate_summary, on=js_keys, suffixes=("_joint", "_separate"))
js["winrate_delta"] = js["other_mean_separate"] - js["other_mean_joint"]

winrate_view = pd.DataFrame({
    "question set": js["question_set"].str.replace("data_", "", regex=False),
    "criteria": js["criteria"],
    f"{other_short} WR (joint)": js["other_mean_joint"],
    f"{other_short} WR (separate)": js["other_mean_separate"],
    "WR Δ (sep-joint)": js["winrate_delta"],
    "p (joint)": js["formatted_corrected_p_value_joint"],
    "p (separate)": js["formatted_corrected_p_value_separate"],
})
print_df(cast(pd.DataFrame, winrate_view), "Joint vs. Separate — Win Rates")

# --- (b) Per-(question, criterion, trial) verdict agreement ---
# The base-condition score encodes the verdict (1=base win, 0=base loss, 0.5=tie),
# so comparing it row-by-row tells us whether the two variants agreed.
base_score_col = f"{diff_base}_score"
pair_keys = ["question_set", "question", "criteria", "trial"]
agree = differential_df[[*pair_keys, base_score_col]].merge(
    separate_df[[*pair_keys, base_score_col]],
    on=pair_keys,
    suffixes=("_joint", "_separate"),
)
agree["agree"] = agree[f"{base_score_col}_joint"] == agree[f"{base_score_col}_separate"]

agreement_view = (
    agree
    .groupby("criteria")["agree"]
    .agg(agreement_rate="mean", n="count")
    .reset_index()
)
agreement_view["question set"] = "all"
overall = pd.DataFrame([
    {
        "criteria": "ALL",
        "agreement_rate": agree["agree"].mean(),
        "n": len(agree),
        "question set": "all",
    }
])
agreement_view = pd.concat([agreement_view, overall], ignore_index=True)
print_df(
    cast(pd.DataFrame, agreement_view[["criteria", "agreement_rate", "n"]]),
    "Joint vs. Separate — Verdict Agreement (1.0 = identical verdicts)",
)

# Run the SEPARATE-call variant on the same answer pairs as the joint differential run. separate_results = [] for question_set in diff_question_sets: for other in diff_others: rich_print(f"[Separate] {diff_base} vs {other} — {question_set}") separate = get_separate_differential_scores( llm_client=llm_client, llm_config=llm_config, base_name=diff_base, other_name=other, base_answers=pd.read_json( f"{diff_input_dir}/{diff_base}/{question_set}_answers.json" ), other_answers=pd.read_json( f"{diff_input_dir}/{other}/{question_set}_answers.json" ), trials=diff_trials, include_score_id_in_prompt=True, question_id_key="question_id", question_text_key="question", ) separate["question_set"] = question_set separate_results.append(separate) separate_df = pd.concat(separate_results, ignore_index=True) separate_df.to_csv(diff_output_dir / "separate_differential_scores.csv", index=False) # --- (a) Aggregate win-rate comparison: joint vs. separate --- joint_summary = analyze_criteria(differential_df, alpha=alpha) separate_summary = analyze_criteria(separate_df, alpha=alpha) js_keys = ["question_set", "criteria", "base_name", "other_name"] js = joint_summary.merge(separate_summary, on=js_keys, suffixes=("_joint", "_separate")) js["winrate_delta"] = js["other_mean_separate"] - js["other_mean_joint"] winrate_view = pd.DataFrame({ "question set": js["question_set"].str.replace("data_", "", regex=False), "criteria": js["criteria"], f"{other_short} WR (joint)": js["other_mean_joint"], f"{other_short} WR (separate)": js["other_mean_separate"], "WR Δ (sep-joint)": js["winrate_delta"], "p (joint)": js["formatted_corrected_p_value_joint"], "p (separate)": js["formatted_corrected_p_value_separate"], }) print_df(cast(pd.DataFrame, winrate_view), "Joint vs. Separate — Win Rates") # --- (b) Per-(question, criterion, trial) verdict agreement --- # The base-condition score encodes the verdict (1=base win, 0=base loss, 0.5=tie), # so comparing it row-by-row tells us whether the two variants agreed. base_score_col = f"{diff_base}_score" pair_keys = ["question_set", "question", "criteria", "trial"] agree = differential_df[[*pair_keys, base_score_col]].merge( separate_df[[*pair_keys, base_score_col]], on=pair_keys, suffixes=("_joint", "_separate"), ) agree["agree"] = agree[f"{base_score_col}_joint"] == agree[f"{base_score_col}_separate"] agreement_view = ( agree .groupby("criteria")["agree"] .agg(agreement_rate="mean", n="count") .reset_index() ) agreement_view["question set"] = "all" overall = pd.DataFrame([ { "criteria": "ALL", "agreement_rate": agree["agree"].mean(), "n": len(agree), "question set": "all", } ]) agreement_view = pd.concat([agreement_view, overall], ignore_index=True) print_df( cast(pd.DataFrame, agreement_view[["criteria", "agreement_rate", "n"]]), "Joint vs. Separate — Verdict Agreement (1.0 = identical verdicts)", )

Is the joint-vs-separate difference statistically significant?¶

The agreement rate above tells us how often verdicts differ, but not whether the two approaches differ systematically (one consistently favouring a method). To test that, we treat each (question_set, question, criterion) as a paired observation: the mean base win rate under the joint judge vs. under the separate judge, and run a paired Wilcoxon signed-rank test per criterion and overall. A large p-value (> alpha) means there is no statistically significant systematic difference between the two approaches.

We then:

1. Inspect the reasoning on the cases where the two approaches disagree, to qualitatively judge whether one is more sound than the other (there is no ground truth here, so this is a manual sanity check).
2. Repeat the whole comparison across several judge models, since the joint-vs-separate gap may depend on model capability (the historical concern was specific to GPT-4-era models).

In [ ]:

# Paired significance test: joint vs. separate judging.
# Unit of analysis = per (question_set, question, criterion) mean base win rate
# (averaged over trials, so within-question trial correlation is not double-counted).
import numpy as np
from scipy import stats


def paired_judging_significance(
    joint_df: pd.DataFrame,
    separate_df: pd.DataFrame,
    *,
    base_col: str,
    alpha: float = 0.05,
) -> pd.DataFrame:
    """Wilcoxon signed-rank test on paired joint vs. separate win rates."""
    keys = ["question_set", "question", "criteria"]
    joint_mean = joint_df.groupby(keys)[base_col].mean().to_frame("joint")
    separate_mean = separate_df.groupby(keys)[base_col].mean().to_frame("separate")
    merged = joint_mean.join(separate_mean, how="inner").dropna().reset_index()

    def _test(sub: pd.DataFrame) -> pd.Series:
        a = sub["joint"].to_numpy()
        b = sub["separate"].to_numpy()
        diff = b - a
        if np.allclose(diff, 0.0):
            stat, p_value = 0.0, 1.0
        else:
            try:
                result = stats.wilcoxon(a, b)
                stat = float(result.statistic)  # type: ignore[attr-defined]
                p_value = float(result.pvalue)  # type: ignore[attr-defined]
            except ValueError:
                stat, p_value = 0.0, 1.0
        return pd.Series({
            "n": len(sub),
            "joint_mean_wr": float(a.mean()),
            "separate_mean_wr": float(b.mean()),
            "mean_diff": float(diff.mean()),
            "wilcoxon_stat": stat,
            "p_value": p_value,
            "significant": p_value < alpha,
        })

    per_criterion = (
        merged.groupby("criteria").apply(_test, include_groups=False).reset_index()
    )
    overall = _test(merged)
    overall["criteria"] = "ALL"
    return pd.concat([per_criterion, pd.DataFrame([overall])], ignore_index=True)


base_score_col = f"{diff_base}_score"
significance_df = paired_judging_significance(
    differential_df, separate_df, base_col=base_score_col, alpha=alpha
)
significance_df["p_value"] = significance_df["p_value"].round(3)
significance_df["mean_diff"] = significance_df["mean_diff"].round(4)
print_df(
    cast(pd.DataFrame, significance_df),
    "Joint vs. Separate — Paired Significance (Wilcoxon; p > alpha ⇒ no systematic difference)",
)

overall_p = float(
    significance_df.loc[significance_df["criteria"] == "ALL", "p_value"].iloc[0]
)
if overall_p > alpha:
    rich_print(
        f"[green]Overall p = {overall_p:.3f} > alpha = {alpha}: no statistically "
        f"significant systematic difference between joint and separate judging.[/green]"
    )
else:
    rich_print(
        f"[yellow]Overall p = {overall_p:.3f} <= alpha = {alpha}: the two approaches "
        f"differ systematically — prefer separate judging.[/yellow]"
    )

# Paired significance test: joint vs. separate judging. # Unit of analysis = per (question_set, question, criterion) mean base win rate # (averaged over trials, so within-question trial correlation is not double-counted). import numpy as np from scipy import stats def paired_judging_significance( joint_df: pd.DataFrame, separate_df: pd.DataFrame, *, base_col: str, alpha: float = 0.05, ) -> pd.DataFrame: """Wilcoxon signed-rank test on paired joint vs. separate win rates.""" keys = ["question_set", "question", "criteria"] joint_mean = joint_df.groupby(keys)[base_col].mean().to_frame("joint") separate_mean = separate_df.groupby(keys)[base_col].mean().to_frame("separate") merged = joint_mean.join(separate_mean, how="inner").dropna().reset_index() def _test(sub: pd.DataFrame) -> pd.Series: a = sub["joint"].to_numpy() b = sub["separate"].to_numpy() diff = b - a if np.allclose(diff, 0.0): stat, p_value = 0.0, 1.0 else: try: result = stats.wilcoxon(a, b) stat = float(result.statistic) # type: ignore[attr-defined] p_value = float(result.pvalue) # type: ignore[attr-defined] except ValueError: stat, p_value = 0.0, 1.0 return pd.Series({ "n": len(sub), "joint_mean_wr": float(a.mean()), "separate_mean_wr": float(b.mean()), "mean_diff": float(diff.mean()), "wilcoxon_stat": stat, "p_value": p_value, "significant": p_value < alpha, }) per_criterion = ( merged.groupby("criteria").apply(_test, include_groups=False).reset_index() ) overall = _test(merged) overall["criteria"] = "ALL" return pd.concat([per_criterion, pd.DataFrame([overall])], ignore_index=True) base_score_col = f"{diff_base}_score" significance_df = paired_judging_significance( differential_df, separate_df, base_col=base_score_col, alpha=alpha ) significance_df["p_value"] = significance_df["p_value"].round(3) significance_df["mean_diff"] = significance_df["mean_diff"].round(4) print_df( cast(pd.DataFrame, significance_df), "Joint vs. Separate — Paired Significance (Wilcoxon; p > alpha ⇒ no systematic difference)", ) overall_p = float( significance_df.loc[significance_df["criteria"] == "ALL", "p_value"].iloc[0] ) if overall_p > alpha: rich_print( f"[green]Overall p = {overall_p:.3f} > alpha = {alpha}: no statistically " f"significant systematic difference between joint and separate judging.[/green]" ) else: rich_print( f"[yellow]Overall p = {overall_p:.3f} <= alpha = {alpha}: the two approaches " f"differ systematically — prefer separate judging.[/yellow]" )

In [ ]:

# Inspect the reasoning where the two approaches DISAGREE, to sanity-check which
# verdict reads as more sound (no ground truth, so this is qualitative).
verdict_labels = {1.0: f"{diff_base} wins", 0.0: "other wins", 0.5: "tie"}
reasoning_keys = ["question_set", "question", "criteria", "trial"]

joint_slim = cast(
    pd.DataFrame, differential_df[[*reasoning_keys, base_score_col, "reasoning"]]
).rename(columns={base_score_col: "joint_score", "reasoning": "joint_reasoning"})
separate_slim = cast(
    pd.DataFrame, separate_df[[*reasoning_keys, base_score_col, "reasoning"]]
).rename(columns={base_score_col: "separate_score", "reasoning": "separate_reasoning"})
paired_reasoning = joint_slim.merge(separate_slim, on=reasoning_keys)
disagreements = paired_reasoning[
    paired_reasoning["joint_score"] != paired_reasoning["separate_score"]
].reset_index(drop=True)

rich_print(
    f"[bold]{len(disagreements)} / {len(paired_reasoning)} verdicts disagree "
    f"({len(disagreements) / len(paired_reasoning):.1%}).[/bold] "
    "Showing a few examples with both rationales:\n"
)

n_examples = min(3, len(disagreements))
for _, row in disagreements.head(n_examples).iterrows():
    rich_print(
        f"[bold cyan]{row['question_set']} · {row['criteria']} · trial {row['trial']}[/bold cyan]"
    )
    rich_print(f"[bold]Question:[/bold] {str(row['question'])[:200]}...")
    rich_print(
        f"[bold]JOINT verdict:[/bold] {verdict_labels[float(row['joint_score'])]}\n"
        f"{row['joint_reasoning']}\n"
    )
    rich_print(
        f"[bold]SEPARATE verdict:[/bold] {verdict_labels[float(row['separate_score'])]}\n"
        f"{row['separate_reasoning']}\n"
    )
    rich_print("[dim]" + "-" * 80 + "[/dim]\n")

# Inspect the reasoning where the two approaches DISAGREE, to sanity-check which # verdict reads as more sound (no ground truth, so this is qualitative). verdict_labels = {1.0: f"{diff_base} wins", 0.0: "other wins", 0.5: "tie"} reasoning_keys = ["question_set", "question", "criteria", "trial"] joint_slim = cast( pd.DataFrame, differential_df[[*reasoning_keys, base_score_col, "reasoning"]] ).rename(columns={base_score_col: "joint_score", "reasoning": "joint_reasoning"}) separate_slim = cast( pd.DataFrame, separate_df[[*reasoning_keys, base_score_col, "reasoning"]] ).rename(columns={base_score_col: "separate_score", "reasoning": "separate_reasoning"}) paired_reasoning = joint_slim.merge(separate_slim, on=reasoning_keys) disagreements = paired_reasoning[ paired_reasoning["joint_score"] != paired_reasoning["separate_score"] ].reset_index(drop=True) rich_print( f"[bold]{len(disagreements)} / {len(paired_reasoning)} verdicts disagree " f"({len(disagreements) / len(paired_reasoning):.1%}).[/bold] " "Showing a few examples with both rationales:\n" ) n_examples = min(3, len(disagreements)) for _, row in disagreements.head(n_examples).iterrows(): rich_print( f"[bold cyan]{row['question_set']} · {row['criteria']} · trial {row['trial']}[/bold cyan]" ) rich_print(f"[bold]Question:[/bold] {str(row['question'])[:200]}...") rich_print( f"[bold]JOINT verdict:[/bold] {verdict_labels[float(row['joint_score'])]}\n" f"{row['joint_reasoning']}\n" ) rich_print( f"[bold]SEPARATE verdict:[/bold] {verdict_labels[float(row['separate_score'])]}\n" f"{row['separate_reasoning']}\n" ) rich_print("[dim]" + "-" * 80 + "[/dim]\n")

Repeat across judge models¶

The joint-vs-separate gap may depend on the judge model. The sweep below re-runs both the joint and separate variants for each model in judge_models and reports, per model, the verdict agreement rate and the overall paired p-value. Edit judge_models to add/remove models.

⚠️ This is expensive: for each model it runs the full joint pass (1 judge call per pair/trial) and the full separate pass (1 extraction + one judge call per criterion). Keep the model list short.

In [ ]:

# Multi-model sweep: per-model detail tables + one combined comparison table.
judge_models = [
    "gpt-5.2",
    "gpt-4.1",
    "gpt-4o",
]  # edit to add/remove judge models


def run_joint_vs_separate_for_model(
    model_name: str, *, concurrent_requests: int = 32
) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Run both joint and separate differential judging for a single judge model."""
    model_config = LLMConfig(
        model=model_name,
        api_key=SecretStr(os.environ["OPENAI_API_KEY"]),
        llm_provider=LLMProvider.OpenAIChat,
        concurrent_requests=concurrent_requests,
        call_args={"temperature": 0.0, "seed": 42},
    )
    model_client = ModelFactory.create_chat_model(model_config)

    joint_parts: list[pd.DataFrame] = []
    separate_parts: list[pd.DataFrame] = []
    for question_set in diff_question_sets:
        for other in diff_others:
            base_answers = pd.read_json(
                f"{diff_input_dir}/{diff_base}/{question_set}_answers.json"
            )
            other_answers = pd.read_json(
                f"{diff_input_dir}/{other}/{question_set}_answers.json"
            )
            joint_part = get_differential_pairwise_scores(
                llm_client=model_client,
                llm_config=model_config,
                base_name=diff_base,
                other_name=other,
                base_answers=base_answers,
                other_answers=other_answers,
                criteria=comparable_criteria,
                trials=diff_trials,
                include_score_id_in_prompt=True,
                question_id_key="question_id",
                question_text_key="question",
            )
            joint_part["question_set"] = question_set
            joint_parts.append(joint_part)

            separate_part = get_separate_differential_scores(
                llm_client=model_client,
                llm_config=model_config,
                base_name=diff_base,
                other_name=other,
                base_answers=base_answers,
                other_answers=other_answers,
                trials=diff_trials,
                include_score_id_in_prompt=True,
                question_id_key="question_id",
                question_text_key="question",
            )
            separate_part["question_set"] = question_set
            separate_parts.append(separate_part)

    return (
        pd.concat(joint_parts, ignore_index=True),
        pd.concat(separate_parts, ignore_index=True),
    )


def report_model_detail(
    model_name: str, joint_df: pd.DataFrame, separate_df: pd.DataFrame
) -> dict[str, object]:
    """Print the per-model detail tables and return the overall summary row."""
    summary_keys = ["question_set", "criteria", "base_name", "other_name"]

    # (1) Win rates: joint vs. separate.
    m_joint_summary = analyze_criteria(joint_df, alpha=alpha)
    m_separate_summary = analyze_criteria(separate_df, alpha=alpha)
    mjs = m_joint_summary.merge(
        m_separate_summary, on=summary_keys, suffixes=("_joint", "_separate")
    )
    mjs["winrate_delta"] = mjs["other_mean_separate"] - mjs["other_mean_joint"]
    win_rate_view = pd.DataFrame({
        "question set": mjs["question_set"].str.replace("data_", "", regex=False),
        "criteria": mjs["criteria"],
        f"{other_short} WR (joint)": mjs["other_mean_joint"].round(4),
        f"{other_short} WR (separate)": mjs["other_mean_separate"].round(4),
        "WR Δ (sep-joint)": mjs["winrate_delta"].round(4),
        "p (joint)": mjs["formatted_corrected_p_value_joint"],
        "p (separate)": mjs["formatted_corrected_p_value_separate"],
    })
    print_df(
        cast(pd.DataFrame, win_rate_view),
        f"[{model_name}] Joint vs. Separate — Win Rates",
    )

    # (2) Verdict agreement per criterion.
    agree_keys = ["question_set", "question", "criteria", "trial"]
    merged_items = joint_df[[*agree_keys, base_score_col]].merge(
        separate_df[[*agree_keys, base_score_col]],
        on=agree_keys,
        suffixes=("_joint", "_separate"),
    )
    merged_items["agree"] = (
        merged_items[f"{base_score_col}_joint"]
        == merged_items[f"{base_score_col}_separate"]
    )
    agreement_view = (
        merged_items
        .groupby("criteria")["agree"]
        .agg(agreement_rate="mean", n="count")
        .reset_index()
    )
    overall_agreement = float(merged_items["agree"].mean())
    agreement_view = pd.concat(
        [
            agreement_view,
            pd.DataFrame([
                {
                    "criteria": "ALL",
                    "agreement_rate": overall_agreement,
                    "n": len(merged_items),
                }
            ]),
        ],
        ignore_index=True,
    )
    agreement_view["agreement_rate"] = agreement_view["agreement_rate"].round(3)
    print_df(
        cast(pd.DataFrame, agreement_view),
        f"[{model_name}] Joint vs. Separate — Verdict Agreement",
    )

    # (3) Paired significance.
    model_sig = paired_judging_significance(
        joint_df, separate_df, base_col=base_score_col, alpha=alpha
    )
    sig_display = model_sig.copy()
    for col in ["joint_mean_wr", "separate_mean_wr", "mean_diff"]:
        sig_display[col] = sig_display[col].round(4)
    sig_display["p_value"] = sig_display["p_value"].round(3)
    print_df(
        cast(
            pd.DataFrame,
            sig_display[
                [
                    "criteria",
                    "n",
                    "joint_mean_wr",
                    "separate_mean_wr",
                    "mean_diff",
                    "p_value",
                    "significant",
                ]
            ],
        ),
        f"[{model_name}] Joint vs. Separate — Paired Significance",
    )

    overall = model_sig[model_sig["criteria"] == "ALL"].iloc[0]
    return {
        "judge_model": model_name,
        "verdict_agreement": round(overall_agreement, 3),
        "joint_mean_wr": round(float(overall["joint_mean_wr"]), 4),
        "separate_mean_wr": round(float(overall["separate_mean_wr"]), 4),
        "overall_p_value": round(float(overall["p_value"]), 3),
        "significant_diff": bool(overall["significant"]),
    }


model_rows: list[dict[str, object]] = []
for model_name in judge_models:
    rich_print(f"[bold]=== Judge model: {model_name} ===[/bold]")
    model_joint_df, model_separate_df = run_joint_vs_separate_for_model(model_name)
    model_rows.append(
        report_model_detail(model_name, model_joint_df, model_separate_df)
    )

# Combined comparison across all models (single shareable table).
model_summary_df = pd.DataFrame(model_rows)
model_summary_df.to_csv(diff_output_dir / "joint_vs_separate_by_model.csv", index=False)
print_df(
    cast(pd.DataFrame, model_summary_df),
    "Joint vs. Separate — All Judge Models (combined)",
)

# Multi-model sweep: per-model detail tables + one combined comparison table. judge_models = [ "gpt-5.2", "gpt-4.1", "gpt-4o", ] # edit to add/remove judge models def run_joint_vs_separate_for_model( model_name: str, *, concurrent_requests: int = 32 ) -> tuple[pd.DataFrame, pd.DataFrame]: """Run both joint and separate differential judging for a single judge model.""" model_config = LLMConfig( model=model_name, api_key=SecretStr(os.environ["OPENAI_API_KEY"]), llm_provider=LLMProvider.OpenAIChat, concurrent_requests=concurrent_requests, call_args={"temperature": 0.0, "seed": 42}, ) model_client = ModelFactory.create_chat_model(model_config) joint_parts: list[pd.DataFrame] = [] separate_parts: list[pd.DataFrame] = [] for question_set in diff_question_sets: for other in diff_others: base_answers = pd.read_json( f"{diff_input_dir}/{diff_base}/{question_set}_answers.json" ) other_answers = pd.read_json( f"{diff_input_dir}/{other}/{question_set}_answers.json" ) joint_part = get_differential_pairwise_scores( llm_client=model_client, llm_config=model_config, base_name=diff_base, other_name=other, base_answers=base_answers, other_answers=other_answers, criteria=comparable_criteria, trials=diff_trials, include_score_id_in_prompt=True, question_id_key="question_id", question_text_key="question", ) joint_part["question_set"] = question_set joint_parts.append(joint_part) separate_part = get_separate_differential_scores( llm_client=model_client, llm_config=model_config, base_name=diff_base, other_name=other, base_answers=base_answers, other_answers=other_answers, trials=diff_trials, include_score_id_in_prompt=True, question_id_key="question_id", question_text_key="question", ) separate_part["question_set"] = question_set separate_parts.append(separate_part) return ( pd.concat(joint_parts, ignore_index=True), pd.concat(separate_parts, ignore_index=True), ) def report_model_detail( model_name: str, joint_df: pd.DataFrame, separate_df: pd.DataFrame ) -> dict[str, object]: """Print the per-model detail tables and return the overall summary row.""" summary_keys = ["question_set", "criteria", "base_name", "other_name"] # (1) Win rates: joint vs. separate. m_joint_summary = analyze_criteria(joint_df, alpha=alpha) m_separate_summary = analyze_criteria(separate_df, alpha=alpha) mjs = m_joint_summary.merge( m_separate_summary, on=summary_keys, suffixes=("_joint", "_separate") ) mjs["winrate_delta"] = mjs["other_mean_separate"] - mjs["other_mean_joint"] win_rate_view = pd.DataFrame({ "question set": mjs["question_set"].str.replace("data_", "", regex=False), "criteria": mjs["criteria"], f"{other_short} WR (joint)": mjs["other_mean_joint"].round(4), f"{other_short} WR (separate)": mjs["other_mean_separate"].round(4), "WR Δ (sep-joint)": mjs["winrate_delta"].round(4), "p (joint)": mjs["formatted_corrected_p_value_joint"], "p (separate)": mjs["formatted_corrected_p_value_separate"], }) print_df( cast(pd.DataFrame, win_rate_view), f"[{model_name}] Joint vs. Separate — Win Rates", ) # (2) Verdict agreement per criterion. agree_keys = ["question_set", "question", "criteria", "trial"] merged_items = joint_df[[*agree_keys, base_score_col]].merge( separate_df[[*agree_keys, base_score_col]], on=agree_keys, suffixes=("_joint", "_separate"), ) merged_items["agree"] = ( merged_items[f"{base_score_col}_joint"] == merged_items[f"{base_score_col}_separate"] ) agreement_view = ( merged_items .groupby("criteria")["agree"] .agg(agreement_rate="mean", n="count") .reset_index() ) overall_agreement = float(merged_items["agree"].mean()) agreement_view = pd.concat( [ agreement_view, pd.DataFrame([ { "criteria": "ALL", "agreement_rate": overall_agreement, "n": len(merged_items), } ]), ], ignore_index=True, ) agreement_view["agreement_rate"] = agreement_view["agreement_rate"].round(3) print_df( cast(pd.DataFrame, agreement_view), f"[{model_name}] Joint vs. Separate — Verdict Agreement", ) # (3) Paired significance. model_sig = paired_judging_significance( joint_df, separate_df, base_col=base_score_col, alpha=alpha ) sig_display = model_sig.copy() for col in ["joint_mean_wr", "separate_mean_wr", "mean_diff"]: sig_display[col] = sig_display[col].round(4) sig_display["p_value"] = sig_display["p_value"].round(3) print_df( cast( pd.DataFrame, sig_display[ [ "criteria", "n", "joint_mean_wr", "separate_mean_wr", "mean_diff", "p_value", "significant", ] ], ), f"[{model_name}] Joint vs. Separate — Paired Significance", ) overall = model_sig[model_sig["criteria"] == "ALL"].iloc[0] return { "judge_model": model_name, "verdict_agreement": round(overall_agreement, 3), "joint_mean_wr": round(float(overall["joint_mean_wr"]), 4), "separate_mean_wr": round(float(overall["separate_mean_wr"]), 4), "overall_p_value": round(float(overall["p_value"]), 3), "significant_diff": bool(overall["significant"]), } model_rows: list[dict[str, object]] = [] for model_name in judge_models: rich_print(f"[bold]=== Judge model: {model_name} ===[/bold]") model_joint_df, model_separate_df = run_joint_vs_separate_for_model(model_name) model_rows.append( report_model_detail(model_name, model_joint_df, model_separate_df) ) # Combined comparison across all models (single shareable table). model_summary_df = pd.DataFrame(model_rows) model_summary_df.to_csv(diff_output_dir / "joint_vs_separate_by_model.csv", index=False) print_df( cast(pd.DataFrame, model_summary_df), "Joint vs. Separate — All Judge Models (combined)", )