3장: 검색 품질 메트릭과 평가

def precision_at_k(retrieved: list[int], relevant: set[int], k: int) -> float:
    """
    P@K 계산
    retrieved: 검색 결과 문서 ID 리스트 (순서대로)
    relevant: 관련 문서 ID 집합
    k: 상위 K개
    """
    top_k = retrieved[:k]
    relevant_in_top_k = sum(1 for doc_id in top_k if doc_id in relevant)
    return relevant_in_top_k / k
 
# 예시: 상위 5개 중 3개가 관련 문서
retrieved = [101, 205, 303, 404, 102]
relevant = {101, 102, 303, 507}
print(f"P@5 = {precision_at_k(retrieved, relevant, 5)}")  # 0.6

def recall_at_k(retrieved: list[int], relevant: set[int], k: int) -> float:
    """R@K 계산"""
    top_k = retrieved[:k]
    relevant_in_top_k = sum(1 for doc_id in top_k if doc_id in relevant)
    return relevant_in_top_k / len(relevant) if relevant else 0.0
 
# 전체 관련 문서 4개 중 3개를 상위 5개에서 찾음
print(f"R@5 = {recall_at_k(retrieved, relevant, 5)}")  # 0.75

def reciprocal_rank(retrieved: list[int], relevant: set[int]) -> float:
    """단일 쿼리의 Reciprocal Rank"""
    for i, doc_id in enumerate(retrieved):
        if doc_id in relevant:
            return 1.0 / (i + 1)
    return 0.0
 
def mean_reciprocal_rank(queries_results: list[tuple[list[int], set[int]]]) -> float:
    """MRR: 여러 쿼리의 RR 평균"""
    rr_scores = [reciprocal_rank(ret, rel) for ret, rel in queries_results]
    return sum(rr_scores) / len(rr_scores)
 
# 예시: 3개 쿼리
queries = [
    ([10, 20, 30], {20}),    # 첫 관련 문서가 2위 -> RR = 0.5
    ([40, 50, 60], {40}),    # 첫 관련 문서가 1위 -> RR = 1.0
    ([70, 80, 90], {90}),    # 첫 관련 문서가 3위 -> RR = 0.333
]
print(f"MRR = {mean_reciprocal_rank(queries):.4f}")  # 0.6111

def average_precision(retrieved: list[int], relevant: set[int]) -> float:
    """단일 쿼리의 Average Precision"""
    hits = 0
    sum_precision = 0.0
 
    for i, doc_id in enumerate(retrieved):
        if doc_id in relevant:
            hits += 1
            sum_precision += hits / (i + 1)
 
    return sum_precision / len(relevant) if relevant else 0.0
 
def mean_average_precision(queries_results: list[tuple[list[int], set[int]]]) -> float:
    """MAP: 여러 쿼리의 AP 평균"""
    ap_scores = [average_precision(ret, rel) for ret, rel in queries_results]
    return sum(ap_scores) / len(ap_scores)

import numpy as np
 
def dcg_at_k(relevance_scores: list[float], k: int) -> float:
    """DCG@K: 순위에 따른 감가 누적 이득"""
    relevance = np.array(relevance_scores[:k])
    positions = np.arange(1, len(relevance) + 1)
    return np.sum(relevance / np.log2(positions + 1))
 
def ndcg_at_k(relevance_scores: list[float], k: int) -> float:
    """NDCG@K: 이상적 순서 대비 정규화된 점수"""
    dcg = dcg_at_k(relevance_scores, k)
    ideal_scores = sorted(relevance_scores, reverse=True)
    idcg = dcg_at_k(ideal_scores, k)
    return dcg / idcg if idcg > 0 else 0.0
 
# 예시: 관련성 점수 (3=매우 관련, 2=관련, 1=약간 관련, 0=무관)
scores = [3, 0, 2, 1, 0]
print(f"NDCG@5 = {ndcg_at_k(scores, 5):.4f}")  # 0.8542

from dataclasses import dataclass
 
@dataclass
class RelevanceJudgment:
    query_id: str
    query_text: str
    doc_id: str
    relevance: int  # 0: 무관, 1: 약간 관련, 2: 관련, 3: 매우 관련
 
# 평가 데이터셋 예시
judgments = [
    RelevanceJudgment("q1", "파이썬 웹 프레임워크", "d101", 3),
    RelevanceJudgment("q1", "파이썬 웹 프레임워크", "d102", 2),
    RelevanceJudgment("q1", "파이썬 웹 프레임워크", "d103", 0),
    RelevanceJudgment("q2", "데이터베이스 인덱스", "d201", 3),
    RelevanceJudgment("q2", "데이터베이스 인덱스", "d202", 1),
]

def evaluate_search_system(search_fn, judgments: list[RelevanceJudgment], k: int = 10):
    """검색 시스템 오프라인 평가"""
    queries = {}
    for j in judgments:
        if j.query_id not in queries:
            queries[j.query_id] = {"text": j.query_text, "relevance": {}}
        queries[j.query_id]["relevance"][j.doc_id] = j.relevance
 
    ndcg_scores = []
    for query_id, query_data in queries.items():
        results = search_fn(query_data["text"], k=k)
        relevance_scores = [
            query_data["relevance"].get(doc_id, 0) for doc_id in results
        ]
        ndcg_scores.append(ndcg_at_k(relevance_scores, k))
 
    return {
        "mean_ndcg": sum(ndcg_scores) / len(ndcg_scores),
        "per_query_ndcg": dict(zip(queries.keys(), ndcg_scores)),
    }

def team_draft_interleave(results_a: list, results_b: list, k: int) -> tuple[list, dict]:
    """팀 드래프트 인터리빙"""
    interleaved = []
    team_a = set()
    team_b = set()
    idx_a, idx_b = 0, 0
 
    while len(interleaved) < k:
        if len(team_a) <= len(team_b):
            # A에서 선택
            while idx_a < len(results_a) and results_a[idx_a] in interleaved:
                idx_a += 1
            if idx_a < len(results_a):
                interleaved.append(results_a[idx_a])
                team_a.add(results_a[idx_a])
                idx_a += 1
        else:
            # B에서 선택
            while idx_b < len(results_b) and results_b[idx_b] in interleaved:
                idx_b += 1
            if idx_b < len(results_b):
                interleaved.append(results_b[idx_b])
                team_b.add(results_b[idx_b])
                idx_b += 1
 
    return interleaved, {"team_a": team_a, "team_b": team_b}