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

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

目次

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

自己修復の全体像（超短縮）
検知 → 判断 → 再取得（幅を広げる）→ 再ランク → 再生成 → 必要なら人へエスカレーション、を自動で回す。各回答には必ずメタ情報を付け、これをトリガーに使います。
メタ＆ログ設計（まずはここを作る）
必ず保存するメタの例（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の分布を確認するところから始めると閾値設計が楽になります。

軽量 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})

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

再検索を起動する閾値設計（具体値例）
小さく始めて運用でチューニング。僕が実運用で試した初期値の例：

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

再取得戦略：幅を広げる（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

再ランク：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 を大きくして高速化。閾値未満ならさらに再取得フェーズへ戻すループを入れる。

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

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

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

Slack 通知（curl）:

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

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

観測性とキャッシュ戦略（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)"

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

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

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

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