RAGの取込経路を1つにまとめる設計｜ファイル・URL・会話・SQL結果を同じ書き込み口に通す理由

概要

社内RAGを運用していると、取込元は少しずつ増えていきます。最初はファイルのアップロードだけだったものが、共有フォルダの一括取込、定期実行のパイプライン、社内ポータルのWebページ、チャットでのやり取り、SQLの集計結果へと広がっていきます。

入口が増えるたびに取込処理をそれぞれに書くと、ある時点から「同じ文書が2件ヒットする」「更新したはずの古い内容が回答に出てくる」「一部の入口だけ、失敗したときに前の文書が消えてしまう」といった不整合が起き始めます。原因は、入口ごとに 重複の判定・古い文書の置き換え・失敗時の扱い が少しずつ違っていることです。

取込元が何であっても、ナレッジへの書き込みは1つの関数を通す設計の考え方を解説します。入口が持つ責務と書き込み口が持つ責務の分け方、取込元ごとの文書化の要点、単一の入口を守り続けるための工夫を順に整理します。

---

入口が増えると何が起きるか

取込処理は、どの入口でもおおむね次の手順をたどります。

1. 元データを読む（ファイルのバイト列、Webページ、会話の履歴、クエリの結果）
2. テキストを取り出す
3. チャンクに分割し、埋め込みを計算して保存する
4. 文書の台帳に1行を作る
5. 同じ取込元の古い文書があれば消す

このうち1と2は取込元によって異なりますが、3〜5は本来どの入口でも同じであるべき処理です。ところが、入口ごとに3〜5をコピーして書くと、次のような差が生まれます。

どれも1つひとつは小さな差ですが、RAGの回答品質と信頼性に直接影響します。利用者からは「AIが古い手順を答えた」としか見えないため、原因の特定にも時間がかかります。

---

入口と書き込みを分ける

この問題への対策は、入口は受付だけを持ち、書き込みは1つの関数に集める ことです。

[ファイル画面] [アップロード] [パイプライン] [URL] [チャット] [SQL結果]
      │            │             │          │       │         │
      └── 受付：権限の確認・元データの読み出し・本文の作成 ──┘
                              │
                              ▼
               書き込み口（1つの関数）
          ├ 同じ中身の判定（変わっていなければ何もしない）
          ├ 抽出（ファイルの場合のみ）
          ├ 分割・埋め込み・保存
          ├ 文書台帳の行を作る
          └ 同じ取込元の古い文書を置き換える

入口に残るのは、「誰が・どこから・何を」取り込もうとしているかの確認と、元データを読んで本文を用意する処理だけです。書き込み口は、受け取った本文が「ファイルのバイト列」か「すでにテキストになった文字列」かのどちらかであることだけを知っていれば十分です。

インターフェースのイメージは次のとおりです（説明用の汎用コードです）。

@dataclass
class StoredDocument:
    status: str          # "ingested" | "partial" | "unchanged"
    document_id: str
    chunks: int = 0
    chunks_failed: int = 0
    replaced: int = 0    # 置き換えた古い文書の数

async def put_document(
    *,
    knowledge_id: str,
    filename: str,
    content: bytes | None = None,   # ファイルならバイト列
    text: str | None = None,        # 会話・SQL結果なら文字列
    key: str | None = None,         # 取込元のキー（パス、URL、chat:<id> など）
    replace: bool = True,
    skip_unchanged: bool = True,
    metadata: dict | None = None,   # source = "web" / "chat" / "sql" など
) -> StoredDocument:
    ...

入口ごとの違いは引数の違いとして表れるだけで、処理の流れは1本になります。

---

書き込み口に集める4つの責務

1. 重複の判定

同じ内容を何度も取り込むと、埋め込みの計算に時間とGPUを使ううえ、検索結果に同じ断片が並びます。書き込み口では、本文のハッシュ（SHA-256など）を計算し、同じ取込元に同じハッシュの文書がすでにあれば何もせずに「変化なし」を返します。

ここで注意したいのは、ハッシュだけでは同一と判断できない ことです。同じ本文でも、埋め込みモデル・チャンクの大きさと重なり・画像読み取りに使うモデルが違えば、保存されるベクトルは別物になります。重複の判定には、本文のハッシュとあわせてこれらの取込条件も比べます。また、埋め込みが一部失敗した不完全な文書は判定の対象から外し、次の取込で作り直せるようにします。

2. 取込元のキーと置き換え

「同じ文書の新しいバージョン」を判定するには、取込元を表すキーが必要です。取込元ごとに、次のようにキーを決めます。

置き換えの順序も重要です。新しい文書を別のIDで最後まで書いてから、古い文書を消す 順にしておけば、途中で失敗しても古い文書は残ります。埋め込みが一部だけ失敗した場合は、新しい文書を「不完全」として記録し、前の完全な文書は残します。消すのは前の不完全な文書だけです。

これを入口ごとに実装すると、どこか1か所で「先に消してから書く」実装が紛れ込みがちです。書き込み口に集めることで、順序の約束が1か所で守られます。

3. 権限

権限の確認は、書き込み口ではなく入口の責務です。ただし、どの入口でも確認すべき点は共通しています。

• ナレッジへの書き込み権限：取り込む先のナレッジに書けるか
• 取込元の読み取り権限：ファイルなら、そのフォルダを読めるか。チャットなら、本人の会話か。SQLなら、その接続とテーブルを読めるか

SQL結果を取り込む入口では、SQL画面と同じ経路・同じ権限でクエリを読み取り専用で実行し直し、監査ログにも取込用の経路として記録します。「RAGに入れる」という操作が、権限や監査の抜け道にならないようにするためです。

Webページを取り込む入口では、取得先の制限が要ります。クラウドのメタデータのアドレスのような宛先は常に拒否し、一方で閉域の社内ポータルを取り込めるよう、社内のアドレスは許可するといった方針を、外部APIの呼び出しと同じ判定で揃えます。

4. 再取込と同期

フォルダから取り込んだナレッジは、元のファイルが更新・削除されるとずれていきます。差分を取り込み直すには、取り込んだときのファイルの状態（サイズ・更新時刻・ETagなど）を記録しておき、次回の一覧と比べます。

差分の確認（中身は読まない）
  ├ 文書がないファイル        → 追加
  ├ 記録と状態が違うファイル  → 変更（取り込み直す）
  ├ 一覧から消えたファイル    → 削除（文書を消す）
  └ 接続できず確認できない   → 何もしない（消さない）

最後の「確認できないときは消さない」が重要です。一時的にファイルサーバーへ接続できなかっただけで、ナレッジの文書がまとめて消えるのは避けなければなりません。中身が同じなら取り込みを飛ばす重複判定と組み合わせると、「変更」と判定されたファイルでも実際の埋め込み計算は最小限で済みます。

---

取込元ごとの文書化の要点

書き込み口に渡す前の「本文の作り方」は、取込元ごとに工夫が要ります。

Webページ

HTMLをそのままテキスト化すると、ナビゲーションやフッターの文字が本文に混ざり、検索の精度が下がります。ページの枠（ナビゲーション・ヘッダー・フッター・非表示要素）を除き、見出し・箇条書き・データの表・画像の説明文（alt）は構造を残して取り出します。見出しを #、箇条書きを - 、表を | a | b | の形で残すと、チャンクに分けた後も文脈が保たれます。

チャット

チャットのやり取りは、要約せずにそのまま1会話を1文書として取り込むのが扱いやすい方法です。利用者とAIの発言を順に並べ、システムの指示や空の発言は除きます。要約すると取込時のLLMの判断が入り、後から元の発言を確かめられなくなるためです。

SQLの結果

表の行は「列名: 値」の形のテキストにし、どのテーブル・どのSQLから作ったかを冒頭に書き添えます。行数の上限で切れた場合はその旨も書きます。SQLの結果はその時点のスナップショットなので、自動では取り込み直さず、同じ取込元をもう一度取り込んだときに置き換える運用とします。最新の数字が必要な質問には、RAGではなくSQLで直接答えさせます。

---

単一の入口を守り続ける工夫

設計を1つにまとめても、機能を追加するうちに、下位の「分割・埋め込み・保存」の関数を直接呼ぶコードが紛れ込むことがあります。これを防ぐには、下位の関数を呼んでよいファイルをテストで限定する のが確実です。

# 説明用の汎用コード。下位の保存関数を呼んでよいモジュールを列挙する
ALLOWED = {"knowledge/store.py", "knowledge/writer.py"}

def test_documents_are_written_only_through_put_document():
    callers = find_callers(r"\bembed_and_store\(")
    stray = sorted(callers - ALLOWED)
    assert not stray, f"put_document() を通してください: {stray}"

DigitalBaseでは、ファイルのアップロード・フォルダの一括取込・パイプラインの取込ステップ・URL・チャット・SQL結果のいずれも、この形で1つの書き込み関数を通しています。同じ中身の判定・文書台帳の行・古い文書の置き換えをこの関数に集め、下位の関数の直接呼び出しは上記のようなテストで止めています。

---

まとめ

RAGの取込元が増えても品質を保つための要点は、入口を増やしても書き込み口は増やさない ことです。

1. 入口と書き込みを分ける：入口は権限の確認と本文の作成だけを持ち、書き込みは1つの関数に集める
2. 重複の判定：本文のハッシュと取込条件（埋め込みモデル・チャンク設定）をあわせて比べる
3. キーと置き換え：取込元ごとにキーを決め、新しい文書を書き終えてから古い文書を消す
4. 権限と監査：取込元の読み取り権限を元の画面と同じ経路で確かめ、取込も監査の対象にする
5. 再取込と同期：取込時の状態を記録して差分を取り、確認できないときは消さない
6. 単一入口の維持：下位の関数を直接呼べる場所をテストで限定する

取込元を増やすたびに「どの入口でも同じ振る舞いになるか」を確かめる手間がなくなり、新しい取込元の追加は「本文をどう作るか」に集中できるようになります。

---

関連記事

• 社内データに答えるAIのSQLとRAGの使い分け｜集計は表から、文書は検索から、両方が要る質問の組み立て方
• RAGのベクトルインデックス選定｜HNSWとIVFFlatの比較と本番運用ガイド