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?

@YushiYamamoto(山本 勇志)

RAGの更新遅れをどう検知するか：同期成功時刻では拾えない旧版と削除漏れ

0
Posted at

2026-10-22時点。RAGの運用を設計するエンジニア・テックリードに向けて、ナレッジの鮮度を何で監視し、いつ再取込するかを考えます。

結論：同期の成功と原本への追随を別々に測る

更新・削除の取りこぼしが許容できないなら、ジョブの最終成功時刻だけで正常判定しない、という設計を推奨します。原本の文書ID・版と索引を照合し、未取込、旧版、削除漏れを分けます。原本を取得できないときは「差分ゼロ」ではなく「判定不能」にします。

少量・低頻度で、原本を人が確認してから回答する運用なら、定期的な手動照合を残す選択もあります。原本を完全に列挙できず、削除の記録も取れない場合は、自動削除の採用を保留します。

本稿の実行結果は、AI支援で準備・実行した架空データのローカル検証です。実ベクトルDBやLLMを動かした結果、実務での導入成果、本人の独力実装を示すものではありません。

前提：どの遅れを検知したいのか

改訂済みのFAQが旧版のまま検索されたり、廃止済みの手順が回答の根拠に残ったりする状態を対象にします。次の3段階を分けます。

  1. 原本の変更を取得できたか。
  2. 取得した版を索引へ適用できたか。
  3. 実際の検索経路から、その版が見えるか。

Azure AI SearchのAzure Storage索引では、更新検知と削除検知は同じ扱いではありません。削除を索引へ反映するには、対応する削除検知の設計が必要です。Microsoftの公式資料

また、Elasticsearchには変更を検索可能にするrefreshがあります。これは「原本を取り込めたか」と別の段階です。ここから、書込み成功と検索可能の確認も分けるという設計判断につながります。Elasticの公式資料

今回検証するのは2段階目までです。1段階目は固定した完全な原本スナップショットを入力し、3段階目の実検索は未検証とします。

選択肢：全件照合を常時回す必要はあるか

方法 判断できること 見落とす条件・負担 採用する条件
現状維持：最終成功時刻だけ監視 ジョブ停止の兆候 一部だけ更新失敗、削除漏れ 鮮度保証には使わず稼働監視に限定
手動で原本と検索結果を確認 少数の重要文書の追随 確認対象外、確認間隔の遅れ 少量で人の確認を残せる
既製の取込機能・コネクタ 製品が対応する差分同期 削除方式、再実行、可視化の制約 要件を満たす機能が既にある
ID・版の全件照合 対象範囲の欠落・旧版・余剰 読取り量、原本の完全列挙が必要 小規模、定期監査、復旧時
変更イベント＋定期照合 日常の差分処理と取りこぼし補完 イベント保持、順序、照合運用 全件走査を毎回行う費用が大きい

本稿では欠落・旧版・余剰を一度に確かめるため、全件照合を選びます。製品の取込機能を置き換える提案ではなく、何を監視できれば採用できるかを確かめる最小モデルです。

検証：件数が同じでも中身は一致しない

環境はPython 3.11.8、SQLite 3.51.0。追加ライブラリ不要、メモリ内DB、ネットワーク呼出しなしです。文書1件を索引1行として扱い、同じ対象範囲を完全取得できた前提に固定します。

文書ID 原本の期待状態 初期索引 判定
faq v2 / new answer v1 / old answer stale：旧版
guide v1 / current guide なし missing：未取込
retired 存在しない v1 / withdrawn answer orphan：余剰候補

原本も索引も2件です。別の処理が成功して最終成功時刻が更新されても、この不一致は解消しません。検証では時刻をモデル上「成功直後」としても、判定には使わず差分を検出しました。スケジューラー自体の動作は試していません。

次は検証に使った処理と主要経路の再現コードです。freshness_demo.pyへ保存し、python3 freshness_demo.pyで実行できます。照合結果がmatchedでも、意味は「この入力スナップショットと一致」であり、現在の原本・本番検索まで最新と保証するものではありません。

import sqlite3

def read_index(db):
    return {k: (rev, body) for k, rev, body in
            db.execute('SELECT id, revision, body FROM docs')}


def audit(db, source, *, complete):
    if not complete:
        return {'status': 'unknown', 'missing': [], 'stale': [], 'orphan': []}
    # `complete` must come from successful enumeration, not an empty-result guess.
    for key, value in source.items():
        if (not isinstance(key, str) or not key
                or not isinstance(value, tuple) or len(value) != 2
                or not all(isinstance(v, str) for v in value) or not value[0]):
            raise ValueError('invalid source record')
    indexed = read_index(db)
    missing = sorted(source.keys() - indexed.keys())
    stale = sorted(k for k in source.keys() & indexed.keys()
                   if source[k] != indexed[k])
    orphan = sorted(indexed.keys() - source.keys())
    return {'status': 'drift' if missing or stale or orphan else 'matched',
            'missing': missing, 'stale': stale, 'orphan': orphan}


def sync(db, source, *, complete, fail=False):
    result = audit(db, source, complete=complete)
    if result['status'] == 'unknown':
        raise RuntimeError('source snapshot incomplete; no writes')
    # The caller must not have a pending transaction on this dedicated connection.
    if db.in_transaction:
        raise RuntimeError('pending transaction')
    with db:
        for key in result['orphan']:
            db.execute('DELETE FROM docs WHERE id = ?', (key,))
        if fail:
            raise RuntimeError('injected failure after delete')
        for key in result['missing'] + result['stale']:
            rev, body = source[key]
            db.execute('INSERT INTO docs VALUES (?, ?, ?) '
                       'ON CONFLICT(id) DO UPDATE SET '
                       'revision=excluded.revision, body=excluded.body',
                       (key, rev, body))
    return audit(db, source, complete=complete)


def fixture():
    db = sqlite3.connect(':memory:')
    db.execute('CREATE TABLE docs '
               '(id TEXT PRIMARY KEY, revision TEXT NOT NULL, body TEXT NOT NULL)')
    with db:
        db.executemany('INSERT INTO docs VALUES (?, ?, ?)',
                       [('faq', 'v1', 'old answer'),
                        ('retired', 'v1', 'withdrawn answer')])
    return db


SOURCE = {'faq': ('v2', 'new answer'), 'guide': ('v1', 'current guide')}

db = fixture()
print("before:", audit(db, SOURCE, complete=True))
before = read_index(db)
try:
    sync(db, SOURCE, complete=True, fail=True)
except RuntimeError as error:
    print("failure:", error)
assert read_index(db) == before  # 削除もロールバックされる
assert audit(db, {}, complete=False)['status'] == 'unknown'
try:
    sync(db, {}, complete=False)
except RuntimeError:
    pass
else:
    raise AssertionError("incomplete source must block writes")
assert read_index(db) == before
print("after:", sync(db, SOURCE, complete=True))
assert read_index(db) == SOURCE
changes = db.total_changes
sync(db, SOURCE, complete=True)
assert db.total_changes == changes  # 同じ状態の再適用は書込みなし
print("retry: no additional writes")
db.close()

主要な出力は次のとおりでした。辞書の表示順には依存しません。

before: {'status': 'drift', 'missing': ['guide'], 'stale': ['faq'], 'orphan': ['retired']}
failure: injected failure after delete
after: {'status': 'matched', 'missing': [], 'stale': [], 'orphan': []}
retry: no additional writes

設計上の要点は3つです。

  • 版番号だけを信用しない。 今回は小さな文字列なので本文も比較します。版を更新し忘れた変更も拾えます。実データでは取得した本文のハッシュ等を検討し、抽出・分割方式の版も別に管理します。
  • 取得失敗を空集合に変換しない。 complete=Falseは判定不能で、再取込も削除もしません。complete=Trueは、列挙の全ページ成功・対象範囲の一致を呼出し側が確認して渡す前提です。権限低下による見え方の変化を削除と誤認しない確認も必要です。
  • 失敗したら元の索引を残す。 削除直後に例外を注入し、削除も取り消されることを確認しました。SQLite接続の既定トランザクション設定とwith db:を使う範囲の結果です。設定を変える場合は再検証します。Python 3.11公式資料

別途、同じ処理に対して次の9項目をテストし、9件成功・失敗0件でした。

確認した経路 確認した結果
成功直後扱い・件数一致 欠落・旧版・余剰を各1件検出
通常同期 追加・置換・削除後に入力と一致
同じ入力を再適用 追加の書込みなし
削除後の例外と再試行 元の状態を保持し、再試行後に一致
原本の不完全な空入力 判定不能、書込み拒否、索引保持
完全取得が確認された空入力 全行削除後に一致
版が同じで本文だけ変更 旧状態として検出
不正な版の入力 エラー、索引保持
未確定トランザクションあり 同期を拒否、呼出し側へ判断を返す

本番設計へ移すときに残る課題

この例は単一処理・固定入力です。照合から書込みまでに別処理が更新する競合、複数チャンク、Embeddingの失敗、プロセス強制終了、外部DBのタイムアウトは検証していません。fail=Trueは通常の例外注入で、電源断や強制終了試験ではありません。

特に、ローカルのロールバックを外部ベクトルDBの保証へ置き換えられません。本番では新しい世代を構築して検証後に参照先を切り替える案や、文書ID・原本版を固定した再適用を比較します。書込み結果が不明なら、読戻しで状態を照合してから再試行します。これらは本稿では設計案です。

また、余剰候補の削除には、同じテナント・同じ範囲の完全なスナップショットが必要です。ページ欠落や列挙権限の変更があれば自動削除を止めます。全件列挙が難しい場合は削除記録（tombstone）を残し、処理済み確認後に原本を物理削除する方法を検討します。Azureのnative blob soft deleteでも、保持期間内に索引処理が削除状態を観測できることが条件です。Microsoftの削除検知要件

何を監視し、誰が戻すか

次は未実装の運用案です。時刻はUTC等に統一し、遅延の許容値は文書の用途と原本担当者との合意で決めます。

指標・状態 意味 次の動作

RAG索引の鮮度監視を表す幾何学図
|
| 最終の完全照合からの経過時間 | 監査自体が止まっていないか | 取得経路を運用担当が確認 |
| missing / stale / orphan件数 | 観測対象内の不一致 | 対象IDを限定して再取込・削除確認 |
| 最も古い未反映変更の待ち時間 | 一部の更新が放置されていないか | 許容時間超過で人へ通知 |
| 原本取得不能・範囲不明 | 鮮度を判定できない | 自動削除を止めて保留 |
| 検索プローブの版不一致 | 書込み済みだが検索経路へ未反映 | キャッシュ・検索可視化を確認 |

待ち時間を測るには、未反映の原本変更時刻が必要です。削除時刻が分からなければ、まず「初めて不一致を検出した時刻からの時間」を別名で記録し、削除からの実遅延と混同しません。今回の固定fixtureでは実時間の遅延測定はしていません。

原本担当は廃止・改訂と許容遅延、取込担当は再処理と削除確認、検索担当は実検索での版照合を受け持つ形を提案します。重大な旧情報が見つかった文書は回答根拠から外し、必要なら人の確認へ戻します。ログには許可されたID、期待版、観測版、処理段階、再試行結果を残し、本文や秘密情報を丸ごと出しません。

費用は、原本列挙・索引読取り・再Embedding・監視対応の回数で分けます。全件照合の頻度を増やす前に、許容遅延と対象件数を決めます。本稿ではサービス料金・処理速度・削減効果を測っておらず、費用削減率は示しません。

条件付きの推奨と次に確認すること

小規模なら、まず完全照合で更新・削除の漏れを説明できる状態を作ることを推奨します。対象が増えたら、既存コネクタの削除保証を確認し、変更イベントと定期照合の組合せへ進むかを判断します。取得範囲の完全性を保証できない段階では、自動削除を見送ります。

次の検証は、実検索経路から指定文書の期待版を確認することです。その後に複数チャンクの欠落、更新中の世代混在、削除通知の保持期限超過を試します。「取込ジョブが成功した」から「必要な版が検索でき、廃止版が出ない」まで証拠をつなげて、本番移行を判断します。

参考資料

いずれも2026-10-22に確認。製品資料は仕様上の根拠であり、本稿の環境で製品を実行した証拠ではありません。

この記事を書いた人✏️@YushiYamamoto
ITPRODX.com代表。AI × Web × 業務自動化の開発・検証に取り組んでいます。
技術選定・品質・運用の判断材料を、検証範囲と制約を示して発信しています。
AI導入や業務自動化の技術選定・設計整理については、対応できる範囲を確認のうえご相談を承ります。

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?