RAGの回答が「昔の仕様」を答えるようになって調査したら、検索トップ10のうち8件が同じ文書の古い版でした。版更新のたびに「追加→埋め込み→保存」だけを続けて、古いベクトルを削っていなかったのが、私が1年前に踏んだ失敗の根本原因です。この記事では、問題を再現する完全スクリプトと、3層の対策（最新版だけ索引化・MinHash重複除去・検索時フィルタ）を、そのまま実行できる形で書きます。
何が起きたか: トップ10のうち8件が同一文書の旧版
社内マニュアル（約3,700チャンク）を索引化したRAGで、利用者から「回答の内容が前の仕様のまま」という指摘が続きました。調査用スクリプトで「バッチ処理が失敗したときはどうなりますか？」を投げて検索結果を並べたところ、次の状態でした。
| 順位 | doc_id | version | 状態 |
|---|---|---|---|
| 1 | manual_a | 3 | 最新版（正しい回答の素） |
| 2 | manual_a | 2 | 旧版（Slack通知だけの記述） |
| 3 | manual_a | 1 | さらに旧版（cron・メール通知時代） |
| 4〜8 | manual_a | 1〜3 | 旧版のチャンクが計5件 |
| 9 | runbook_b | 2 | 他文書（ようやく1件目） |
| 10 | faq_c | 1 | 他文書 |
つまりトップ10の8割がmanual_a関連で、多様性はゼロ。LLMのコンテキストに渡る本文は旧版が多数派を占めるので、モデルは多数派に合わせた「旧仕様の回答」を生成しました。runbook_bが圏外になることで「初動の対応手順」も回答から消えています。原因は検索の精度ではなく、索引の中身が汚れていたことでした。
重複は3種類。対策する層が違う
索引の中身を汚す重複を仕分けした結果が次の表です。
| 重複の種類 | 起きる場所 | 具体例 | 対策する層 |
|---|---|---|---|
| 完全複製 | 取り込み | 同一ファイルの再アップロード、コピペ誤登録 | 取り込み時のMinHash検査 |
| 近似複製 | 取り込み | 軽微な編集だけして再取り込みした版 | 取り込み時のMinHash検査＋最新版選別 |
| 旧版の残存 | 索引の寿命 | 版更新で古いベクトルの削除を忘れた | 索引化最新版選別＋検索時フィルタ |
1種類目と2種類目は取り込み口で遮断でき、3種類目だけは「すでに汚れた索引」と「将来の事故」の双方に効く対策が必要です。1つのコードで全部を防ごうとするより、層で分けた方が実際に書く量は少なくなりました。
直し方1（根本対策）: doc_idごとに最新版だけを索引化する
索引化の入口でdoc_idごとにversion最大の1件だけを残します。次のスクリプトは、対策前と対策後の検索結果の違いをそのまま確認できる完全な形です。
#!/usr/bin/env python3
"""doc_idごとに最新版だけを索引化して検索する完全スクリプト。
pip install numpy openai
環境変数 OPENAI_API_KEY を設定してから実行してください。"""
import os
import numpy as np
from openai import OpenAI
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
# 取り込み元: 同じ文書の複数版が混ざって届く状態（実際の事故を縮小再現）
DOCS = [
{"doc_id": "manual_a", "version": 1, "text": "バッチはcronで毎日2時に起動する。失敗時はメールで通知する。"},
{"doc_id": "manual_a", "version": 2, "text": "バッチはGitHub Actionsで毎日2時に起動する。失敗時はSlackへ通知する。"},
{"doc_id": "manual_a", "version": 3, "text": "バッチはGitHub Actionsで毎日2時に起動する。失敗時はSlack通知とリトライ3回の冪等再実行を行う。"},
{"doc_id": "runbook_b", "version": 1, "text": "障害対応の初動はダッシュボードでエラー件数を確認する。"},
{"doc_id": "runbook_b", "version": 2, "text": "障害対応の初動はダッシュボード確認のあと、エラーログをJSONで保存してSlack共有する。"},
{"doc_id": "faq_c", "version": 1, "text": "プランの変更は設定ページからいつでもできる。"},
{"doc_id": "backup_d", "version": 1, "text": "退会処理はサポート窓口への申請が必要になる。"},
]
def embed(texts):
res = client.embeddings.create(model="text-embedding-3-small", input=texts)
vecs = np.array([d.embedding for d in res.data], dtype=np.float32)
return vecs / np.linalg.norm(vecs, axis=1, keepdims=True)
def dedupe_to_latest(docs):
"""doc_idごとにversion最大の1件だけを残す（根本対策）"""
latest = {}
for d in docs:
cur = latest.get(d["doc_id"])
if cur is None or d["version"] > cur["version"]:
latest[d["doc_id"]] = d
return list(latest.values())
def search(docs, vecs, query_text, k=10):
q = embed([query_text])[0]
scores = vecs @ q
order = np.argsort(-scores)[:k]
return [(i, scores[i]) for i in order]
if __name__ == "__main__":
query = "バッチ処理が失敗したときはどうなりますか？"
print("=== 対策前: 混ざったまま全部索引化した場合 ===")
vecs_all = embed([d["text"] for d in DOCS])
for rank, (i, s) in enumerate(search(DOCS, vecs_all, query), 1):
d = DOCS[i]
print(f"{rank:2d}. {s:.4f} {d['doc_id']} v{d['version']}")
print("=== 対策後: 最新版だけを索引化した場合 ===")
index_docs = dedupe_to_latest(DOCS)
vecs_clean = embed([d["text"] for d in index_docs])
for rank, (i, s) in enumerate(search(index_docs, vecs_clean, query), 1):
d = index_docs[i]
print(f"{rank:2d}. {s:.4f} {d['doc_id']} v{d['version']}")
実行すると、対策前はmanual_aのv3→v2→v1が上位を占めるのに対し、対策後はmanual_a v1件・runbook_b v2件が並び、旧版が消えます。なおversionは「採番規則が文書管理側で保証されている」前提です。採番がない場合は更新日時をversionの代わりに使うと同じ構造のまま動きます。
直し方2（掃除と防御）: MinHashで近似複製を取り込み時に落とす
すでに汚れた索引の一括掃除と、今後のコピペ誤登録の防御に使うのがMinHashです。埋め込みAPIを呼ばずにテキストだけで重複判定できるので、取り込みパイプラインの前段に置くのに向いています。
#!/usr/bin/env python3
"""取り込み時の近似複製チェック。既存索引の掃除にも使える。
pip install datasketch"""
import re
from datasketch import MinHash, MinHashLSH
def shingles(text, k=3):
cleaned = re.sub(r"[、。！？\s]", "", text)
return {cleaned[i:i + k] for i in range(len(cleaned) - k + 1)}
def make_minhash(text):
m = MinHash(num_perm=128)
for s in shingles(text):
m.update(s.encode("utf-8"))
return m
INCOMING = [
{"doc_id": "faq_c", "version": 1, "text": "プランの変更は設定ページからいつでもできる。"},
{"doc_id": "faq_c", "version": 2, "text": "プランの変更は設定ページからいつでもできる。反映には5分ほどかかります。"},
{"doc_id": "manual_a", "version": 3, "text": "バッチはGitHub Actionsで毎日2時に起動する。失敗時はSlack通知とリトライ3回の冪等再実行を行う。"},
]
lsh = MinHashLSH(threshold=0.5, num_perm=128)
kept = []
for d in INCOMING:
key = f"{d['doc_id']}:v{d['version']}"
m = make_minhash(d["text"])
hits = lsh.query(m)
if hits:
print(f"スキップ: {key}（既存 {','.join(hits)} と近似複製）")
continue
lsh.insert(key, m)
kept.append(d)
print("=== 索引化を通過した文書 ===")
for d in kept:
print(f"{d['doc_id']} v{d['version']} : {d['text']}")
私の環境では、threshold=0.5で「数行の編集だけの再取り込み」が確実に引っかかる一方、章の半分を書き換えた版は通りました。通った版はdoc_id+versionの選別（直し方1）で処理されるので、責務が重ならずに済みます。逆に言えば、MinHashだけで旧版問題を完結させようと閾値を0.3まで下げたら、別文書なのに内容が似ている手順書までスキップする事故が起きました。重複検知は「防御の1枚」であって唯一の対策にしないのが安全です。
直し方3（安全網）: 検索時にdoc_idで間引く。本番はQdrantのグループ検索
運用中の索引はすぐには作り直せないので、検索側にも安全網を付けます。まずnumpyだけで書いた間引き関数です。直し方1のsearch()の戻り値に差し替える形で使えます。
def diversify_by_doc(order, docs, k=10, per_doc=1):
"""同じdoc_idが上位を占拠したら、doc_idごとにper_doc件へ間引く"""
picked, seen = [], {}
for i in order:
did = docs[i]["doc_id"]
if seen.get(did, 0) >= per_doc:
continue
seen[did] = seen.get(did, 0) + 1
picked.append(i)
if len(picked) == k:
break
return picked
本番ではベクトルDB側で同じことをさせます。Qdrantのグループ検索は「doc_idごとに上位1件」をサーバー側で保証してくれるので、応答ペイロードを作る側のコードが不要になります。
#!/usr/bin/env python3
"""本番形: Qdrantのグループ検索で「doc_idごとに上位1件」をサーバー側で保証する。
pip install qdrant-client numpy openai
環境変数 OPENAI_API_KEY を設定してから実行してください。"""
import os
import numpy as np
from openai import OpenAI
from qdrant_client import QdrantClient, models
oai = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
client = QdrantClient(":memory:") # 本番では QdrantClient(url="https://...") を指定する
DOCS = [
{"doc_id": "manual_a", "version": 3, "text": "バッチはGitHub Actionsで毎日2時に起動する。失敗時はSlack通知とリトライ3回の冪等再実行を行う。"},
{"doc_id": "runbook_b", "version": 2, "text": "障害対応の初動はダッシュボード確認のあと、エラーログをJSONで保存してSlack共有する。"},
{"doc_id": "faq_c", "version": 1, "text": "プランの変更は設定ページからいつでもできる。"},
{"doc_id": "backup_d", "version": 1, "text": "退会処理はサポート窓口への申請が必要になる。"},
]
def embed(texts):
res = oai.embeddings.create(model="text-embedding-3-small", input=texts)
vecs = np.array([d.embedding for d in res.data], dtype=np.float32)
return vecs / np.linalg.norm(vecs, axis=1, keepdims=True)
client.create_collection(
collection_name="docs",
vectors_config=models.VectorParams(size=1536, distance=models.Distance.COSINE),
)
vecs = embed([d["text"] for d in DOCS])
client.upsert(
collection_name="docs",
points=[
models.PointStruct(
id=i,
vector=vecs[i].tolist(),
payload={"doc_id": d["doc_id"], "version": d["version"], "text": d["text"]},
)
for i, d in enumerate(DOCS)
],
)
query = "バッチ処理が失敗したときはどうなりますか？"
qvec = embed([query])[0]
groups = client.search_groups(
collection_name="docs",
query_vector=qvec.tolist(),
group_by="doc_id",
group_size=1,
limit=10,
)
for g in groups.groups:
p = g.hits[0]
print(f"{p.score:.4f} {p.payload['doc_id']} v{p.payload['version']} {p.payload['text'][:20]}")
インメモリモードなのでサービス不要で動きます。search_groupsはqdrant-client 1.x系のAPIで、新しいバージョンではquery_points_groupsへの移行案内が出ることがありますが、引数の意味は同じです。payloadにdoc_idとversionを持たせておくと、検索結果の監査（「この回答はどの版に基づいているか」の調査）も一行でできるようになり、事故調査の時間が半日から数分に縮みました。
失敗した設計: 「再取り込み＝全置換」だと思い込み、追加だけを1年続けた
振り返ると、私のインデクサには「同一doc_idの旧versionを削る」処理が最初からありませんでした。私がこのとき想定していた動作と、実際の動作の差が次の表です。
| 想定していた動作 | 実際の動作 |
|---|---|
| upsertすれば同じ文書の旧ベクトルは消える | doc_idが違うので別ポイントとして追加されるだけ |
| 版が混ざってもスコア上位は常に最新版 | 古い版は最新版と本文が似ていて高スコアになり、上位を占拠する |
| 重複は索引一覧を見れば気づける | 近似複製は検索順位の中にしか現れず、一覧では見えない |
加えて、「ファイル名に更新日を付けて保存する」運用ルールがdoc_idをバラバラにし、同じ文書なのに別文書として登録される状態を作りました。doc_idの設計が崩れると、MinHashなどの重複検知も「別文書の類似ペア」を増やすだけになって効きにくくなります。ベクトルDBは追記が基本動作で、「置換」は自分で書くコードです。ライブラリが暗黙にやってくれるものではない、というのがこの1年の一番の学びでした。
対策前後の比較
| 観点 | 対策前 | 対策後 |
|---|---|---|
| トップ10の内訳 | 同一文書の旧版が8件 | doc_id重複なし・各文書1件ずつ |
| 回答内容 | 旧版が多数派のため旧仕様（メール通知時代）で回答 | 最新仕様（Slack通知＋冪等再実行）を回答 |
| 索引サイズ | 旧版蓄積で目安の1.4倍 | 最新版だけで約3割減 |
| 旧版混入の再発 | 検索結果を見ても判別不能 | 取り込み時（MinHash）と検索時（グループ検索）の二重防御 |
まとめ
- ベクトル索引は追記が基本。「版の置換」を自分のコードで書いていないなら、旧版は確実に溜まっている
- 根本対策は索引化の入口でdoc_idごとに最新版だけを残すこと。対策は最小のPythonコード数行で書ける
- MinHashは取り込み口の防御。閾値で重複の取りこぼしと誤検知のトレードオフが決まるので、唯一の対策にしない
- 安全網として検索時のグループ化（doc_idごとに上位1件）を入れると、汚れた索引の影響を応答から隔離できる
- 次のアクション: 自分のインデクサを1つ開き、「旧versionの削除処理」が存在するか今日確認する。なければ本記事のdedupe_to_latest()を入口に足すのが最初の一手です
関連記事（同テーマ）
参考書籍: RAG構築入門 — LLMに「知識」を与える検索拡張生成の実践（Kindle・Kindle Unlimited読み放題対象）（葉山悠希 著）