RAG品質評価と失敗パターン分析：検索拡張生成の診断と改善

# Dense Retrievalの例
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(documents, embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# クエリに対する関連ドキュメントの検索
results = retriever.get_relevant_documents("RAGの評価方法は？")

# Cohere Rerankerの例
from cohere import Client

co = Client(api_key="...")
reranked = co.rerank(
    model="rerank-v3.5",
    query="RAG評価方法",
    documents=retrieved_docs,
    top_n=3
)

# コンテキストベースの回答生成
prompt = f"""以下のコンテキストに基づいて質問に回答してください。

コンテキスト：
{context}

質問：{query}

回答："""

response = llm.generate(prompt)

# Recall@K 計算例
def recall_at_k(retrieved_ids, relevant_ids, k):
    retrieved_set = set(retrieved_ids[:k])
    relevant_set = set(relevant_ids)
    return len(retrieved_set & relevant_set) / len(relevant_set)

# MRR 計算例
def mrr(retrieved_ids, relevant_ids):
    for i, doc_id in enumerate(retrieved_ids):
        if doc_id in relevant_ids:
            return 1.0 / (i + 1)
    return 0.0

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

# 評価データの準備
eval_data = {
    "question": ["RAGとは何ですか？"],
    "answer": ["RAGは検索拡張生成で..."],
    "contexts": [["RAG（Retrieval-Augmented Generation）は..."]],
    "ground_truth": ["RAGは外部知識を検索して..."]
}

dataset = Dataset.from_dict(eval_data)
results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall]
)
print(results)
# {'faithfulness': 0.92, 'answer_relevancy': 0.87, ...}

from deepeval import evaluate
from deepeval.metrics import (
    FaithfulnessMetric,
    AnswerRelevancyMetric,
    ContextualRelevancyMetric,
    HallucinationMetric,
)
from deepeval.test_case import LLMTestCase

test_case = LLMTestCase(
    input="RAGの主要な評価指標は？",
    actual_output="RAGの主要な評価指標にはfaithfulness、relevancy...",
    expected_output="Faithfulness、Answer Relevancy、Context Precision...",
    retrieval_context=["RAG評価にはさまざまな指標が使用されます..."]
)

faithfulness_metric = FaithfulnessMetric(threshold=0.7)
relevancy_metric = AnswerRelevancyMetric(threshold=0.7)

evaluate([test_case], [faithfulness_metric, relevancy_metric])

from llama_index.core.evaluation import (
    FaithfulnessEvaluator,
    RelevancyEvaluator,
    CorrectnessEvaluator,
    BatchEvalRunner,
)
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-4o")
faithfulness_evaluator = FaithfulnessEvaluator(llm=llm)
relevancy_evaluator = RelevancyEvaluator(llm=llm)

# バッチ評価
runner = BatchEvalRunner(
    {"faithfulness": faithfulness_evaluator, "relevancy": relevancy_evaluator},
    workers=4,
)
eval_results = await runner.aevaluate_queries(query_engine, queries=queries)

JUDGE_PROMPT = """以下のRAGシステムの回答を評価してください。

[質問]: {question}
[コンテキスト]: {context}
[回答]: {answer}

評価基準：
1. 忠実度（1-5）：回答はコンテキストに基づいているか？
2. 完全性（1-5）：質問の全ての側面をカバーしているか？
3. 簡潔性（1-5）：不要な情報なく要点を伝えているか？

JSON形式でスコアと根拠を出力してください。"""

def evaluate_with_judge(question, context, answer, judge_llm):
    prompt = JUDGE_PROMPT.format(
        question=question, context=context, answer=answer
    )
    result = judge_llm.generate(prompt)
    return json.loads(result)

質問：「2024年第4四半期の売上はいくらですか？」

検索されたチャンク：「2023年第4四半期の売上は150億円を記録しました...」
→ 年度が異なるドキュメントが検索される（メタデータフィルタリング欠如）

期待されるチャンク：「2024年第4四半期の売上は200億円で前年比33%成長...」

# メタデータフィルタを活用した検索
results = vectorstore.similarity_search(
    query="第4四半期 売上",
    filter={"year": 2024, "quarter": "Q4"},
    k=5
)

# Semantic Chunkingの適用
from langchain.text_splitter import SemanticChunker

splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=90,
)
chunks = splitter.split_documents(documents)

# 適応的K選択戦略
def adaptive_retrieval(query, retriever, min_k=3, max_k=10, threshold=0.7):
    """類似度閾値に基づいて動的にKを調整"""
    results = retriever.similarity_search_with_score(query, k=max_k)

    filtered = [
        (doc, score) for doc, score in results
        if score >= threshold
    ]

    if len(filtered) < min_k:
        return [doc for doc, _ in results[:min_k]]

    return [doc for doc, _ in filtered]

# ハルシネーション防止のための強化プロンプト
ANTI_HALLUCINATION_PROMPT = """あなたは与えられたコンテキストのみに基づいて回答するアシスタントです。

ルール：
1. 必ずコンテキストにある情報のみを使用してください。
2. コンテキストにない情報は「提供されたドキュメントにその情報は見つかりません」と回答してください。
3. 推測したり事前知識を使用しないでください。
4. 回答の各文の末尾に[出典：ドキュメントN]を表示してください。

コンテキスト：
{context}

質問：{question}

回答："""

# Lost-in-the-Middle緩和戦略：重要なドキュメントを両端に配置
def reorder_for_lost_in_middle(documents, scores):
    """最も関連性の高いドキュメントを先頭と末尾に配置"""
    sorted_docs = sorted(
        zip(documents, scores), key=lambda x: x[1], reverse=True
    )

    reordered = []
    for i, (doc, score) in enumerate(sorted_docs):
        if i % 2 == 0:
            reordered.insert(0, doc)  # 先頭に挿入
        else:
            reordered.append(doc)      # 末尾に追加

    return reordered

# クエリにinstruction prefixを追加してミスマッチを緩和
# （Instructor系埋め込みモデルの活用）
from InstructorEmbedding import INSTRUCTOR

model = INSTRUCTOR("hkunlp/instructor-xl")

# クエリ用埋め込み
query_embedding = model.encode(
    [["Represent the question for retrieving supporting documents:", query]]
)

# ドキュメント用埋め込み
doc_embedding = model.encode(
    [["Represent the technical document for retrieval:", document]]
)

# ナレッジベースの鮮度管理システム
class KnowledgeBaseFreshnessManager:
    def __init__(self, vectorstore, max_age_days=30):
        self.vectorstore = vectorstore
        self.max_age_days = max_age_days

    def check_staleness(self):
        """古いドキュメントの検出"""
        cutoff = datetime.now() - timedelta(days=self.max_age_days)
        stale_docs = self.vectorstore.query(
            filter={"updated_at": {"$lt": cutoff.isoformat()}}
        )
        return stale_docs

    def incremental_update(self, new_documents):
        """増分更新：変更されたドキュメントのみ再埋め込み"""
        for doc in new_documents:
            existing = self.vectorstore.get(
                filter={"source_id": doc.metadata["source_id"]}
            )
            if existing and self._content_changed(existing, doc):
                self.vectorstore.delete(ids=[existing.id])
                self.vectorstore.add_documents([doc])
            elif not existing:
                self.vectorstore.add_documents([doc])

    def add_temporal_boost(self, results, recency_weight=0.1):
        """最新ドキュメントにボーナススコアを付与"""
        now = datetime.now()
        boosted = []
        for doc, score in results:
            age_days = (now - doc.metadata["updated_at"]).days
            recency_score = max(0, 1 - age_days / 365)
            final_score = score + recency_weight * recency_score
            boosted.append((doc, final_score))
        return sorted(boosted, key=lambda x: x[1], reverse=True)

def classify_failure(question, retrieved_docs, generated_answer, ground_truth):
    """RAG失敗を体系的に分類"""

    # ステップ1：検索品質の確認
    retrieval_recall = calculate_recall(retrieved_docs, ground_truth_docs)

    if retrieval_recall < 0.5:
        return "RETRIEVAL_FAILURE"

    # ステップ2：コンテキスト関連性の確認
    context_relevancy = evaluate_context_relevancy(question, retrieved_docs)

    if context_relevancy < 0.5:
        return "CONTEXT_NOISE"

    # ステップ3：回答の忠実度確認
    faithfulness = evaluate_faithfulness(generated_answer, retrieved_docs)

    if faithfulness < 0.7:
        return "HALLUCINATION"

    # ステップ4：回答の正確度確認
    correctness = evaluate_correctness(generated_answer, ground_truth)

    if correctness < 0.7:
        return "GENERATION_QUALITY"

    return "ACCEPTABLE"

# 検索段階のデバッグ
def debug_retrieval(query, vectorstore, k=10):
    results = vectorstore.similarity_search_with_score(query, k=k)

    print(f"Query: {query}")
    print(f"{'='*60}")
    for i, (doc, score) in enumerate(results):
        print(f"\n[{i+1}] Score: {score:.4f}")
        print(f"Source: {doc.metadata.get('source', 'unknown')}")
        print(f"Content: {doc.page_content[:200]}...")
        print(f"Metadata: {doc.metadata}")

    # クエリ埋め込み分析
    query_embedding = embeddings.embed_query(query)
    print(f"\nQuery embedding norm: {np.linalg.norm(query_embedding):.4f}")
    print(f"Query embedding dim: {len(query_embedding)}")

    return results

# RAG設定のA/Bテストフレームワーク
class RAGABTest:
    def __init__(self, test_queries, ground_truths):
        self.test_queries = test_queries
        self.ground_truths = ground_truths

    def run_experiment(self, config_a, config_b, metrics):
        results_a = self._evaluate_config(config_a, metrics)
        results_b = self._evaluate_config(config_b, metrics)

        comparison = {}
        for metric_name in metrics:
            score_a = np.mean(results_a[metric_name])
            score_b = np.mean(results_b[metric_name])
            improvement = (score_b - score_a) / score_a * 100

            comparison[metric_name] = {
                "config_a": score_a,
                "config_b": score_b,
                "improvement_pct": improvement,
            }

        return comparison

# 使用例
ab_test = RAGABTest(test_queries, ground_truths)
result = ab_test.run_experiment(
    config_a={"chunk_size": 512, "k": 5, "model": "gpt-4o-mini"},
    config_b={"chunk_size": 1024, "k": 3, "model": "gpt-4o"},
    metrics=["faithfulness", "answer_relevancy", "recall"]
)

ドキュメントタイプ別のチャンキング戦略：

1. 技術文書 / APIドキュメント
   → Markdownヘッダーベースの分割 + 小さいチャンク（256-512トークン）

2. 法律/規制文書
   → 条項単位の分割 + 階層的インデキシング

3. 会話ログ / FAQ
   → 質問-回答ペア単位の分割

4. 学術論文
   → セクションベースの分割 + アブストラクト/結論の個別インデキシング

5. 一般テキスト
   → Semantic Chunking（意味単位の分割）

# 本番環境モニタリングダッシュボードの指標
monitoring_metrics = {
    "retrieval": {
        "avg_similarity_score": 0.82,
        "hit_rate": 0.94,
        "avg_retrieved_docs": 4.2,
        "empty_retrieval_rate": 0.02,
    },
    "generation": {
        "avg_faithfulness": 0.89,
        "avg_answer_length": 245,
        "refusal_rate": 0.05,
        "avg_latency_ms": 1200,
    },
    "user_feedback": {
        "thumbs_up_rate": 0.78,
        "escalation_rate": 0.08,
    }
}