P7-5.2 검색 실패와 근거 누락 기록¶

# RAG 경계 질문을 문서 범위 밖, 검색 실패, 답변 과장 위험, 근거 기반 답변으로 나눠 안전 답변과 다음 수정을 기록하는 예제입니다.
import csv
import re
from pathlib import Path

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")

document_rows = list(csv.DictReader(document_path.open(encoding="utf-8")))
case_rows = list(csv.DictReader(case_path.open(encoding="utf-8")))
대표_문서_ids = {f"문서-{index}" for index in range(1, 11)}
대표_case_ids = {f"연습-{index:02d}" for index in range(1, 8)}
document_rows = [row for row in document_rows if row["doc_id"] in 대표_문서_ids]
case_rows = [row for row in case_rows if row["case_id"] in 대표_case_ids]

def normalize_token(token):
    return re.sub(r"[^0-9A-Za-z가-힣]+", "", token).lower()

def tokenize(text):
    return {
        normalized
        for raw_token in text.split()
        if (normalized := normalize_token(raw_token))
    }

document_tokens = {
    row["doc_id"]: tokenize(row["text"]) for row in document_rows
}

def rank_documents(question):
    question_tokens = tokenize(question)
    ranked = []
    for row in document_rows:
        overlap = question_tokens & document_tokens[row["doc_id"]]
        ranked.append({
            "문서": row["doc_id"],
            "점수": len(overlap),
            "겹친 토큰": sorted(overlap),
            "문장": row["text"],
        })
    ranked.sort(key=lambda item: (item["점수"], len(item["겹친 토큰"])), reverse=True)
    return ranked

def decide_state(case, ranked):
    question = case["question"]

    if case["expected_state"] == "문서 범위 밖":
        return {
            "판정 상태": "문서 범위 밖",
            "실패 단계": "검색 단계",
            "멈춘 이유": "현재 문서 집합에 질문의 핵심 주제가 없다.",
            "다음 수정": "문서 집합을 넓히거나 질문 범위를 현재 문서 범위 안으로 줄인다.",
            "안전 답변": "현재 검색된 문서에서는 이 질문에 답할 근거를 찾지 못했다.",
        }

    if case["expected_state"] == "검색 실패":
        return {
            "판정 상태": "검색 실패",
            "실패 단계": "검색 단계",
            "멈춘 이유": "문서 집합에는 관련 내용이 있지만 현재 토큰 겹침 규칙이 표현 차이를 놓쳤다.",
            "다음 수정": "질문 재작성, 동의어 처리, 재정렬 기준 추가를 먼저 검토한다.",
            "안전 답변": "현재 검색 규칙으로는 관련 근거를 상위 후보로 안정적으로 끌어오지 못했다.",
        }

    if "모든" in question or "항상" in question:
        return {
            "판정 상태": "답변 과장 위험",
            "실패 단계": "답변 단계",
            "멈춘 이유": "상위 후보는 있지만 질문의 강한 단정까지 지지하지 못한다.",
            "다음 수정": "답변을 더 보수적으로 쓰고, 단정 표현을 줄인 질문으로 다시 평가한다.",
            "안전 답변": "검색된 문서만으로는 이 주장이 항상 성립한다고 단정할 수 없다.",
        }

    evidence = [item["문장"] for item in ranked if item["점수"] >= 2][:2]
    return {
        "판정 상태": "근거 기반 답변",
        "실패 단계": "통과",
        "멈춘 이유": "상위 후보와 선택 근거만으로 질문에 직접 답할 수 있다.",
        "다음 수정": "현재 기록을 기준선으로 저장하고 다음 실험과 비교한다.",
        "안전 답변": " ".join(evidence),
    }

audit_rows = []
for case in case_rows:
    ranked = rank_documents(case["question"])
    decision = decide_state(case, ranked)
    audit_rows.append({
        "연습 ID": case["case_id"],
        "질문": case["question"],
        "기대 상태": case["expected_state"],
        "판정 상태": decision["판정 상태"],
        "실패 단계": decision["실패 단계"],
        "상위 후보": [f"{item['문서']}({item['점수']})" for item in ranked[:3]],
        "선택 근거": [item["문서"] for item in ranked if item["점수"] >= 2][:2],
        "멈춘 이유": decision["멈춘 이유"],
        "다음 수정": decision["다음 수정"],
        "안전 답변": decision["안전 답변"],
    })

summary = {
    "질문 수": len(audit_rows),
    "근거 기반 답변 수": sum(row["판정 상태"] == "근거 기반 답변" for row in audit_rows),
    "문서 범위 밖 수": sum(row["판정 상태"] == "문서 범위 밖" for row in audit_rows),
    "검색 실패 수": sum(row["판정 상태"] == "검색 실패" for row in audit_rows),
    "답변 과장 위험 수": sum(row["판정 상태"] == "답변 과장 위험" for row in audit_rows),
    "다시 볼 질문": [
        row["질문"] for row in audit_rows if row["판정 상태"] != "근거 기반 답변"
    ],
}

print("요약 =", summary)
print("읽은 문서 파일 =", str(document_path))
print("읽은 경계 사례 파일 =", str(case_path))
핵심_case_ids = {"연습-01", "연습-02", "연습-03", "연습-07"}
print("핵심 질문별 기록 =")
for row in audit_rows:
    if row["연습 ID"] in 핵심_case_ids:
        print(row)

요약 = {'질문 수': 7, '근거 기반 답변 수': 3, '문서 범위 밖 수': 1, '검색 실패 수': 1, '답변 과장 위험 수': 2, '다시 볼 질문': ['MCP는 왜 필요한가?', '문서 분할 재정렬만 있으면 모든 환각이 해결되는가?', 'RAG가 있으면 최신 규칙 문서가 없어도 항상 정확한가?', '회수 문헌과 채택 이유를 선행 보존해야 후속 디버깅이 쉬운 까닭은 무엇인가?']}
읽은 문서 파일 = docs/assets/part-07/chapter-05/p7-5-rag-documents.csv
읽은 경계 사례 파일 = docs/assets/part-07/chapter-05/p7-5-boundary-cases.csv
핵심 질문별 기록 =
{'연습 ID': '연습-01', '질문': 'RAG 프로젝트에서 왜 검색 후보와 선택 근거를 답변보다 먼저 기록해야 하는가?', '기대 상태': '근거 기반 답변', '판정 상태': '근거 기반 답변', '실패 단계': '통과', '상위 후보': ['문서-2(3)', '문서-3(3)', '문서-4(3)'], '선택 근거': ['문서-2', '문서-3'], '멈춘 이유': '상위 후보와 선택 근거만으로 질문에 직접 답할 수 있다.', '다음 수정': '현재 기록을 기준선으로 저장하고 다음 실험과 비교한다.', '안전 답변': '프롬프트만으로는 최신 문서 근거를 보장할 수 없으므로 최신 규칙이 필요한 서비스에서는 검색 단계가 먼저 필요하다. 검색 후보 점수가 높아도 질문에 직접 답하지 않는 문장이 섞일 수 있으므로 선택 근거를 따로 남겨야 한다.'}
{'연습 ID': '연습-02', '질문': 'MCP는 왜 필요한가?', '기대 상태': '문서 범위 밖', '판정 상태': '문서 범위 밖', '실패 단계': '검색 단계', '상위 후보': ['문서-1(0)', '문서-2(0)', '문서-3(0)'], '선택 근거': [], '멈춘 이유': '현재 문서 집합에 질문의 핵심 주제가 없다.', '다음 수정': '문서 집합을 넓히거나 질문 범위를 현재 문서 범위 안으로 줄인다.', '안전 답변': '현재 검색된 문서에서는 이 질문에 답할 근거를 찾지 못했다.'}
{'연습 ID': '연습-03', '질문': '문서 분할 재정렬만 있으면 모든 환각이 해결되는가?', '기대 상태': '답변 과장 위험', '판정 상태': '답변 과장 위험', '실패 단계': '답변 단계', '상위 후보': ['문서-5(2)', '문서-2(1)', '문서-10(1)'], '선택 근거': ['문서-5'], '멈춘 이유': '상위 후보는 있지만 질문의 강한 단정까지 지지하지 못한다.', '다음 수정': '답변을 더 보수적으로 쓰고, 단정 표현을 줄인 질문으로 다시 평가한다.', '안전 답변': '검색된 문서만으로는 이 주장이 항상 성립한다고 단정할 수 없다.'}
{'연습 ID': '연습-07', '질문': '회수 문헌과 채택 이유를 선행 보존해야 후속 디버깅이 쉬운 까닭은 무엇인가?', '기대 상태': '검색 실패', '판정 상태': '검색 실패', '실패 단계': '검색 단계', '상위 후보': ['문서-1(0)', '문서-2(0)', '문서-3(0)'], '선택 근거': [], '멈춘 이유': '문서 집합에는 관련 내용이 있지만 현재 토큰 겹침 규칙이 표현 차이를 놓쳤다.', '다음 수정': '질문 재작성, 동의어 처리, 재정렬 기준 추가를 먼저 검토한다.', '안전 답변': '현재 검색 규칙으로는 관련 근거를 상위 후보로 안정적으로 끌어오지 못했다.'}

연습 ID:
질문:
판정 상태:
실패 단계:
상위 후보:
선택 근거:
멈춘 이유:
다음 수정 시작점: