1
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?

@aidocs

RAGの文書を差分更新する：ファイル変更検知とEmbedding再生成を効率化する実装

1
Last updated at Posted at 2026-11-06

RAG（Retrieval-Augmented Generation）の検証を進めていると、文書を1ファイル追加しただけなのに、すべての文書を再分割してEmbeddingし直す処理が気になってきます。文書数が少ないうちは問題ありませんが、対象ファイルが増えるほど、実行時間とEmbedding APIの呼び出し回数が増えていきます。

この記事では、ファイルの内容からハッシュ値を作成し、前回処理時との差分だけを検出してEmbeddingを再生成する方法を実装します。結論としては、文書単位の差分検知に加えて、チャンク単位のハッシュも保存しておくと、変更された部分だけを効率よく更新できます。

実装方針

処理全体の流れを図にすると次の通りです。

今回の処理では、次の情報をSQLiteに保存します。

  • ファイルパス
  • ファイル全体のハッシュ値
  • チャンクの内容
  • チャンクのハッシュ値
  • Embedding
  • 更新日時

ファイルの更新日時だけを見る方法もありますが、コピーや復元によって日時が変わる場合があります。そこで、最終的にはファイル内容からSHA-256を計算します。

処理の流れは次のとおりです。

  1. docs/ 以下のファイルを列挙する
  2. ファイルのSHA-256を計算する
  3. 前回保存したハッシュ値と比較する
  4. 変更がなければEmbedding処理をスキップする
  5. 変更があればチャンク化し、チャンク単位で再利用する
  6. 削除されたファイルのレコードを削除する

環境を準備する

今回はPythonとSQLite、Sentence Transformersを使いました。ベクトル検索部分は差し替えやすいように、Embeddingの保存までを小さなスクリプトにまとめます。

python -m venv .venv

# Windows
.venv\Scripts\activate

# macOS / Linux
source .venv/bin/activate

pip install sentence-transformers numpy

ディレクトリは次のようにしました。

project/
├── docs/
│   ├── faq.md
│   └── operation.md
└── incremental_index.py

日本語文書を扱うため、今回は多言語対応のモデルを利用します。実運用では、検索対象の文体やドメインに合わせてモデルを評価してください。

ファイルとチャンクのハッシュを計算する

まず、ファイル全体とチャンクのハッシュを計算する関数を用意します。

from hashlib import sha256
from pathlib import Path


def text_hash(text: str) -> str:
    return sha256(text.encode("utf-8")).hexdigest()


def file_hash(path: Path) -> str:
    return text_hash(path.read_text(encoding="utf-8"))


def split_text(text: str, chunk_size: int = 500) -> list[str]:
    # 実運用では、Markdownの見出しや段落を考慮して分割すると検索精度を改善しやすい
    return [
        text[i:i + chunk_size]
        for i in range(0, len(text), chunk_size)
        if text[i:i + chunk_size].strip()
    ]

単純な文字数分割は実装が分かりやすい反面、段落の途中で切れる可能性があります。まず差分更新の仕組みを検証するには十分ですが、実際のRAGでは、見出し・段落・コードブロックなどを考慮したチャンク分割を検討します。

SQLiteにインデックスを保存する

次に、文書とチャンクを保存するテーブルを作ります。チャンクの主キーとして、ファイルパスとチャンクハッシュを利用します。

import sqlite3


def open_db(db_path: str = "rag.db") -> sqlite3.Connection:
    conn = sqlite3.connect(db_path)

    conn.executescript("""
    CREATE TABLE IF NOT EXISTS documents (
        path TEXT PRIMARY KEY,
        file_hash TEXT NOT NULL,
        updated_at TEXT NOT NULL
    );

    CREATE TABLE IF NOT EXISTS chunks (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        path TEXT NOT NULL,
        chunk_hash TEXT NOT NULL,
        content TEXT NOT NULL,
        embedding TEXT NOT NULL,
        UNIQUE(path, chunk_hash)
    );

    CREATE INDEX IF NOT EXISTS idx_chunks_path
    ON chunks(path);
    """)

    return conn

Embeddingは、ここではJSON文字列として保存します。大量のベクトルを扱う場合は、専用のベクトルデータベースやSQLite用の拡張などを検討しますが、差分更新の考え方を確認する目的ではSQLiteでも十分に試せます。

変更された文書だけを更新する

更新処理の本体です。ファイルハッシュが一致している場合は、すぐに処理を終了します。変更があった場合だけ新しいチャンクを作り、既存のチャンクハッシュと比較します。

import json
from datetime import datetime, timezone
from pathlib import Path

from sentence_transformers import SentenceTransformer


DOCS_DIR = Path("docs")
MODEL_NAME = "paraphrase-multilingual-MiniLM-L12-v2"


def update_file(conn, model, path: Path) -> None:
    content = path.read_text(encoding="utf-8")
    current_file_hash = text_hash(content)

    row = conn.execute(
        "SELECT file_hash FROM documents WHERE path = ?",
        (str(path),)
    ).fetchone()

    if row and row[0] == current_file_hash:
        print(f"skip: {path}")
        return

    old_rows = conn.execute(
        "SELECT chunk_hash FROM chunks WHERE path = ?",
        (str(path),)
    ).fetchall()
    old_hashes = {row[0] for row in old_rows}

    new_chunks = split_text(content)
    new_hashes = set()

    for chunk in new_chunks:
        chunk_hash = text_hash(chunk)
        new_hashes.add(chunk_hash)

        if chunk_hash in old_hashes:
            continue

        vector = model.encode(chunk, normalize_embeddings=True)
        conn.execute(
            """
            INSERT INTO chunks(path, chunk_hash, content, embedding)
            VALUES (?, ?, ?, ?)
            """,
            (
                str(path),
                chunk_hash,
                chunk,
                json.dumps(vector.tolist(), ensure_ascii=False),
            )
        )
        print(f"embed: {path} / {chunk_hash[:8]}")

    removed_hashes = old_hashes - new_hashes
    for chunk_hash in removed_hashes:
        conn.execute(
            "DELETE FROM chunks WHERE path = ? AND chunk_hash = ?",
            (str(path), chunk_hash)
        )

    conn.execute(
        """
        INSERT INTO documents(path, file_hash, updated_at)
        VALUES (?, ?, ?)
        ON CONFLICT(path) DO UPDATE SET
            file_hash = excluded.file_hash,
            updated_at = excluded.updated_at
        """,
        (
            str(path),
            current_file_hash,
            datetime.now(timezone.utc).isoformat(),
        )
    )


def remove_deleted_files(conn, current_paths: set[str]) -> None:
    rows = conn.execute("SELECT path FROM documents").fetchall()

    for (path,) in rows:
        if path not in current_paths:
            conn.execute("DELETE FROM chunks WHERE path = ?", (path,))
            conn.execute("DELETE FROM documents WHERE path = ?", (path,))
            print(f"delete: {path}")


def main() -> None:
    conn = open_db()
    model = SentenceTransformer(MODEL_NAME)

    paths = sorted(DOCS_DIR.rglob("*.md"))
    current_paths = {str(path) for path in paths}

    for path in paths:
        update_file(conn, model, path)

    remove_deleted_files(conn, current_paths)
    conn.commit()
    conn.close()


if __name__ == "__main__":
    main()

実行します。

python incremental_index.py

初回はすべてのチャンクに対してembed:が表示されます。次にdocs/faq.mdの末尾へ1段落追加して再実行すると、追加されたチャンクだけがEmbedding対象になります。

skip: docs/operation.md
embed: docs/faq.md / 9b13f2a1

ただし、固定長のチャンク分割では、文書の先頭に文字を追加すると後続チャンクがすべてずれるため、多くのチャンクが変更扱いになります。この問題を避けるには、段落単位や見出し単位で分割する、あるいは安定したチャンクIDを付ける方法が有効です。

検索処理で利用する

保存したEmbeddingを使って、簡単なコサイン類似度検索も確認しました。

import json
import numpy as np


def search(conn, model, query: str, top_k: int = 3):
    query_vector = model.encode(query, normalize_embeddings=True)

    rows = conn.execute(
        "SELECT path, content, embedding FROM chunks"
    ).fetchall()

    results = []
    for path, content, embedding_json in rows:
        vector = np.array(json.loads(embedding_json))
        score = float(np.dot(query_vector, vector))
        results.append((score, path, content))

    results.sort(reverse=True, key=lambda item: item[0])
    return results[:top_k]

normalize_embeddings=Trueで正規化しているため、内積がコサイン類似度に相当します。検索処理そのものは差分更新とは独立しているため、後からFAISSや別のベクトルストアへ置き換えても、文書ハッシュとチャンクハッシュを管理する部分は流用できます。

実装時に注意した点

今回の検証で特に注意が必要だと感じた点は3つあります。

1つ目は、Embeddingモデルを変更したときです。文書が同じでもベクトルの意味が変わるため、モデル名やモデルのバージョンをインデックスのメタデータに保存し、モデル変更時は全件再生成する必要があります。

2つ目は、チャンク分割ルールの変更です。チャンクサイズやオーバーラップを変えた場合、古いチャンクをそのまま再利用すると検索結果が混在します。分割ルールのバージョンも保存しておくと安全です。

3つ目は、更新中の失敗です。Embedding生成の途中でプロセスが停止すると、中途半端な状態が残る可能性があります。実運用では、トランザクションを利用する、または一時テーブルへ書き込んでから切り替える構成にすると復旧しやすくなります。

また、ファイルパスは相対パス・絶対パスを混在させないことも重要です。同じファイルを別物として登録してしまうため、アプリケーション内で基準ディレクトリを固定して管理します。

まとめ

RAGの文書更新では、毎回すべての文書をEmbeddingし直すのではなく、次の情報を保存すると差分更新を実装できます。

  • ファイル内容のSHA-256
  • チャンク内容のハッシュ
  • Embeddingモデルと分割ルールのバージョン
  • 削除されたファイルやチャンクの情報

まずファイル単位で変更を検知し、変更されたファイルだけを処理します。さらにチャンク単位のハッシュを使えば、文書内の未変更部分も再利用できます。

文書数が増えたRAGでは、検索精度だけでなく、インデックス更新の速さや再現性も運用上の重要な要素になります。小さなSQLite実装で差分更新の仕組みを確認してから、必要に応じてベクトル検索基盤へ拡張するのが進めやすい方法だと感じました。

1
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
1
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?