skillhub-195-advanced-evaluation/references/implementation-patterns.md

LLM-as-Judge 实现模式

本文档提供了构建生产级 LLM 评估系统的详细实现模式。

模式 1：结构化评估流水线

最可靠的评估系统遵循结构化的流水线来分离关注点：

Input Validation → Criteria Loading → Scoring → Bias Mitigation → Output Formatting

输入验证层

在评估开始之前，验证以下内容：

1. 响应存在性：待评估的响应不能为空
2. 提示词存在性：用于提供上下文的原始提示词不能为空
3. 标准有效性：至少有一条标准，且包含名称和描述
4. 权重归一化：权重之和为 1.0（或对其进行归一化）

def validate_input(response, prompt, criteria):
    if not response or not response.strip():
        raise ValueError("Response cannot be empty")
    if not prompt or not prompt.strip():
        raise ValueError("Prompt cannot be empty")
    if not criteria or len(criteria) == 0:
        raise ValueError("At least one criterion required")
    
    # Normalize weights
    total_weight = sum(c.get('weight', 1) for c in criteria)
    for c in criteria:
        c['weight'] = c.get('weight', 1) / total_weight

标准加载层

标准应从配置中加载，而非硬编码：

class CriteriaLoader:
    def __init__(self, rubric_path=None):
        self.rubrics = self._load_rubrics(rubric_path)
    
    def get_criteria(self, task_type):
        return self.rubrics.get(task_type, self.default_criteria)
    
    def get_rubric(self, criterion_name):
        return self.rubrics.get(criterion_name, {}).get('levels', [])

评分层

评分层负责实际的 LLM 调用：

async def score_response(response, prompt, criteria, rubric, model):
    system_prompt = build_system_prompt(criteria, rubric)
    user_prompt = build_user_prompt(response, prompt, criteria)
    
    result = await generate_text(
        model=model,
        system=system_prompt,
        prompt=user_prompt,
        temperature=0.3  # Lower temperature for consistency
    )
    
    return parse_scores(result.text)

偏差缓解层

对于成对比较，始终包含位置交换：

async def compare_with_bias_mitigation(response_a, response_b, prompt, criteria, model):
    # First pass: A first
    pass1 = await compare_pair(response_a, response_b, prompt, criteria, model)
    
    # Second pass: B first
    pass2 = await compare_pair(response_b, response_a, prompt, criteria, model)
    
    # Map pass2 winner back
    pass2_mapped = map_winner(pass2.winner)  # A→B, B→A, TIE→TIE
    
    # Check consistency
    if pass1.winner == pass2_mapped:
        return {
            'winner': pass1.winner,
            'confidence': (pass1.confidence + pass2.confidence) / 2,
            'consistent': True
        }
    else:
        return {
            'winner': 'TIE',
            'confidence': 0.5,
            'consistent': False
        }

模式 2：分层评估

对于复杂的评估，使用分层方法：

Quick Screen (cheap model) → Detailed Evaluation (expensive model) → Human Review (edge cases)

快速筛选实现

async def quick_screen(response, prompt, threshold=0.7):
    """Fast, cheap screening for obvious passes/fails."""
    result = await generate_text(
        model='gpt-5.2',  # Cheaper model
        prompt=f"Rate 0-1 if this response adequately addresses the prompt:\n\nPrompt: {prompt}\n\nResponse: {response}",
        temperature=0
    )
    score = float(result.text.strip())
    return score, score > threshold

详细评估

async def detailed_evaluation(response, prompt, criteria):
    """Full evaluation for borderline or important cases."""
    result = await generate_text(
        model='gpt-5.2',  # More capable model
        system=DETAILED_EVALUATION_PROMPT,
        prompt=build_detailed_prompt(response, prompt, criteria),
        temperature=0.3
    )
    return parse_detailed_scores(result.text)

模式 3：LLM 评委专家组（PoLL）

对于高风险评估，使用多个模型：

async def poll_evaluation(response, prompt, criteria, models):
    """Aggregate judgments from multiple LLM judges."""
    results = await asyncio.gather(*[
        score_with_model(response, prompt, criteria, model)
        for model in models
    ])
    
    # Aggregate scores
    aggregated = aggregate_scores(results)
    
    # Calculate agreement
    agreement = calculate_agreement(results)
    
    return {
        'scores': aggregated,
        'agreement': agreement,
        'individual_results': results
    }

def aggregate_scores(results):
    """Aggregate scores using median (robust to outliers)."""
    scores = {}
    for criterion in results[0]['scores'].keys():
        criterion_scores = [r['scores'][criterion] for r in results]
        scores[criterion] = {
            'score': statistics.median(criterion_scores),
            'std': statistics.stdev(criterion_scores) if len(criterion_scores) > 1 else 0
        }
    return scores

模式 4：置信度校准

置信度得分应与实际可靠性相匹配：

def calibrate_confidence(raw_confidence, position_consistent, evidence_count):
    """Calibrate confidence based on multiple signals."""
    
    # Base confidence from model output
    calibrated = raw_confidence
    
    # Position consistency is a strong signal
    if not position_consistent:
        calibrated *= 0.6  # Significant reduction
    
    # More evidence = higher confidence
    evidence_factor = min(evidence_count / 3, 1.0)  # Cap at 3 pieces
    calibrated *= (0.7 + 0.3 * evidence_factor)
    
    return min(calibrated, 0.99)  # Never 100% confident

模式 5：输出格式化

始终使用一致的 Schema 返回结构化输出：

@dataclass
class ScoreResult:
    criterion: str
    score: float
    max_score: float
    justification: str
    evidence: List[str]
    improvement: str

@dataclass
class EvaluationResult:
    success: bool
    scores: List[ScoreResult]
    overall_score: float
    weighted_score: float
    summary: Dict[str, Any]
    metadata: Dict[str, Any]

def format_output(scores, metadata) -> EvaluationResult:
    """Format evaluation results consistently."""
    return EvaluationResult(
        success=True,
        scores=scores,
        overall_score=sum(s.score for s in scores) / len(scores),
        weighted_score=calculate_weighted_score(scores),
        summary=generate_summary(scores),
        metadata=metadata
    )

错误处理模式

优雅降级

async def evaluate_with_fallback(response, prompt, criteria):
    try:
        return await full_evaluation(response, prompt, criteria)
    except RateLimitError:
        # Fall back to simpler evaluation
        return await simple_evaluation(response, prompt, criteria)
    except ParseError as e:
        # Return partial results with error flag
        return {
            'success': False,
            'partial_results': e.partial_data,
            'error': str(e)
        }

重试逻辑

async def evaluate_with_retry(response, prompt, criteria, max_retries=3):
    for attempt in range(max_retries):
        try:
            result = await evaluate(response, prompt, criteria)
            if is_valid_result(result):
                return result
        except TransientError:
            await asyncio.sleep(2 ** attempt)  # Exponential backoff
    
    raise EvaluationError("Max retries exceeded")

测试模式

解析单元测试

def test_score_parsing():
    raw_output = '{"scores": [{"criterion": "Accuracy", "score": 4}]}'
    result = parse_scores(raw_output)
    assert result.scores[0].criterion == "Accuracy"
    assert result.scores[0].score == 4

def test_malformed_output():
    raw_output = 'Invalid JSON'
    with pytest.raises(ParseError):
        parse_scores(raw_output)

集成测试（使用真实 API）

@pytest.mark.integration
async def test_full_evaluation_pipeline():
    result = await evaluate(
        response="Water boils at 100°C at sea level.",
        prompt="At what temperature does water boil?",
        criteria=[{"name": "Accuracy", "description": "Factual correctness", "weight": 1}]
    )
    
    assert result.success
    assert len(result.scores) == 1
    assert result.scores[0].score >= 4  # Should score high for accurate response

偏差检测测试

async def test_position_bias_mitigation():
    # Same response in both positions should tie
    result = await compare(
        response_a="Same response",
        response_b="Same response",
        prompt="Test prompt",
        criteria=["quality"],
        swap_positions=True
    )
    
    assert result.winner == "TIE"
    assert result.consistent == True