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?

@kenji_harada

運用で効く：誤答を自動で潰す Self‑Healing RAG 実践ガイド

0
Posted at

この記事は自社ブログ(nands.tech)の要約版です

はじめに
モデルが自信満々に間違った答えを返し、人が手で直す――そんな運用はもう終わりにしましょう。僕（原田賢治）が現場で試した「自動で失敗を検知して再取得・再評価する」ワークフローを、実践的なスクリプト例とともに整理しました。まずは小さなループ（ログ化→検知→再検索）を作るところから始めてください。

目次

    1. 概要: 自己修復の基本ループ
    1. メタ＆ログ設計（必須）
    1. 軽量 verifier（ルール＋短プロンプト）
    1. 再検索トリガー（閾値具体例）
    1. 再取得戦略：幅を広げる
    1. 再ランク（Cross‑encoder 実例）
    1. フェールオーバーとヒューマンパス
    1. 観測性・キャッシュ運用
  • 次の一手（実行プラン）

  1. 自己修復の全体像（超短縮）
    検知 → 判断 → 再取得（幅を広げる）→ 再ランク → 再生成 → 必要なら人へエスカレーション、を自動で回す。各回答には必ずメタ情報を付け、これをトリガーに使います。

  2. メタ＆ログ設計（まずはここを作る）
    必ず保存するメタの例（JSON）。これがトリガーや解析の基盤になります。

{
  "id": "resp-0001",
  "created_at": "2026-07-24T10:00:00Z",
  "query": "営業報告のテンプレートは？",
  "answer": "〜（本文）〜",
  "model": "llm-v1",
  "sources": ["doc:123","doc:456"],
  "top_scores": [0.12,0.23,0.35],
  "avg_score": 0.23,
  "verifier_score": 0.48,
  "retries": 0
}

サンプルDBテーブル（Postgres + pgvector を想定）:

CREATE TABLE rag_events (
  id TEXT PRIMARY KEY,
  query TEXT,
  response_json JSONB,
  retrieval_ids TEXT[],
  created_at TIMESTAMP WITH TIME ZONE DEFAULT now()
);

簡易ログ書き込み（Python / psycopg2）:

# log_writer.py
import json, psycopg2
def write_log(conn, payload):
    cur = conn.cursor()
    cur.execute("INSERT INTO rag_events (id, query, response_json, retrieval_ids) VALUES (%s,%s,%s,%s)",
                (payload['id'], payload['query'], json.dumps(payload), payload.get('sources',[])))
    conn.commit()

運用メモ: まずは24時間分のログを集め、top_scoresの分布を確認するところから始めると閾値設計が楽になります。

  1. 軽量 verifier を入れる（ルール＋短検証モデル）
    「毎回大きな検証モデルを呼ばずに最初のフィルタを掛ける」のが肝心。僕は次の2段構成で運用している。
  • ルール層（正規表現、必須語句、参照有無）
  • 短い検証プロンプトを使ったライトモデル（短タイムアウト）

例: verifier_mini.py（擬似）

# verifier_mini.py
import re, json, requests, sys
def rule_check(answer):
    if not re.search(r'[出典|参考]', answer):
        return ("MISSING_SOURCE", 0.0)
    return ("PASS", 0.9)

def model_check(query, answer):
    # 短いプロンプトで「妥当か」を聞く（タイムアウト短め）
    r = requests.post("http://localhost:8000/mini-llm", json={"prompt": f"Q:{query}\nA:{answer}\n妥当なら1.0, 不妥当なら0.0で返せ"})
    return ("MODEL_OK" if r.json().get("score",0)>0.7 else "MODEL_BAD", r.json().get("score",0))

q, a = sys.argv[1], open(sys.argv[2]).read()
r1 = rule_check(a)
r2 = model_check(q,a)
print({"rule":r1,"model":r2})

判定結果はログに格納して、自動再検索のトリガー判定に使います。

  1. 再検索を起動する閾値設計（具体値例）
    小さく始めて運用でチューニング。僕が実運用で試した初期値の例：
  • sources が空 → 即再検索
  • avg_score < 0.30 → 再検索（k を増やす）
  • max(top_scores) - second_score < 0.03 → 曖昧と見なし再検索
  • verifier_score < 0.6 → 再検索 or 人へ投げる（retriesカウンタ次第）
  • consecutive_retries >= 2 → ヒューマンパス

擬似コード:

def needs_refetch(meta):
    if len(meta['sources']) == 0: return True
    if meta['avg_score'] < 0.30: return True
    if meta['verifier_score'] < 0.6 and meta['retries'] < 2: return True
    return False
  1. 再取得戦略：幅を広げる（k増加＋パラフレーズ＋BM25ハイブリッド）
    再取得は"同じ検索をもう一度"ではなく戦略を変えて投げること。
  • ステップ化する例: k=10 → k=50 → k=300
  • クエリ拡張: 上位3件の要約＋元クエリを用いてLLMに「検索語を改善」してもらう
  • 別アルゴリズム: 埋め込み検索で500件取って、その集合で BM25 を回す（ハイブリッド）
  • チャンク粒度を切り替える（512→256 token）

再検索スクリプト例（shell）:

# retriever_expand.sh
QUERY="$1"
python embed_query.py --query "$QUERY" --k 10 > hits.json
if python need_widen.py hits.json; then
  python expand_query.py --hits hits.json --query "$QUERY" > expanded.txt
  curl -s -X POST http://vector-db.local/query -d "{\"query\":\"$(cat expanded.txt)\",\"top_k\":200}" > hits_wide.json
fi
  1. 再ランク：Cross‑encoder の入れ方（実践コード）
    広く拾った候補は Cross‑encoder で精査してから生成に回すと精度が大きく改善します。sentence-transformers の CrossEncoder を使った例:
# rerank_ce.py
from sentence_transformers import CrossEncoder
import json
ce = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
with open("candidates.jsonl") as f:
    docs = [json.loads(l) for l in f]
pairs = [(QUERY, d['text']) for d in docs]
scores = ce.predict(pairs, batch_size=16)
ranked = sorted(zip(docs, scores), key=lambda x: x[1], reverse=True)[:5]
print([{"id":d['id'],"score":s} for d,s in ranked])

運用TIP: GPUがある環境では batch_size を大きくして高速化。閾値未満ならさらに再取得フェーズへ戻すループを入れる。

  1. フェールセーフとヒューマンパス
    自動化でどうしても直らないケースは人に回す。フローは短くするほど良い。

サンプルの最短フェールオーバー:

  1. 自動再取得ループを MAX_RETRIES 回試す
  2. それでも解決しない場合は Slack に通知してチケットを作成
  3. 人が解決したら、そのソースを取り込んで再インデックス

Slack 通知（curl）:

curl -X POST -H "Content-type: application/json" \
  --data "{\"text\":\"[RAG] 自動修復失敗: query=${Q} / id=${ID}. 対応お願いします.\"}" \
  $SLACK_WEBHOOK

インデックス後の学習ループ（人が承認したら自動で取り込む）を用意しておけば、同じ失敗は減っていきます。

  1. 観測性とキャッシュ戦略（SLO と再検証）
    安定運用にはメトリクスとキャッシュルールが不可欠。

Prometheus のメトリクス例（Python）:

from prometheus_client import Counter, Histogram
REQ = Counter('rag_requests_total','total requests')
FAIL = Counter('rag_failures_total','failures')
LAT = Histogram('rag_latency_seconds','latency')

キャッシュ運用のベストプラクティス:

  • TTL を短め（例 1時間）に設定
  • キャッシュヒット時でも簡易 verifier を実行し、evidence_scoreが低ければ再取得
  • キャッシュキーは query の正規化 + ドキュメント指紋のハッシュ

Redis 保存例:

KEY=$(printf "%s|%s" "$QUERY" "$DOC_FP" | sha256sum | awk '{print $1}')
redis-cli SETEX rag:cache:$KEY 3600 "$(cat response.json)"

次の一手（短期実行プラン）

  1. 24時間分のログを取る（まずは1週間のサンプルを目標）
  2. verifier_mini.py を本番の1%トラフィックで有効化する（false positives を観察）
  3. retriever_expand.sh と rerank_ce.py を組み合わせて "再取得→再ランク→再生成" の1ループを動かす

最後に（僕の勧め）
最初から全部作ろうとせず、小さなループを1つ入れて効果を確認してください。僕は「ログ化→verifier（ルール層）→k増加の1回リトライ」をまず週内に入れることを推奨します。ここまで作れば、残りは閾値調整と運用で解決します。

詳細な実装手順はこちら → https://nands.tech/posts/self-healing-rag-design-jua09i

0
0
0

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
Sign upLogin
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?