P7-5.5 검색 평가셋으로 근거 품질을 다시 점검하기¶

import csv
from collections import Counter
from pathlib import Path

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

document_path = Path("docs/assets/part-07/chapter-05/p7-5-rag-documents.csv")
case_path = Path("docs/assets/part-07/chapter-05/p7-5-boundary-cases.csv")

documents = list(csv.DictReader(document_path.open(encoding="utf-8")))
cases = list(csv.DictReader(case_path.open(encoding="utf-8")))

# 조작 변수: 후보 수와 낮은 점수 기준을 바꾸면 mismatch 구조가 달라진다.
top_k = 3
min_score = 0.16
margin_threshold = 0.04

vectorizer = TfidfVectorizer(ngram_range=(1, 2), min_df=1)
document_matrix = vectorizer.fit_transform([document["text"] for document in documents])
query_matrix = vectorizer.transform([case["question"] for case in cases])
similarity_matrix = cosine_similarity(query_matrix, document_matrix)

def classify_search_state(case, retrieved_documents):
    question = case["question"]
    top_score = retrieved_documents[0]["score"]
    second_score = retrieved_documents[1]["score"] if len(retrieved_documents) > 1 else 0
    margin = top_score - second_score
    joined_text = " ".join(document["text"] for document in retrieved_documents)

    if top_score < min_score:
        return "문서 범위 밖"
    if any(word in joined_text for word in ["충돌", "최신 정책", "예전 FAQ", "버전"]):
        return "근거 충돌"
    if any(word in question for word in ["항상", "모든", "완전히", "해결되는가"]):
        return "답변 과장 위험"
    if margin < margin_threshold:
        return "근거 재검토"
    return "근거 기반 답변"

records = []

for case_index, case in enumerate(cases):
    ranking = similarity_matrix[case_index].argsort()[::-1][:top_k]
    retrieved_documents = [
        {
            "doc_id": documents[document_index]["doc_id"],
            "text": documents[document_index]["text"],
            "score": round(float(similarity_matrix[case_index, document_index]), 3),
        }
        for document_index in ranking
    ]

    predicted_state = classify_search_state(case, retrieved_documents)
    records.append(
        {
            "case_id": case["case_id"],
            "expected_state": case["expected_state"],
            "predicted_state": predicted_state,
            "match": predicted_state == case["expected_state"],
            "failure_stage": case["failure_stage"],
            "top_doc": retrieved_documents[0]["doc_id"],
            "top_score": retrieved_documents[0]["score"],
            "focus": case["focus"],
        }
    )

summary = {
    "평가 질문 수": len(records),
    "상태 일치 수": sum(record["match"] for record in records),
    "상태 불일치 수": sum(not record["match"] for record in records),
    "예측 상태 분포": dict(Counter(record["predicted_state"] for record in records)),
}

mismatches = [record for record in records if not record["match"]]
priority_reviews = sorted(
    mismatches,
    key=lambda record: (record["failure_stage"] != "검색 단계", -record["top_score"]),
)[:6]

print("검색 평가 요약 =", summary)
print("먼저 검토할 mismatch =")
for record in priority_reviews:
    print(record)

검색 평가 요약 = {'평가 질문 수': 36, '상태 일치 수': 12, '상태 불일치 수': 24, '예측 상태 분포': {'근거 기반 답변': 10, '문서 범위 밖': 5, '답변 과장 위험': 9, '근거 충돌': 12}}
먼저 검토할 mismatch =
{'case_id': '연습-11', 'expected_state': '문서 범위 밖', 'predicted_state': '근거 충돌', 'match': False, 'failure_stage': '검색 단계', 'top_doc': '문서-13', 'top_score': 0.38, 'focus': '근거 부족 거절'}
{'case_id': '연습-19', 'expected_state': '문서 범위 밖', 'predicted_state': '근거 충돌', 'match': False, 'failure_stage': '검색 단계', 'top_doc': '문서-13', 'top_score': 0.38, 'focus': '근거 부족 거절'}
{'case_id': '연습-07', 'expected_state': '검색 실패', 'predicted_state': '근거 기반 답변', 'match': False, 'failure_stage': '검색 단계', 'top_doc': '문서-11', 'top_score': 0.274, 'focus': '문서 집합에는 관련 내용이 있지만 현재 검색 규칙이 표현 차이를 놓치는 경우'}
...