Калибровка моделей-судей по оценкам людей

Why Calibration Is Non-Negotiable

Building a Calibration Dataset

# Calibration dataset structure:
# [
#   {
#     'question': 'What causes inflation?',
#     'response': '...',
#     'human_scores': [4, 4, 3, 5, 4],  # 5 raters
#     'human_consensus': 4,              # mean or mode
#     'rater_ids': ['r1', 'r2', 'r3', 'r4', 'r5']
#   },
#   ...
# ]

# Typical calibration set composition:
# 30% excellent responses (score 4-5)
# 40% adequate responses (score 2-4)
# 30% poor responses (score 1-2)
# Include adversarial / edge cases

Measuring Inter-Rater Agreement

from sklearn.metrics import cohen_kappa_score
import numpy as np

def measure_inter_rater_agreement(rater1_scores: list, rater2_scores: list) -> dict:
    kappa = cohen_kappa_score(rater1_scores, rater2_scores, weights='linear')
    exact_agreement = sum(a == b for a, b in zip(rater1_scores, rater2_scores)) / len(rater1_scores)
    adjacent_agreement = sum(abs(a - b) <= 1 for a, b in zip(rater1_scores, rater2_scores)) / len(rater1_scores)
    return {
        'weighted_kappa': round(kappa, 3),
        'exact_agreement': round(exact_agreement, 3),
        'adjacent_agreement': round(adjacent_agreement, 3),
        'acceptable': kappa >= 0.6
    }

Running the Judge on Calibration Data

async def run_judge_on_calibration(calibration_set: list) -> list:
    pairs = []
    for item in calibration_set:
        judge_result = await async_judge(item['question'], item['response'])
        pairs.append({
            'question': item['question'],
            'human': item['human_consensus'],
            'judge': judge_result.correctness,
            'judge_rationale': judge_result.rationale
        })
    return pairs

Computing Judge-Human Correlation

from scipy.stats import pearsonr, spearmanr
import numpy as np

def calibration_metrics(pairs: list) -> dict:
    humans = [p['human'] for p in pairs]
    judges = [p['judge'] for p in pairs]
    pearson_r, p_val = pearsonr(humans, judges)
    spearman_r, _ = spearmanr(humans, judges)
    mae = np.mean([abs(h - j) for h, j in zip(humans, judges)])
    return {
        'pearson_r': round(pearson_r, 3),
        'spearman_r': round(spearman_r, 3),
        'p_value': round(p_val, 5),
        'mae': round(mae, 3),
        'reliable': pearson_r >= 0.7 and mae <= 0.8
    }

Identifying Systematic Biases

import numpy as np
from scipy import stats

def detect_score_bias(pairs: list) -> dict:
    humans = np.array([p['human'] for p in pairs])
    judges = np.array([p['judge'] for p in pairs])
    slope, intercept, r, p, se = stats.linregress(judges, humans)
    bias = np.mean(judges - humans)  # positive = judge over-scores
    return {
        'mean_bias': round(bias, 3),  # > 0 means judge inflates
        'calibration_slope': round(slope, 3),
        'calibration_intercept': round(intercept, 3),
        # Corrected score = slope * judge_score + intercept
        'correction_formula': f'human_score ≈ {slope:.2f} * judge + {intercept:.2f}'
    }

Applying Calibration Correction

def calibrate_score(raw_judge_score: float, slope: float, intercept: float,
                    min_score: int = 1, max_score: int = 5) -> float:
    corrected = slope * raw_judge_score + intercept
    return max(min_score, min(max_score, round(corrected, 1)))

# Example: if judge inflates by 0.5 points on average
# slope=0.85, intercept=0.3
# raw_score=4 -> corrected = 0.85*4 + 0.3 = 3.7

Finding Systematic Error Patterns

from collections import defaultdict

def error_by_category(pairs: list) -> dict:
    by_cat = defaultdict(list)
    for p in pairs:
        error = abs(p['judge'] - p['human'])
        by_cat[p.get('category', 'unknown')].append(error)
    return {
        cat: {
            'mean_error': round(sum(errs)/len(errs), 2),
            'max_error': max(errs),
            'n': len(errs)
        }
        for cat, errs in by_cat.items()
    }

Re-calibrating as Data Grows

def run_calibration_pipeline(calibration_data: list) -> dict:
    # 1. Measure human inter-rater agreement
    ira = measure_inter_rater_agreement(
        [d['rater_1'] for d in calibration_data],
        [d['rater_2'] for d in calibration_data]
    )
    # 2. Run judge on all items
    pairs = run_judge_on_calibration(calibration_data)
    # 3. Compute correlation metrics
    metrics = calibration_metrics(pairs)
    # 4. Detect bias
    bias = detect_score_bias(pairs)
    return {'ira': ira, 'metrics': metrics, 'bias': bias}

Documenting Your Judge Configuration

# judge_config.yaml
# judge_model: gpt-4o-2025-01-01
# judge_prompt_version: v3.2
# calibration_date: 2026-05-15
# calibration_set_size: 312
# calibration_metrics:
#   pearson_r: 0.81
#   spearman_r: 0.79
#   mae: 0.54
# bias_correction:
#   slope: 0.88
#   intercept: 0.45
# next_recalibration: 2026-08-15

Dimension-Specific Judge Prompts

async def multi_dimension_judge(question: str, answer: str) -> dict:
    # Run each dimension as a separate judge call
    correctness, helpfulness, clarity = await asyncio.gather(
        judge_single_dimension(question, answer, 'correctness'),
        judge_single_dimension(question, answer, 'helpfulness'),
        judge_single_dimension(question, answer, 'clarity')
    )
    return {
        'correctness': correctness,
        'helpfulness': helpfulness,
        'clarity': clarity,
        'composite': 0.5 * correctness + 0.3 * helpfulness + 0.2 * clarity
    }

Quick Check

Lesson Recap