P6-11.2 검색 실패와 생성 실패를 가르는 RAG 흐름¶

# 검색 결과와 생성 답변을 따로 기록해 RAG 실패 위치를 나누어 보는 예제입니다.
import csv
from pathlib import Path

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

question = "벡터 검색이 왜 필요한가요?"

document_path = Path("docs/assets/part-06/chapter-11/p6-11-rag-documents.csv")
experiment_path = Path("docs/assets/part-06/chapter-11/p6-11-rag-experiments.csv")

def read_csv(path):
    with path.open(encoding="utf-8", newline="") as file:
        return list(csv.DictReader(file))

documents = read_csv(document_path)
experiments = read_csv(experiment_path)

# 문서 제목과 본문을 함께 벡터화해 검색 후보 공간을 만든다.
document_texts = [
    f"{doc['title']} {doc['text']}"
    for doc in documents
]
vectorizer = TfidfVectorizer(analyzer="char_wb", ngram_range=(2, 4))
document_vectors = vectorizer.fit_transform(document_texts)

def build_query(experiment):
    terms = experiment["retrieval_terms"].split(";")
    return f"{question} {' '.join(terms)}"

def retrieve_documents(query, top_k=2):
    query_vector = vectorizer.transform([query])
    scores = cosine_similarity(query_vector, document_vectors).ravel()
    ranked_indexes = scores.argsort()[::-1]

    retrieved = []
    for index in ranked_indexes:
        if scores[index] <= 0:
            continue
        retrieved.append(
            {
                **documents[index],
                "similarity": round(float(scores[index]), 3),
            }
        )
        if len(retrieved) == top_k:
            break
    return retrieved

def generate_answer(retrieved_docs, generation_style):
    first = retrieved_docs[0]["text"] if retrieved_docs else "참고 문서가 없다."
    second = retrieved_docs[1]["text"] if len(retrieved_docs) > 1 else "추가 근거가 부족하다."

    if generation_style == "overclaim":
        return (
            f"{first} "
            "그래서 항상 최신 정보와 정답을 자동으로 보장한다."
        )

    return (
        f"{first} "
        f"그래서 {second}"
    )

def inspect_result(retrieved_docs, answer):
    source_trace = [
        {
            "doc_id": doc["doc_id"],
            "title": doc["title"],
            "source_role": doc["source_role"],
            "similarity": doc["similarity"],
            "text_preview": doc["text"][:34],
        }
        for doc in retrieved_docs
    ]
    contains_irrelevant_doc = any(
        doc["category"] == "irrelevant" for doc in retrieved_docs
    )
    irrelevant_fragments = [
        doc["text"].split(".")[0]
        for doc in retrieved_docs
        if doc["category"] == "irrelevant"
    ]
    answer_mentions_irrelevant_content = any(
        fragment and fragment in answer
        for fragment in irrelevant_fragments
    )
    answer_overclaims = "항상 최신 정보와 정답을 자동으로 보장" in answer

    return {
        "source_trace": source_trace,
        "doc_titles": [doc["title"] for doc in retrieved_docs],
        "doc_similarities": [doc["similarity"] for doc in retrieved_docs],
        "top_doc_category": retrieved_docs[0]["category"] if retrieved_docs else "none",
        "contains_irrelevant_doc": contains_irrelevant_doc,
        "answer_mentions_irrelevant_content": answer_mentions_irrelevant_content,
        "answer_overclaims": answer_overclaims,
        "retrieval_failed": contains_irrelevant_doc,
        "generation_failed": (not contains_irrelevant_doc) and answer_overclaims,
    }

reports = []
for experiment in experiments:
    query = build_query(experiment)
    retrieved_docs = retrieve_documents(query)
    answer = generate_answer(retrieved_docs, experiment["generation_style"])
    inspect = inspect_result(retrieved_docs, answer)
    reports.append(
        {
            "experiment": {
                "name": experiment["name"],
                "query": query,
                "generation_style": experiment["generation_style"],
            },
            "answer": answer,
            "inspect": inspect,
        }
    )

summary = {
    "retrieval_failure_count": sum(report["inspect"]["retrieval_failed"] for report in reports),
    "generation_failure_count": sum(report["inspect"]["generation_failed"] for report in reports),
    "irrelevant_leak_count": sum(report["inspect"]["answer_mentions_irrelevant_content"] for report in reports),
    "overclaim_count": sum(report["inspect"]["answer_overclaims"] for report in reports),
    "retrieval_failure_ratio": round(
        sum(report["inspect"]["retrieval_failed"] for report in reports) / len(reports),
        2,
    ),
    "generation_failure_ratio": round(
        sum(report["inspect"]["generation_failed"] for report in reports) / len(reports),
        2,
    ),
}

print("[summary]")
print(summary)
print()

selected_names = {
    "clean_grounded_vector_search",
    "noisy_retrieval_marketing_copy",
    "clean_but_overclaim_vector_search",
}
selected_reports = [
    report for report in reports
    if report["experiment"]["name"] in selected_names
]

for report in selected_reports:
    print("=" * 80)
    print("[experiment]")
    print(report["experiment"])
    print("[generated answer]")
    print(report["answer"])
    print("[inspect]")
    print(report["inspect"])

[summary]
{'retrieval_failure_count': 12, 'generation_failure_count': 12, 'irrelevant_leak_count': 12, 'overclaim_count': 12, 'retrieval_failure_ratio': 0.33, 'generation_failure_ratio': 0.33}

================================================================================
[experiment]
{'name': 'clean_grounded_vector_search', 'query': '벡터 검색이 왜 필요한가요? 의미 벡터 검색', 'generation_style': 'grounded'}
[generated answer]
벡터 검색은 의미가 비슷한 텍스트를 벡터 공간에서 가깝게 찾는 검색 방식이다. 키워드가 달라도 의미 기반 검색이 가능하다. 그래서 키워드 검색은 같은 단어가 있는지 먼저 보지만, 의미 검색은 질문과 문서의 의미가 가까운지 비교한다. 그래서 표현이 달라도 관련 문서를 찾을 수 있다.
[inspect]
{'source_trace': [{'doc_id': 'R01', 'title': '벡터 검색 기본 설명', 'source_role': 'primary_evidence', 'similarity': 0.555, 'text_preview': '벡터 검색은 의미가 비슷한 텍스트를 벡터 공간에서 가깝게 찾는'}, {'doc_id': 'R02', 'title': '키워드 검색과 의미 검색 차이', 'source_role': 'primary_evidence', 'similarity': 0.189, 'text_preview': '키워드 검색은 같은 단어가 있는지 먼저 보지만, 의미 검색은 '}], 'doc_titles': ['벡터 검색 기본 설명', '키워드 검색과 의미 검색 차이'], 'doc_similarities': [0.555, 0.189], 'top_doc_category': 'retrieval', 'contains_irrelevant_doc': False, 'answer_mentions_irrelevant_content': False, 'answer_overclaims': False, 'retrieval_failed': False, 'generation_failed': False}
================================================================================
[experiment]
{'name': 'noisy_retrieval_marketing_copy', 'query': '벡터 검색이 왜 필요한가요? 마케팅 문구 홍보', 'generation_style': 'grounded'}
[generated answer]
마케팅 캠페인용 문구와 홍보 배너 문장을 다양하게 바꾸는 방법을 설명한다. 벡터 검색 근거가 아니다. 그래서 무관한 마케팅 문구를 더 다양하게 조합해 홍보 문안을 만드는 설명이다. 검색 품질 판단과 직접 관련이 없다.
[inspect]
{'source_trace': [{'doc_id': 'X02', 'title': '홍보 배너 문장 후보', 'source_role': 'off_topic_noise', 'similarity': 0.334, 'text_preview': '마케팅 캠페인용 문구와 홍보 배너 문장을 다양하게 바꾸는 방법'}, {'doc_id': 'X01', 'title': '마케팅 문구 A/B 테스트', 'source_role': 'off_topic_noise', 'similarity': 0.321, 'text_preview': '무관한 마케팅 문구를 더 다양하게 조합해 홍보 문안을 만드는 '}], 'doc_titles': ['홍보 배너 문장 후보', '마케팅 문구 A/B 테스트'], 'doc_similarities': [0.334, 0.321], 'top_doc_category': 'irrelevant', 'contains_irrelevant_doc': True, 'answer_mentions_irrelevant_content': True, 'answer_overclaims': False, 'retrieval_failed': True, 'generation_failed': False}
================================================================================
[experiment]
{'name': 'clean_but_overclaim_vector_search', 'query': '벡터 검색이 왜 필요한가요? 의미 벡터 검색', 'generation_style': 'overclaim'}
[generated answer]
벡터 검색은 의미가 비슷한 텍스트를 벡터 공간에서 가깝게 찾는 검색 방식이다. 키워드가 달라도 의미 기반 검색이 가능하다. 그래서 항상 최신 정보와 정답을 자동으로 보장한다.
[inspect]
{'source_trace': [{'doc_id': 'R01', 'title': '벡터 검색 기본 설명', 'source_role': 'primary_evidence', 'similarity': 0.555, 'text_preview': '벡터 검색은 의미가 비슷한 텍스트를 벡터 공간에서 가깝게 찾는'}, {'doc_id': 'R02', 'title': '키워드 검색과 의미 검색 차이', 'source_role': 'primary_evidence', 'similarity': 0.189, 'text_preview': '키워드 검색은 같은 단어가 있는지 먼저 보지만, 의미 검색은 '}], 'doc_titles': ['벡터 검색 기본 설명', '키워드 검색과 의미 검색 차이'], 'doc_similarities': [0.555, 0.189], 'top_doc_category': 'retrieval', 'contains_irrelevant_doc': False, 'answer_mentions_irrelevant_content': False, 'answer_overclaims': True, 'retrieval_failed': False, 'generation_failed': True}