自己修復するRAG設計術：誤答を検知して自動で再検索する7選

モデルが自信満々に誤答を返し、チームが手動で再検索している――そんな不毛な光景にうんざりしていませんか？

RAGは外部知識で応答を強化しますが、堂々とした誤答や不十分な根拠が現場の信頼を損ないます。理想はユーザーの介入を待たず、失敗を自動で検知して再検索や検証フローに回す仕組みです。本記事では、誤答検出から自動リトライ、プロンプト改良やフェイルオーバーまで、実務で使える「自己修復するRAG」設計術を7つの視点で紹介します。各手法の実装上の注意点とトレードオフも交え、すぐに試せる実例を届けます。

そもそもどう動くのか

家電量販店の説明書みたいに「動いているふう」でも、根拠がなければ上司に突っ込まれて終わる――そんな痛みから逃げられない人へ。

鍵は「検知→判断→再取得→再評価」のループをプログラム化すること。仕組みが分かれば後は応用である。要点は、LLMの返答に対して必ず「メタ情報（検証結果・類似度・参照ID）」を添えてログ化し、そのメタをトリガーに自動でベクトルDB再検索やクエリ拡張、再ランクを走らせることだ。

1. メタ設計（何を必ず残すか）

• 保存例（JSON） { "response_id": "...", "timestamp": "...", "answer": "...", "sources": [...], "avg_similarity": 0.72, "verifier_score": 0.54, "retries": 0 }
• 実装: 返答生成API直後にこのJSONをDB/ログに書く。

1. 検知：トリガー条件を明確化する

• トリガー例: sources空欄／avg_similarityがしきい値未満／verifierが低い／矛盾検出
• 実装例（疑似コード）: if meta["sources"] == [] or meta["avg_similarity"] < SIMILARITY_TH: mark_for_refetch()

1. 判断ルール：優先度と閾値

• 閾値は運用でチューニング。まずは「根拠なし＝即再取得」「低類似度＝再取得＋拡張」を置く。
• 実装設定: MAX_RETRIES=3, BACKOFF=2s。

1. 再取得：ベクトルDBを叩く

• 具体例（HTTP）: curl -X POSThttps://vector-db/query-d '{"query":"ユーザ質問","top_k":10}'
• 手順: top_kを増やす／metadataフィルタを追加／近傍を全文取得してContextを作る。

1. クエリ拡張（自動リライト）

• 取得した上位文書のタイトル＋要約を元にクエリを拡張する。例: "元クエリ + 上位3件の要旨" をLLMに渡し再検索。
• 実装例（Python）: ctx = "\n".join([d['snippet'] for d in hits[:3]]) expanded = call_model(f"次の文脈を踏まえて検索クエリを改善してください:\n{ctx}\n元クエリ:{q}")

1. 再ランク：候補を精査して順序付け

• cross-encoderやLLMによるペアwiseスコアで再ランク。簡易実装: scores = [score_fn(q, doc['text']) for doc in candidates] candidates.sort(key=lambda d: scores[d_idx], reverse=True)

1. 再評価とループ制御

• 上位候補で回答を再生成し、再度メタを付与。retry回数を増やし、MAX_RETRIES到達で「人間レビュー」フラグを付ける。
• 実装サンプル（擬似Python）: retries = 0 while retries < MAX_RETRIES: hits = vector_db.query(q, top_k=10*(retries+1)) expanded_q = expand_query(q, hits) answer, meta = generate_and_verify(expanded_q) log(meta) if not needs_refetch(meta): break retries += 1

まとめ：レスポンスに必ずメタを付けてログ化し、そのメタを起点に再検索→拡張→再ランク→再生成する流れを作れば、誤答を自動的に潰せる率が上がる。まずは「メタ定義」と「再取得の1パス」を週内に実装して、ログを1日分ためて閾値をチューニングするのが現実的な次の一歩です。

①まずは失敗検知の基盤を作る（ログ＋メタ収集）

家で手を動かしている最中に「これ答え合ってる？」と確認できない不安、これがRAG運用の一番の痛みである。

そもそもどう動くのか：クエリが来たら（1）埋め込みで近傍検索→（2）取得したドキュメントIDとスコアをモデルへ渡す→（3）モデル応答を受け取る、の3点を必ず時系列で保存すれば、誤答の原因追跡と自動再検索が可能になる。

1. 永続化の土台を作る（pgvectorテーブル） 必須項目を含むテーブルを用意する。例:

CREATE TABLE rag_logs (
  id SERIAL PRIMARY KEY,
  query TEXT,
  response TEXT,
  retrieval_ids TEXT[],
  top_score REAL,
  created_at TIMESTAMP DEFAULT NOW()
);

1. ログを書き込む（SQL挿入） クエリ→取得→応答が揃ったら即座にINSERT。プレースホルダ使用例:

INSERT INTO rag_logs (query,response,retrieval_ids,top_score) VALUES ($1,$2,$3,$4);

1. 取得結果（ID＋スコア）を確実に取る ベクトル検索でIDと距離を返す。pgvectorなら:

SELECT id, embedding <-> query_embedding AS score
FROM documents
ORDER BY score
LIMIT 5;

この結果をretrieval_idsとtop_scoreへ流す。

1. ミドルウェアで一貫ログを取る（即時性） APIハンドラで「取得直後」「応答直後」に書き込む。Python例（psycopg2概略）:

cur.execute(insert_sql, (query, response, ids_array, top_score))
conn.commit()

1. モデル応答とメタをセットで保存 応答だけでなく、モデル名・温度などのメタをresponseにJSON文字列で付与して保存する。例:response = json.dumps({"text": out, "meta": {"model":"x","temp":0.0}})

2. 低頻度データはS3へ退避 バッチでログをまとめてS3へ送る:

aws s3 cp rag_logs.json s3://my-bucket/logs/$(date +%s).json

1. 検索・集計が速くなるインデックスと運用 時系列検索用にインデックス作成:

CREATE INDEX idx_rag_logs_created_at ON rag_logs(created_at);

週次でtop_scoreの分布を集計し、閾値を決める。

まとめ まずは上のテーブルを作って24時間分のログを流してみてください。次はそのログから「top_scoreが閾値以下のケース」を抽出して再検索ループをテストしましょう。

②軽量な検証器(Verifier)を置く（モデル+ルール両面）

家の鍵代わりに「正しい答えか？」を毎回人がチェックしている時間、もう終わりにしませんか。

そもそもどう動くのか：一次応答を受け取ったら、軽量な検証器が短いルールチェック（正規表現や必須フィールドの有無）と簡潔なモデル検証（短いプロンプトで妥当性を判定）を並列で実行し、OK/NOT_OK を返す。判定結果に応じてDBにフラグ立て、NOT_OKなら自動で再検索を起動する——仕組みは単純で、あとは閾値やルールを増やすだけで現場適用できる。

1. 入力抽出（高速）

• コマンド例：jqでレスポンスを取り出す jq -r '.answer' resp.json > answer.txt
• 必須フィールドがあるかはgrepで即チェック： grep -E '^[A-Z].+' answer.txt || echo "MISSING_HEADER"

1. ルールベースの短判定

• 正規表現で事実性をチェック： python -c "import re,sys;print('NOT_OK' if re.search(r'(参考文献|出典なし)',open('answer.txt').read()) else 'OK')"

1. モデル検証の短プロンプト

• 実行例（検証モデルを叩く）： python verifier.py --query "${QUERY}" --answer-file resp.json
• verifier.py は「要点3点に事実が含まれるか」「矛盾がないか」を短いプロンプトで聞いて OK/NOT_OK を返す設計にする。

1. 非同期・軽量化

• バックグラウンド実行：nohup python verifier.py --query "${QUERY}" --answer-file resp.json &
• レイテンシが問題ならモデル検証のタイムアウトを 2s に設定する等の仕掛けを入れる。

1. スコアリングと閾値運用

• スコアを数値化して比較：python -c "print(0.78>=0.7)" で閾値判定を自動化。
• スコア→OK基準を設定して、false positive を減らす。

1. DB反映（必須）

• 判定結果をログに書く。例： UPDATE rag_logs SET verified=$1 WHERE id=$2;
• CLI例（psql/sqlite いずれかに合わせて）： psql -c "UPDATE rag_logs SET verified='NOT_OK' WHERE id=123;"

1. NOT_OK時の自動再検索トリガー

• トリガー用スクリプト例： if [ "$VERDICT" = "NOT_OK" ]; then python retriever.py --query "${QUERY}" --mode rerank --out new_resp.json; fi
• retriever.py は再取得→要約→再検証ループを起動すること。

まとめ：まずは既存のレスポンスパイプラインに verifier.py を差し込み、1件のログで OK/NOT_OK が書き換わるところまで動かしてみてください。次の一歩は、あなたの環境で verifier.py を 1回だけ実行して結果を rag_logs に反映させることです。

③再検索をトリガーする条件を明示化する（閾値設計）

家電量販店の返品対応みたいに「正解が合っているか？」を人間が全部確認していると、運用が死ぬ。

そもそもどう動くのか：検証器(verifier)や類似度スコアは信頼度の代理変数である。これらを定量的に閾値化して「ここがダメなら自動で再検索する」というルールに落とし込めば、誤答を現場から自動的に潰せる。以下は実務で使える7つの閾値設計（具体コマンド/設定つき）。

1. 検証器がNO/NOT_OKを返したら即再検索

   • 条件: log.verified == 'NO' or verifier == 'NOT_OK'
   • 実装例:

   if log.verified == 'NO':
       subprocess.run(['python','requery.py','--query',query,'--k','200'])
   

2. トップ類似度が低い（確信度不足）

   • 条件: top_score < 0.65（例）
   • 実装例:

   if log.top_score < 0.65:
       trigger_requery(query, widen=True)
   

3. トップと2位の差が小さい（曖昧）

   • 条件: top_score - second_score < 0.05
   • 実装例:

   if (log.top_score - log.second_score) < 0.05:
       subprocess.run(['python','requery.py','--query',query,'--k','300'])
   

4. 上位Kの平均スコアが低い（全体的に弱い）

   • 条件: avg(top_k_scores) < 0.6
   • 実装例（k=10）:

   if sum(log.top_k_scores)/10 < 0.6:
       subprocess.run(['python','requery.py','--query',query,'--k','400'])
   

5. 取得ドキュメントの多様性が低い（同一ソース偏り）

   • 条件: unique_doc_count / k < 0.5
   • 実装例:

   uniqueness = len(set(doc.id for doc in log.top_docs)) / len(log.top_docs)
   if uniqueness < 0.5:
       subprocess.run(['python','requery.py','--query',query,'--k','500','--widen'])
   

6. 外部ファクトチェックやルール違反を検出したら

   • 条件: fact_checker(query, answer) == 'CONFLICT'
   • 実装例:

   if fact_check_result == 'CONFLICT':
       subprocess.run(['python','fact_check.py','--query',query,'--action','requery'])
   

7. 再検索しても改善しない“繰り返し失敗”の扱い（エスカレーション）

   • 条件: consecutive_requeries >= 2 and top_score < 0.7
   • 実装例（エスカレーション時にwidenフラグ）:

   if log.requery_count >= 2 and log.top_score < 0.7:
       subprocess.run(['python','requery.py','--query',query,'--k','200','--widen'])
   

実装時の運用メモ：閾値は最初から本番全流量には当てないこと。まずはログ出力を有効化して1%〜5%でA/B観察し、誤検知・過検知のトレードオフを定量的に調整する。

まとめ：検証器結果・類似度・多様性・履歴の4軸で閾値を作れば、再検索の発火ルールは運用で十分役に立つ。まずやること：上記ルールのうち1つ（例：top_score < 0.65）を本番の1%トラフィックで有効にしてログを30日収集してみてください。

④再検索は幅を広げて複数戦略で（k増加・クエリ拡張）

家の資料を見つけられずに「AIが間違ってる…」で止まること、もう嫌じゃないですか。

そもそもどう動くのか：最初のレトリーブで候補が足りない／類似度が揺れると、生成モデルは誤答を引き起こす。対策は単純で「再検索の幅を広げて別角度のクエリを投げる」こと。具体的には（1）kを増やす、（2）クエリをパラフレーズやキーワードに分解して複数検索、（3）意味検索とキーワード検索を組み合わせる。これだけで誤答検知→再検索→再生成の成功率が上がる。

1. kを段階的に増やす

• 初回はk=10、信頼度が低ければk=50→k=200と段階的に広げる。
• 実際のpgvector再検索例：

SELECT id, content, embedding <-> query_embedding AS distance
FROM documents
ORDER BY distance
LIMIT 200;

• 実装: 最初はLIMIT 10、閾値を超えたらLIMIT 200で再実行。

1. クエリをキーワード抽出で拡張する

• 元クエリを名詞・固有表現で抽出しOR検索を追加する。
• コマンド例:

python expand_query.py --query "${QUERY}" --strategy keyword && \
python retrieve.py --k 200 --query "${EXPANDED}"

1. クエリをパラフレーズで複数投げる

• パラフレーズを3つ作り、それぞれで上位を取得してマージする。
• 実装例: paraphrase.pyで3文生成→各々でretrieve.pyを実行→結果をスコアで統合。

1. 意味検索 + BM25ハイブリッド

• まず埋め込みで上位500件を取る→その集合でBM25スコアを計算して再ランキングする。
• 実行手順: semantic_retrieve(k=500) → elasticsearch/_scoreで再ランキング。

1. チャンク粒度を変えて再検索

• ドキュメントを短めチャンク（256トークン）でもう一度検索。長文だと埋め込みが希釈される。
• 実行: chunker.py --size 256 → 再インデックス → 再検索。

1. クエリ拡張の自動ルール化

• 「疑問詞」「日付」「スコープ」を必須抽出し、欠けていれば補完してから再検索する。
• 実装: preprocess.pyでルール適用→expand_query.pyへ流す。

1. 重複除去とフェイルセーフ

• 再取得結果は重複を消してから先に進める（IDベース）。全候補で情報の有無を判定し、閾値以下なら外部検索（Web API）をトリガー。
• コマンド例: retrieve.py --k 200 | jq unique_by(.id) > candidates.json

まとめ：まずは「kを増やす＋クエリ拡張」の組合せを手早く入れてみてください。まずの一歩として、手元のパイプラインで上記のpgvectorクエリと expand_query コマンドを1セットで動かして効果を測ることをおすすめします。

⑤再ランクで候補を絞り込み（Cross-encoder推奨）

検索で候補は拾えても、上位が正解かどうかを人手で確かめている時間がムダだと感じていませんか。

そもそもどう動くのか：Retrieverで広く候補を拾ったあと、Cross-encoder（queryとdocをまとめて入力するモデル）で「1対1の関連度」を高精度にスコア付けする。二段構成にすることで高速な初期検索と高精度な最終選別を両立できる。

1. candidates.jsonl を用意する

• フォーマットは1行1ドキュメント。最低限 id と text を入れておく。
• 例:

{"id": "doc1", "text": "この文書は..."}
{"id": "doc2", "text": "別の候補..."}

1. CLIでまず試す（即効で動作確認）

• 例: rerank スクリプトがあるならまず下記で動かしてみる。

python rerank.py --model reranker --query "${QUERY}" --candidates candidates.jsonl --top 5

• 出力は上位5件を JSON/JSONL で保存する設定にしておく。

1. Pythonでバッチ再ランク（transformers系のCrossEncoder例）

• query と docs のペアを作って一括スコア算出する。簡潔な例:

from sentence_transformers import CrossEncoder
cross_encoder = CrossEncoder("reranker")
pairs = [(query, doc["text"]) for doc in docs]
scores = cross_encoder.predict(pairs)  # [s1, s2, ...]

• 大量候補はチャンク処理でメモリとレイテンシを制御する（下の手順4参照）。

1. バッチ/デバイス制御で実運用化

• chunk_size を決め、GPUがある場合はバッチサイズを上げる。擬似コード：

scores = []
for i in range(0, len(pairs), chunk_size):
    scores.extend(cross_encoder.predict(pairs[i:i+chunk_size]))

• chunk_size を 16/32/64 で試し、メモリとレイテンシのバランスを測る。

1. スコア正規化と上位抽出

• ソートして top-k を取るのは基本。信頼度比較したければ min-max か softmax で正規化する。

import numpy as np
probs = np.exp(scores) / np.sum(np.exp(scores))
top_idx = np.argsort(scores)[-k:][::-1]

1. 閾値判定で「再検索するか」を自動化

• 最高スコアが閾値未満なら、クエリ拡張（同義語追加・コンテキスト増）や再検索を自動でトリガーする。例: if max(scores) < 0.35: rerun retrieval with expanded_query

1. 出力保存とモニタリング

• 最終上位を JSONL に書き出し、QA や人手検査のログと突合する。

with open("top5.jsonl","w") as f:
    for i in top_idx:
        f.write(json.dumps({"id": docs[i]["id"], "score": float(scores[i])}) + "\n")

• 定期的にヒューマンラベルを回して精度変動を監視する。

まとめ：Cross-encoderで再ランクすると上位品質が格段に安定する。まずは小さい候補セット（candidates.jsonl）で上のCLIとPython例を回し、batch/thresholdをチューニングしてください。次の一歩：まず手元で candidates.jsonl を作って、上の CLI コマンドを一度実行してみましょう。

⑥フェールセーフ：代替ソースとヒューマンパスを用意する

RAGが誤答を出したときにそのまま放置していると、信頼が一気に失われる——対処できる仕組みがなければ運用が止まる。

そもそもどう動くのか：まず「誤答を検知」→「自動で追加ソースを取得して再索引」→「再検索して回答を更新」→「それでもダメなら人に通知・対応」の順でフォールバックする。検知と自動化がポイントで、手順化すれば運用コストは小さい。

7ステップ（実践できる実装例付き）

1. 失敗検知ルールを作る

   • 例: 出力に引用が無い、semantic similarityが閾値以下、あるいはモデルが"不確か"フラグを返したら失敗扱い。
   • 実装例: check_answer.py でスコアを計算して exit code を返す。
     コマンド: python check_answer.py --answer-file /tmp/resp.json || exit 1

2. まず再検索（パラメータ変更）を試す

   • retrieverのtop_kやドキュメント重みを変えて再照会。
   • 例: python rag_query.py --query "Q" --top_k 20

3. 外部Webフェッチでソースを補強

   • まず関連URLを自動取得して原典を取り込む。
   • 具体例: wget -O /tmp/page.html 'https://example.com/doc' && python index_document.py /tmp/page.html

4. キャッシュ/ナレッジベースの別インデックスを引く

   • 直近のFAQや手動承認済みドキュメントを優先する。
   • 例: python retriever.py --index approved_faqs --query "Q"

5. 追加ソースをキューに入れて非同期で再索引

   • 一時的にS3やメッセージキューに保存してバッチでindexing。
   • 例: aws s3 cp /tmp/page.html s3://my-bucket/new-sources/ && enqueue-job indexer new-sources/page.html

6. ヒューマンパス（通知とエスカレーション）を用意する

   • 自動で解決できない場合は即座に人に渡す。Slackやチケットを投げる。
   • 具体例（Slack通知）:
     curl -X POST -H "Content-type: application/json" --data '{"text":"RAG failed for query:Q. Auto-fetch attempted:https://example.com/doc"}' $SLACK_WEBHOOK

7. レポート＆学習ループを回す

   • ヒューマン解決後に正解ドキュメントを取り込み、retrieverの候補やルールを更新。
   • 例: human_accept.sh が受け取ったファイルを index_document.py に投げる: bash human_accept.sh /tmp/correct.html

実用スクリプト例（部分）

wget -O /tmp/page.html 'https://example.com/doc' && \
python index_document.py /tmp/page.html && \
python rag_query.py --query "$Q" --top_k 20 || \
curl -X POST -H "Content-type: application/json" --data "{\"text\":\"RAG failed for query: $Q\"}" $SLACK_WEBHOOK

まとめ：まずは「検知→自動取得→再検索→通知」の最短ルートを作ること。まずは1つのFAQページで上記フローを回して、失敗ケースと人の対応時間を計測してみてください。

⑦観測性・SLOとキャッシュ戦略で安定運用する

RAGの誤答でユーザーに謝り続けていませんか？

そもそもどう動くのか：RAGは「検索→統合→生成」のループ。観測性はその各段階（リクエスト数、検索回数、生成失敗）に指標を振り、SLOで許容外の挙動を検知してアラートを張る。キャッシュは負荷とコストを下げる一方で古い答えを返すリスクがあるため、TTLと再検証（evidence-check）で自己修復させるのが肝。

7選（具体手順・コマンド付き）

1. メトリクスを定義して吐く
   • 必須カウンタ: rag_requests_total, rag_failures_total, rag_retrievals_total。遅延はヒストグラム rag_latency_seconds。Python例：

   from prometheus_client import Counter, Histogram
   req = Counter('rag_requests_total','...')
   fail = Counter('rag_failures_total','...')
   lat = Histogram('rag_latency_seconds','...')
   req.inc()
   with lat.time(): ...

2. SLOとエラーバジェットの計算式を決める
   • エラー率 = rate(rag_failures_total[5m]) / rate(rag_requests_total[5m])。まずはこの比率で閾値を決める。
3. Prometheusアラートを作る（例）

   groups:
   - name: rag.rules
     rules:
     - alert: RAGHighErrorRate
       expr: rate(rag_failures_total[5m]) / rate(rag_requests_total[5m]) > 0.01
       for: 5m
       labels:
         severity: page

4. RedisでTTLキャッシュを使う（キーはクエリ+ドキュメントのハッシュ推奨）
   • ハッシュ生成例（Linux）：

   echo -n "query|doc_fingerprint" | sha256sum | awk '{print $1}'

   • キャッシュ保存例：

   redis-cli SETEX rag:cache:<hash> 3600 '<response-json>'

5. キャッシュ無効化・更新ルール
   • ソース更新時は該当ハッシュをDEL：

   redis-cli DEL rag:cache:<hash>

   • TTL切れで自動再検索される設計にする。
6. キャッシュヒット時も“簡易検証”を挟む
   • キャッシュ内に retrieval_time と evidence_score を入れておき、score が閾値未満なら再検索。簡単な判定例（Python）：

   data = json.loads(redis.get(key))
   if data['evidence_score'] < 0.6 or time.time()-data['retrieval_time']>3600:
       requery()

7. ダッシュボードと運用手順を整備する
   • 表示パネル：error rate、avg retrieval latency、cache hit ratio。簡易Cache Hit率PromQL例：

   rate(rag_cache_hits_total[5m]) / (rate(rag_cache_hits_total[5m]) + rate(rag_cache_misses_total[5m]))
   

   • Runbook：アラート発生時の切り分け（ログ→再現→強制キャッシュ削除→SLO評価）。

まとめ：観測指標とTTL付きキャッシュを組み合わせ、アラートで自動検知→条件付きで再検索する流れを作ればRAGの安定性は大きく改善する。まずは1週間で上記のメトリクスを立て、Prometheusルールを1つ張って様子を見ましょう。

まとめ

要点を簡潔に振り返ると、まず誤答を可視化するためにログ収集と簡易検証器を入れ、検出された失敗に対して再検索→再ランクの自動ループを段階的に組み込むことが肝心です。優先度の高いループから1つずつ実装して効果を測りながら改善していきましょう。まずは①を実装してログを数日分集めることを次の一歩にしてください。