Verify solution quality through pairwise comparison rather than pointwise scoring. Implement topology coverage and Swiss refinement to allocate verification compute to uncertain pairs, improving calibration and reducing verification overhead.
Verify solution quality through pairwise comparison rather than pointwise scoring. Implement topology coverage and Swiss refinement to allocate verification compute to uncertain pairs, improving calibration and reducing verification overhead.
V_1: Pairwise Self-Verification for Parallel Reasoning
Pointwise solution verification (score each candidate independently) suffers from calibration collapse: models lack comparative context and assign arbitrary absolute scores. Pairwise verification (compare solutions head-to-head) provides superior discrimination by grounding judgments in relative comparisons. V_1 extends this insight with uncertainty-guided allocation: allocate verification compute to uncertain pairs using Bradley-Terry models.
The core innovation treats solution verification as a ranking problem with adaptive compute allocation. Pairs with similar quality scores yield high information gain and receive more verification passes; pairs with clear winners are skipped or verified once.
Core Concept
V_1 implements two coordinated mechanisms:
Pairwise Ranking: Models compare any two solutions, assigning probabilistic judgments via Bradley-Terry model to provide calibrated confidence
Uncertainty-Guided Allocation: Use confidence magnitude as proxy for information gain; allocate extra compute passes to uncertain pairs
This combination achieves superior calibration while reducing total verification compute compared to pointwise approaches.
Architecture Overview
Input: k candidate solutions from generation models
Topology Generation: Ensure all pairs covered (complete tournament graph or subset)
Swiss Refinement: Iteratively allocate verification passes to uncertain pairs
Bradley-Terry Model: Accumulate pairwise judgments into global ranking
Output: Ranked solution list with confidence intervals
Implementation Steps
Step 1: Implement pairwise comparison oracle
Create a verifier that compares two solutions and returns confidence-calibrated judgments.
f"""
Question/Task: {prompt}
Solution A:
{solution_a}
Solution B:
{solution_b}
Which solution is better? Respond with: BETTER_A or BETTER_B, and confidence 0-100.
"""
self
50
# Parse response
if
'BETTER_A'
in
0
elif
'BETTER_B'
in
1
else
0
1
# Extract confidence (normalize to [0.5, 1.0])
try
int
''
filter
str
'confidence'
1
0.5
100
0.5
# Map [0, 100] → [0.5, 1.0]
except
0.75
# Default if parsing fails
return
def
batch_compare
self, solution_pairs, prompt
"""Compare multiple pairs in parallel."""
for
in
self
return
Step 2: Design comparison topology
Organize comparisons to ensure coverage while minimizing total comparisons.
classComparisonTopology:
"""Manage comparison graph to ensure all solutions are ranked."""def__init__(self, num_solutions):
self.num_solutions = num_solutions
self.comparisons = [] # List of (idx_a, idx_b) pairsself.comparison_count = {} # Track how many times each pair compareddefround_robin_topology(self):
"""
Round-robin: each solution compared with 2-3 others.
Ensures coverage with ~2K comparisons per K solutions.
"""
topology = []
for i inrange(self.num_solutions):
# Compare solution i with next 2-3 solutions (circular)for offset in [1, 2]:
j = (i + offset) % self.num_solutions
if (i, j) notin topology and (j, i) notin topology:
topology.append((i, j))
return topology
defcomplete_tournament_topology(self):
"""
Complete tournament: every pair compared once.
Expensive (~K²/2 comparisons) but highest quality.
Use for k ≤ 10.
"""
topology = []
for i inrange(self.num_solutions):
for j inrange(i + 1, self.num_solutions):
topology.append((i, j))
return topology
defadaptive_topology(self, previous_rankings):
"""
Adaptive: focus on pairs with similar quality scores.
Skip comparisons between clearly different solutions.
"""
topology = []
quality_scores = [r['score'] for r in previous_rankings]
for i inrange(self.num_solutions):
for j inrange(i + 1, self.num_solutions):
score_diff = abs(quality_scores[i] - quality_scores[j])
# Include pair if scores are close (uncertain rank)if score_diff < 0.3: # Threshold
topology.append((i, j))
return topology
Step 3: Implement Bradley-Terry model for ranking
Accumulate pairwise judgments into calibrated global ranking.
import numpy as np
classBradleyTerryRanker:
"""
Bradley-Terry model: models pairwise comparison outcomes.
P(A beats B) = λ_A / (λ_A + λ_B), where λ_i is strength parameter.
"""def__init__(self, num_solutions):
self.num_solutions = num_solutions
self.strengths = np.ones(num_solutions) # Initial uniform strengthsself.win_counts = np.zeros(num_solutions) # Wins per solutionself.comparison_counts = np.zeros((num_solutions, num_solutions))
defupdate_from_comparison(self, winner_idx, loser_idx, confidence):
"""Update strength parameters from a single comparison."""self.win_counts[winner_idx] += confidence
# Update comparison matrix for information trackingself.comparison_counts[winner_idx][loser_idx] += 1self.comparison_counts[loser_idx][winner_idx] += 1deffit_strengths(self, num_iterations=10):
"""
EM-style fitting: iteratively estimate strength parameters.
λ_i ∝ (wins_i) / (expected_matchups_i)
"""for iteration inrange(num_iterations):
# E-step: expected wins
new_strengths = np.zeros(self.num_solutions)
for i inrange(self.num_solutions):
expected_wins = 0.0for j inrange(self.num_solutions):
if i != j:
# Probability i beats j under current model
p_ij = self.strengths[i] / (self.strengths[i] + self.strengths[j])
# Update with observed comparisonifself.comparison_counts[i][j] > 0:
expected_wins += (self.win_counts[i] * p_ij)
new_strengths[i] = expected_wins + 1e-8# Avoid zero# Normalizeself.strengths = new_strengths / new_strengths.sum() * len(new_strengths)
defget_ranking(self):
"""Return solutions ranked by fitted strength parameters."""
indices = np.argsort(-self.strengths) # Descending order
scores = self.strengths[indices]
return [
{'solution_idx': idx, 'score': float(scores[rank])}
for rank, idx inenumerate(indices)
]
defget_uncertainty(self, idx_a, idx_b):
"""
Estimate uncertainty for a specific pair.
High uncertainty when strengths are similar.
"""
strength_diff = abs(self.strengths[idx_a] - self.strengths[idx_b])
strength_sum = self.strengths[idx_a] + self.strengths[idx_b]
# Normalize difference to [0, 1]
uncertainty = 1.0 - (strength_diff / (strength_sum + 1e-8))
return uncertainty
Step 4: Swiss refinement for adaptive allocation
Iteratively identify uncertain pairs and allocate more verification passes.
defswiss_refinement(solutions, verifier, prompt, max_comparison_budget=100):
"""
Swiss-system refinement: iteratively identify uncertain pairs,
allocate extra comparison passes to them.
"""
k = len(solutions)
ranker = BradleyTerryRanker(k)
# Initial topology: round-robin to establish baseline ranking
topology = ComparisonTopology(k)
initial_pairs = topology.round_robin_topology()
comparison_budget_used = 0for pair_idx, (i, j) inenumerate(initial_pairs):
if comparison_budget_used >= max_comparison_budget:
break
winner, confidence = verifier.compare(solutions[i], solutions[j], prompt)
ranker.update_from_comparison(winner, 1 - winner, confidence)
comparison_budget_used += 1# Fit initial ranking
ranker.fit_strengths(num_iterations=5)
# Refinement phase: allocate extra comparisons to uncertain pairsfor refinement_round inrange(3):
# Find most uncertain pairs
uncertain_pairs = []
for i inrange(k):
for j inrange(i + 1, k):
uncertainty = ranker.get_uncertainty(i, j)
uncertain_pairs.append((uncertainty, i, j))
# Sort by uncertainty (descending)
uncertain_pairs.sort(reverse=True)
# Compare top uncertain pairsfor uncertainty, i, j in uncertain_pairs[:5]: # Top 5 uncertainif comparison_budget_used >= max_comparison_budget:
break
winner, confidence = verifier.compare(solutions[i], solutions[j], prompt)
ranker.update_from_comparison(winner, 1 - winner, confidence)
comparison_budget_used += 1# Refit
ranker.fit_strengths(num_iterations=3)
return ranker.get_ranking(), comparison_budget_used
Step 5: Integration with generation and training
Combine parallel generation with verification and optional training.
defgenerate_and_verify(generator, verifier, prompt, num_candidates=8):
"""
Generate multiple solutions in parallel, then rank via pairwise verification.
"""# Parallel generation (can be distributed)
solutions = [
generator.generate(prompt, temperature=0.7)
for _ inrange(num_candidates)
]
# Pairwise verification with Swiss refinement
ranking, budget_used = swiss_refinement(
solutions,
verifier,
prompt,
max_comparison_budget=30
)
return {
'ranking': ranking,
'solutions': [solutions[r['solution_idx']] for r in ranking],
'verifications': budget_used,
'efficiency': (num_candidates * (num_candidates - 1) / 2) / budget_used
}
deftrain_with_pairwise_ranking(generator, verifier, train_prompts,
num_iterations=1000):
"""
Train generator using pairwise verification feedback.
PairRL: co-evolve generation and verification capabilities.
"""
optimizer = torch.optim.AdamW(generator.parameters(), lr=1e-4)
for iteration inrange(num_iterations):
prompt = random.choice(train_prompts)
# Generate candidates
solutions = [
generator.generate(prompt, temperature=0.7)
for _ inrange(4)
]
# Verify ranking
ranking, _ = swiss_refinement(solutions, verifier, prompt,
max_comparison_budget=10)
# Policy gradient: encourage generating top-ranked solution
top_solution = ranking[0]['solution_idx']
logprob_top = generator.compute_logprob(prompt, solutions[top_solution])
# Bottom-ranked solution (negative example)
bottom_solution = ranking[-1]['solution_idx']
logprob_bottom = generator.compute_logprob(prompt, solutions[bottom_solution])
# Loss: margin between top and bottom
loss = -logprob_top + logprob_bottom
optimizer.zero_grad()
loss.backward()
optimizer.step()
if (iteration + 1) % 100 == 0:
print(f"Iteration {iteration + 1}: Loss = {loss.item():.4f}")
return generator
Step 6: Evaluation on reasoning tasks
Benchmark on tasks with multiple valid solutions requiring careful discrimination.
