0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

PythonでRAGの回答が正解でも検索失敗を見逃す理由

0
Posted at

PythonでRAGの回答が正解でも検索失敗を見逃す理由

RAGの回答が期待どおりでも、検索が成功したとは限りません。
LLMが事前知識から正解を出した場合、根拠文書を取得できていなくても回答一致率は上がります。
回答の正誤と検索の成否を同じ指標で測ると、修正すべき箇所を間違えます。
PythonでRecall@kと回答一致率を分離し、どちらの段階で失敗したかを記録する方法を考えます。

正解したのに検索が失敗している

たとえば、社内規程を検索するRAGを想像してください。

質問は「経費精算の申請期限はいつですか?」です。

評価用データには次の情報があります。

  • 期待回答:翌月5日
  • 正しい根拠文書ID:expense-policy
  • 検索結果:general-guide、travel-guide
  • 生成された回答:翌月5日

回答文字列だけを比較すると正解です。

しかし、検索結果には根拠となる expense-policy が含まれていません。

LLMが一般的な知識や別の文脈から回答した可能性があります。たまたま期待回答と一致したとしても、検索が正常に動作した証拠にはなりません。

この状態で検索モデルを変更すると、回答一致率だけでは検索品質の劣化を見落とします。

逆のケースもあります。

expense-policy が正しく取得されているのに、生成された回答が「翌月10日」になる場合です。

こちらは検索結果に必要な文書が存在します。少なくとも検索段階で根拠文書を取得するという条件は満たしています。

問題は、取得した情報を使って回答を作る段階にあります。

この二つを区別しない評価では、検索設定を調整すべきなのか、プロンプトや生成モデルを調整すべきなのか判断できません。

検索と回答を別々の評価対象にする

PythonでRAGの回答が正解でも検索失敗を見逃す理由

RAGを評価するとき、まず検索段階と生成段階を分けます。

検索段階では、期待する根拠文書が上位k件に含まれているかを確認します。

ここで使うのがRecall@kです。

質問ごとに正解となる根拠文書IDの集合を用意し、検索結果の上位k件に含まれた割合を計算します。

Recall@k =
  上位k件に含まれる正解文書ID数
  / 正解文書ID総数

たとえば正解文書が3件あり、上位5件の検索結果に2件含まれていれば、Recall@5は約0.667です。

正解文書が1件だけの場合は、取得できれば1.0、取得できなければ0.0になります。

ここでのRecall@kは、検索対象の文書集合に対して付与した正解IDを基準とする指標です。

文書をチャンクに分割している場合、チャンクIDと文書IDのどちらを正解単位にするか決めておく必要があります。

同じ文書から複数チャンクを取得しても、文書単位のRecall@kでは正解文書を重複して数えません。

一方、生成段階では期待回答と生成回答を比較します。

今回は動作を確認しやすくするため、正規化後の文字列完全一致を使います。

answer_match =
  normalize(actual) == normalize(expected)

この方法は、同じ意味でも表現が異なる回答を不正解にします。

たとえば「翌月5日」と「翌月の5日」は異なる文字列です。

したがって、これは意味的な回答品質を測る指標ではありません。評価処理の分離を確認するための簡易判定です。

実際の評価では、回答に必要な情報が含まれるか、取得文書に回答が裏付けられているかも確認する必要があります。

また、検索結果に正解文書が存在することと、LLMがその文書を根拠に回答したことも別です。

Recall@kは前者を測ります。後者まで保証するものではありません。

Pythonで失敗箇所を分類する

Python 3.14の標準ライブラリだけで、評価結果をJSONに出力する最小構成を作ります。

外部パッケージは不要です。

今回は検索器とLLMを直接呼び出さず、両者の実行結果を評価データとして渡します。

検索器や生成モデルを変更しても、評価処理をそのまま再利用するためです。

まず、質問、期待回答、根拠文書ID、実際の検索結果と回答を保持します。

rag_eval.py を作成します。

import json
from pathlib import Path

CASES = [
    {
        "question": "経費精算の期限は?",
        "expected": "翌月5日",
        "gold_ids": ["expense-policy"],
        "retrieved": ["general-guide", "travel-guide"],
        "actual": "翌月5日",
    },
    {
        "question": "交通費の申請方法は?",
        "expected": "申請フォームを使う",
        "gold_ids": ["travel-policy"],
        "retrieved": ["travel-policy", "general-guide"],
        "actual": "メールで申請する",
    },
    {
        "question": "備品購入の承認者は?",
        "expected": "部門責任者",
        "gold_ids": ["purchase-policy"],
        "retrieved": ["purchase-policy", "general-guide"],
        "actual": "部門責任者",
    },
]


def recall_at_k(gold_ids, retrieved, k):
    gold = set(gold_ids)
    if not gold:
        raise ValueError("gold_ids must not be empty")
    return len(gold & set(retrieved[:k])) / len(gold)


def normalize(text):
    return " ".join(text.split())


def evaluate(case, k=2):
    recall = recall_at_k(
        case["gold_ids"], case["retrieved"], k
    )
    matched = (
        normalize(case["actual"])
        == normalize(case["expected"])
    )

    if recall < 1.0:
        status = "retrieval_failure"
    elif not matched:
        status = "generation_failure"
    else:
        status = "passed"

    return {
        "question": case["question"],
        "recall_at_k": recall,
        "answer_match": matched,
        "status": status,
    }


if __name__ == "__main__":
    results = [evaluate(c) for c in CASES]
    report = {
        "k": 2,
        "cases": results,
        "mean_recall": sum(
            r["recall_at_k"] for r in results
        ) / len(results),
        "answer_accuracy": sum(
            r["answer_match"] for r in results
        ) / len(results),
    }
    Path("report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    print(json.dumps(report, ensure_ascii=False, indent=2))

実行します。

python rag_eval.py

生成される report.json には、次のような評価結果が含まれます。

{
  "k": 2,
  "cases": [
    {
      "question": "経費精算の期限は?",
      "recall_at_k": 0.0,
      "answer_match": true,
      "status": "retrieval_failure"
    },
    {
      "question": "交通費の申請方法は?",
      "recall_at_k": 1.0,
      "answer_match": false,
      "status": "generation_failure"
    },
    {
      "question": "備品購入の承認者は?",
      "recall_at_k": 1.0,
      "answer_match": true,
      "status": "passed"
    }
  ],
  "mean_recall": 0.6666666666666666,
  "answer_accuracy": 0.6666666666666666
}

この例では、検索の平均Recall@2と回答一致率が同じ値になっています。

しかし、失敗している質問は異なります。

1件目は検索失敗、2件目は生成失敗です。

集計値が同じでも、修正すべきコンポーネントは同じではありません。

なお、この実装では正解文書をすべて取得できなかった場合、回答が正しくても retrieval_failure を優先します。

そのため status は、検索と生成の失敗を完全に表現するものではありません。

検索が不完全で回答も間違っている場合、answer_match は false として残りますが、status は retrieval_failure です。

運用時には二つの判定値を保持し、必要なら両方失敗したケースを別分類にします。

評価データを固定して変更前後を比較する

この評価方法で気をつけたいのは、比較対象を固定することです。

検索モデルを変更した前後で質問や正解文書IDまで変わってしまうと、Recall@kの差がモデル変更によるものなのか判断できません。

最低限、次の条件をそろえます。

  • 同じ質問セットと正解文書IDを使う
  • 同じ検索対象文書とチャンク分割条件を使う
  • 同じkでRecall@kを計算する
  • 生成モデルの比較では検索結果も固定する

検索モデルの変更を評価するときは、取得結果を差し替えてRecall@kを比較します。

生成モデルの変更を評価するときは、取得文書を固定して回答品質を比較します。

両方を同時に変更すると、品質差の原因を切り分けにくくなります。

また、今回のコードでは正解文書IDが複数ある場合、すべて取得できなければ検索失敗と分類します。

これは厳しめの条件です。

実際のRAGでは、複数の根拠文書のうち1件だけで十分な質問もあります。その場合は、Recall@kの数値と検索成功条件を分けて設計した方が適切です。

たとえば、正解文書のいずれかを取得すれば成功とするHit@kや、回答に必要な根拠の組み合わせを判定する方法が考えられます。

評価用データも3件では足りません。

実際に使うなら、10件程度から始め、検索しにくい質問、表記揺れ、複数文書を参照する質問などを追加します。

Pythonの unittest を使えば、既知の評価ケースに対する判定結果を回帰テストとして固定できます。

たとえば、最初のケースが回答一致にもかかわらず検索失敗と判定されることを確認します。

import unittest

from rag_eval import CASES, evaluate


class TestRagEvaluation(unittest.TestCase):
    def test_correct_answer_with_failed_retrieval(self):
        result = evaluate(CASES[0])

        self.assertTrue(result["answer_match"])
        self.assertEqual(result["recall_at_k"], 0.0)
        self.assertEqual(
            result["status"], "retrieval_failure"
        )

    def test_generation_failure(self):
        result = evaluate(CASES[1])

        self.assertEqual(result["recall_at_k"], 1.0)
        self.assertFalse(result["answer_match"])
        self.assertEqual(
            result["status"], "generation_failure"
        )


if __name__ == "__main__":
    unittest.main()

このコードを test_rag_eval.py として保存し、次のコマンドで実行します。

python -m unittest -v

評価コードを変更したとき、検索失敗と生成失敗の分類が意図せず変わっていないかを確認できます。

ただし、これは評価器自体のテストです。RAGシステムの検索品質や回答品質が保証されるわけではありません。

本番のRAGでは、取得文書に正解IDが含まれていても、回答に必要な箇所がチャンクから欠落していることがあります。また、文字列一致では正しい言い換えを評価できません。次に確認したいのは、検索結果に必要な根拠が実際に含まれているかという判定と、その根拠に回答が支持されているかという判定です。この二つを追加しても、検索と生成を独立して記録する設計は維持できます。

参考

0
0
1

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?