LLM-as-Judge: Reliability, Debiasing, and Training Judge Models

# Detecting and quantifying position bias in an LLM judge
from openai import OpenAI
import json

client = OpenAI()
JUDGE_MODEL = "gpt-4o"

def pairwise_judge(
 question: str,
 response_a: str,
 response_b: str,
 judge_model: str = JUDGE_MODEL,
) -> str:
 """Ask the judge to select the better response. Returns 'A' or 'B'."""
 prompt = f"""You are an impartial judge. Given a question and two responses,
select which response is better. Output only 'A' or 'B'.

Question: {question}

Response A: {response_a}

Response B: {response_b}

Better response:"""

 result = client.chat.completions.create(
 model=judge_model,
 messages=[{"role": "user", "content": prompt}],
 max_tokens=1,
 temperature=0.0,
 )
 return result.choices[0].message.content.strip()

def detect_position_bias(
 question: str,
 response_1: str,
 response_2: str,
) -> dict:
 """Run pairwise comparison in both orders to detect position bias."""
 # Order 1: response_1 as A, response_2 as B
 verdict_1 = pairwise_judge(question, response_1, response_2)

 # Order 2: response_2 as A, response_1 as B
 verdict_2 = pairwise_judge(question, response_2, response_1)

 # Check consistency: if the judge is unbiased, the verdicts should
 # point to the same underlying response regardless of position
 consistent = (
 (verdict_1 == "A" and verdict_2 == "B") # Both prefer response_1
 or (verdict_1 == "B" and verdict_2 == "A") # Both prefer response_2
 )

 return {
 "order_1_verdict": verdict_1,
 "order_2_verdict": verdict_2,
 "consistent": consistent,
 "position_bias_detected": not consistent,
 }

# Run on a batch and compute position bias rate
bias_results = []
for sample in eval_samples:
 result = detect_position_bias(sample["q"], sample["r1"], sample["r2"])
 bias_results.append(result)

bias_rate = sum(1 for r in bias_results if r["position_bias_detected"]) / len(bias_results)
print(f"Position bias rate: {bias_rate:.1%}")
# Typical range for GPT-4: 10-25% of pairwise comparisons are inconsistent

# G-Eval: chain-of-thought scoring with probability weighting
from openai import OpenAI
import numpy as np

client = OpenAI()

GEVAL_COHERENCE_PROMPT = """You will be given a summary of a news article.
Your task is to rate the summary on coherence (1-5).

Evaluation Criteria:
Coherence (1-5): The collective quality of all sentences. The summary
should be well-structured and well-organized. It should not just be
a heap of related information, but should build from sentence to
sentence to a coherent body of information about a topic.

Evaluation Steps:
1. Read the summary carefully.
2. Check if the sentences logically follow each other.
3. Check if there is a clear topic progression.
4. Check if the summary has a clear beginning, middle, and end.
5. Assign a score from 1 to 5.

Summary: {summary}

After providing your evaluation steps, output your score (1-5):"""

def geval_score(
 summary: str,
 prompt_template: str = GEVAL_COHERENCE_PROMPT,
 model: str = "gpt-4o",
) -> dict:
 """Compute G-Eval score with probability weighting."""
 prompt = prompt_template.format(summary=summary)

 response = client.chat.completions.create(
 model=model,
 messages=[{"role": "user", "content": prompt}],
 max_tokens=512,
 temperature=0.0,
 logprobs=True,
 top_logprobs=5,
 )

 content = response.choices[0].message.content
 logprobs = response.choices[0].logprobs

 # Extract probability distribution over score tokens (1-5)
 # from the last token's logprobs
 score_probs = {}
 if logprobs and logprobs.content:
 last_token_logprobs = logprobs.content[-1].top_logprobs
 for lp in last_token_logprobs:
 if lp.token.strip() in ["1", "2", "3", "4", "5"]:
 score_probs[int(lp.token.strip())] = np.exp(lp.logprob)

 # Compute probability-weighted score
 if score_probs:
 total_prob = sum(score_probs.values())
 weighted_score = sum(
 score * (prob / total_prob)
 for score, prob in score_probs.items()
 )
 else:
 # Fallback: extract integer score from text
 weighted_score = float(content.strip()[-1])

 return {
 "chain_of_thought": content,
 "score_distribution": score_probs,
 "weighted_score": round(weighted_score, 3),
 }

# Prometheus 2: rubric-based evaluation with an open-source judge
# Install: pip install transformers torch
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

model_name = "prometheus-eval/prometheus-7b-v2.0"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
 model_name,
 torch_dtype=torch.bfloat16,
 device_map="auto",
)

DIRECT_ASSESSMENT_TEMPLATE = """###Task Description:
An instruction (might include an Input inside it), a response to evaluate,
a reference answer that gets a score of 5, and a score rubric representing
the evaluation criteria are given.
1. Write a detailed feedback that assesses the quality of the response
 strictly based on the given score rubric.
2. After writing the feedback, write a score that is an integer between
 1 and 5. Refer to the score rubric.

###Instruction:
{instruction}

###Response to evaluate:
{response}

###Reference Answer (Score 5):
{reference}

###Score Rubric:
[{criteria}]
Score 1: {score_1_desc}
Score 2: {score_2_desc}
Score 3: {score_3_desc}
Score 4: {score_4_desc}
Score 5: {score_5_desc}

###Feedback:"""

def prometheus_evaluate(
 instruction: str,
 response: str,
 reference: str,
 criteria: str,
 rubric: dict,
) -> dict:
 """Run Prometheus 2 direct assessment."""
 prompt = DIRECT_ASSESSMENT_TEMPLATE.format(
 instruction=instruction,
 response=response,
 reference=reference,
 criteria=criteria,
 score_1_desc=rubric[1],
 score_2_desc=rubric[2],
 score_3_desc=rubric[3],
 score_4_desc=rubric[4],
 score_5_desc=rubric[5],
 )

 inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
 with torch.no_grad():
 outputs = model.generate(
 **inputs,
 max_new_tokens=512,
 temperature=0.0,
 do_sample=False,
 )

 generated = tokenizer.decode(outputs[0][inputs.input_ids.shape[1]:], skip_special_tokens=True)

 # Parse feedback and score from generated text
 feedback = generated.strip()
 score = None
 for line in reversed(feedback.split("\n")):
 line = line.strip()
 if line and line[0].isdigit():
 score = int(line[0])
 break

 return {"feedback": feedback, "score": score}

# AlpacaEval length-controlled win rate debiasing
# Install: pip install alpaca-eval
import numpy as np
from sklearn.linear_model import LogisticRegression

def compute_length_controlled_winrate(
 wins: list[bool],
 model_lengths: list[int],
 reference_lengths: list[int],
) -> dict:
 """Compute length-controlled win rate using logistic regression.

 The idea: fit a model that predicts win/loss from length difference,
 then compute the expected win rate at zero length difference.
 """
 # Feature: log ratio of model length to reference length
 length_ratios = np.array([
 np.log(ml / rl) for ml, rl in zip(model_lengths, reference_lengths)
 ]).reshape(-1, 1)

 wins_array = np.array(wins, dtype=int)

 # Raw win rate (length-biased)
 raw_winrate = wins_array.mean()

 # Fit logistic regression: P(win) = sigmoid(a * log_length_ratio + b)
 clf = LogisticRegression(fit_intercept=True)
 clf.fit(length_ratios, wins_array)

 # Length-controlled win rate: predict at zero length difference
 lc_winrate = clf.predict_proba(np.array([[0.0]]))[0][1]

 # Length coefficient: positive means length bias exists
 length_coefficient = clf.coef_[0][0]

 return {
 "raw_winrate": round(raw_winrate, 4),
 "lc_winrate": round(lc_winrate, 4),
 "length_coefficient": round(length_coefficient, 4),
 "length_bias_detected": length_coefficient > 0.1,
 "debiasing_effect": round(raw_winrate - lc_winrate, 4),
 }

# Example: a model with high raw win rate but large length bias
# results = compute_length_controlled_winrate(
# wins=[True]*70 + [False]*30,
# model_lengths=[500]*70 + [200]*30, # Wins are mostly on longer outputs
# reference_lengths=[300]*100,
# )
# raw_winrate: 0.70, lc_winrate: 0.55, debiasing_effect: 0.15

# Meta-evaluation: measuring LLM judge reliability
import numpy as np
from scipy.stats import kendalltau
from sklearn.metrics import cohen_kappa_score

def meta_evaluate_judge(
 judge_scores: list[int],
 human_scores: list[int],
 judge_scores_swapped: list[int], # Scores with response order swapped
) -> dict:
 """Comprehensive meta-evaluation of an LLM judge."""

 # 1. Inter-annotator agreement with humans
 kappa = cohen_kappa_score(judge_scores, human_scores)
 tau, tau_pvalue = kendalltau(judge_scores, human_scores)

 # 2. Consistency under order swap
 # For pairwise judges: how often does the verdict flip when order changes?
 n_samples = len(judge_scores)
 consistency_count = sum(
 1 for j, js in zip(judge_scores, judge_scores_swapped)
 if j == js # Same absolute score regardless of presentation order
 )
 swap_consistency = consistency_count / n_samples

 # 3. Rank inversion rate
 # Count pairs where judge ranking contradicts human ranking
 inversions = 0
 total_pairs = 0
 for i in range(n_samples):
 for j in range(i + 1, n_samples):
 if human_scores[i] != human_scores[j]:
 total_pairs += 1
 human_order = human_scores[i] > human_scores[j]
 judge_order = judge_scores[i] > judge_scores[j]
 if human_order != judge_order:
 inversions += 1

 inversion_rate = inversions / total_pairs if total_pairs > 0 else 0.0

 return {
 "cohens_kappa": round(kappa, 3),
 "kendalls_tau": round(tau, 3),
 "tau_pvalue": round(tau_pvalue, 4),
 "swap_consistency": round(swap_consistency, 3),
 "rank_inversion_rate": round(inversion_rate, 3),
 "reliability_grade": (
 "high" if kappa > 0.6 and swap_consistency > 0.85 else
 "moderate" if kappa > 0.4 and swap_consistency > 0.70 else
 "low"
 ),
 }