社内文書Q&Aを本番で動かす:権限・更新・評価・運用の設計

試作と本番の違い

文書に質問できるシステムは、自分1人が試す形でした。これを複数人が使う社内システムにすると、モデル以外の問題が増えます。本番では、先に次の要件を決めておきます。

この記事は、特定の製品を勧めるものではありません。製品ごとの機能と設定は変わるので、考え方と設計の勘所を中心に書きます。コードは、仕組みが分かる最小の形です。

全体のアーキテクチャ

本番の構成は、「質問に答える経路」と「文書を取り込む経路」の2本に分けて考えます。2つをつなぐのが、検索用の索引です。

• 質問の経路:認証で誰かを確定し、質問サービスが検索して、権限で絞り、モデルに答えさせ、答えを検証して返す
• 取り込みの経路:文書置き場の変更を検知して、変換、分割、埋め込みをして、索引を更新する
• ログと評価:どちらの経路も、記録を残し、評価セットで定期的に測る

取り込み:文書を最新に保つ

プロトタイプでは、文書を起動のたびに全部読み込みました。本番では、文書が変わったときだけ、その文書の分だけを作り直します。

• 文書の内容のハッシュを保存しておき、変わったものだけを再処理する
• 文書が消えたら、索引からも消す(消し忘れると、削除したはずの内容で答える)
• 文書ごとに、出典(パスやURL)、更新日、閲覧できるグループを一緒に持つ
• PDFやOfficeファイルは、先にテキストに変換する。表や図は崩れやすいので、変換結果を目で確かめる
• 分割は、見出しの構造を保つ。見出しをチャンクの先頭に付けると、検索で見つけやすくなる

import hashlib
import re
import sqlite3

from rag import split  # 前の記事の分割関数

db = sqlite3.connect("index.db")
db.executescript("""
CREATE TABLE IF NOT EXISTS docs (path TEXT PRIMARY KEY, hash TEXT, groups TEXT);
CREATE TABLE IF NOT EXISTS chunks (id INTEGER PRIMARY KEY, path TEXT, ord INTEGER, text TEXT);
""")

# 取り込み: 変わった文書だけ作り直し、なくなった文書は索引から消す
def sync(files):  # files: {パス: (本文, 閲覧できるグループの集合)}
    changed = removed = 0
    for path, (text, groups) in files.items():
        digest = hashlib.sha256(text.encode()).hexdigest()
        row = db.execute("SELECT hash FROM docs WHERE path = ?", (path,)).fetchone()
        if row and row[0] == digest:
            continue  # 変わっていない
        db.execute("DELETE FROM chunks WHERE path = ?", (path,))
        for order, chunk in enumerate(split(text)):
            db.execute("INSERT INTO chunks (path, ord, text) VALUES (?, ?, ?)", (path, order, chunk))
        db.execute("INSERT OR REPLACE INTO docs VALUES (?, ?, ?)", (path, digest, ",".join(sorted(groups))))
        changed += 1  # ここで、このチャンクの埋め込みも計算し直す
    for (path,) in db.execute("SELECT path FROM docs").fetchall():
        if path not in files:
            db.execute("DELETE FROM chunks WHERE path = ?", (path,))
            db.execute("DELETE FROM docs WHERE path = ?", (path,))
            removed += 1
    db.commit()
    return changed, removed

埋め込みモデルを入れ替えると、既存のベクトルは使えなくなります。このときは、全文書のベクトルを作り直します。どのモデルで作ったベクトルかを、索引に記録しておくと、入れ替えのときに困りません。

権限は、検索の前に絞る

社内文書には、全員が見てよいものと、特定の部署だけが見てよいものが混ざります。ここで大事なのは、権限の確認を、モデルに任せないことです。「この人には見せないで」とプロンプトに書いても、モデルが従う保証はありません。

絞り込みは、検索の前(または検索と同時)に行います。見てはいけない文書は、そもそも検索の対象に入れません。

# 権限: 検索の前に、その人が見られるチャンクだけに絞る
def visible_chunks(user_groups):
    rows = db.execute("SELECT chunks.id, chunks.path, chunks.text, docs.groups FROM chunks JOIN docs USING (path)").fetchall()
    return [(i, path, text) for i, path, text, groups in rows if set(groups.split(",")) & set(user_groups)]

• 認証は、社内の既存の仕組み(SSOなど)に任せ、ユーザーのグループを取得する
• 権限が変わった(退職、異動)ときに、すぐ反映される経路を作る。索引側にグループをコピーしている場合は、同期の遅れが穴になる
• 「文書があること自体」を隠したいなら、「資料が見つかりません」という答え方も、権限で絞った結果に基づいて返す

検索の質を上げる

答えの品質は、ほとんど、検索で決まります。モデルに渡す資料が間違っていれば、どんなモデルでも正しく答えられません。

• ハイブリッド検索:単語の検索(固有名詞や型番に強い)と、ベクトルの検索(言い換えに強い)を両方使う
• 結果の統合:2つの点数は比べられないので、順位だけを使って合わせる方法(RRF)がよく使われる
• 再ランキング:上位数十件を、より精密な方法で並べ替えて、数件に絞る
• メタデータで絞る:部署、文書の種類、更新日などで、検索の範囲を狭める

# 検索結果の統合: 順位だけを使う(RRF)。単語検索とベクトル検索の点数は比べられないので、順位で合わせる
def rrf(rankings, k=60):
    score = {}
    for ranking in rankings:
        for rank, chunk_id in enumerate(ranking, start=1):
            score[chunk_id] = score.get(chunk_id, 0) + 1 / (k + rank)
    return sorted(score, key=score.get, reverse=True)

回答を作って、検証する

プロトタイプでは、モデルの答えをそのまま返しました。本番では、返す前に確認します。

• 引用の検証:答えに付いた資料の番号が、実際に渡した資料の番号か確認する。番号がなければ、根拠のない答えとして扱う
• 答えない選択:検索で何も見つからない、または関連が弱いときは、モデルを呼ばずに「分からない」と返す
• 出典の表示:答えと一緒に、元の文書へのリンクを出す。人が原文を確かめられるようにする
• プロンプトインジェクションへの備え:文書の中の指示に従わせない。外部の人が書ける文書(問い合わせ、Wikiの編集)は、とくに警戒する

# 引用の検証: 答えに付いた番号が、実際に渡した資料の番号か
def check_citations(answer, found_count):
    cited = {int(n) for n in re.findall(r"\[(\d+)\]", answer)}
    return bool(cited) and cited <= set(range(1, found_count + 1))

引用の番号が正しくても、内容が合っているとは限りません。番号の検証は最低限の確認で、内容の確認は、次の評価で行います。

評価:変えるたびに測る

分割の大きさ、検索の方法、モデル、プロンプトのどれかを変えると、品質が上がることも下がることもあります。感覚で判断せず、同じ質問のセットで測って比べます。

1. 実際の質問を、50〜100個集める(現場の人に聞くか、問い合わせの履歴を使う)
2. それぞれに、正解の根拠になる文書の部分を付ける
3. 変更のたびに、検索の成績(上位k件に正解が入った割合)を測る
4. 答えは、人が抜き取りで確認する。モデルに採点させる方法もあるが、モデルの採点も間違えるので、人の確認を併用する

# 評価: 正解のチャンクが、上位k件に入っている質問の割合
def recall_at_k(cases, search, k=5):  # cases: [(質問, 正解のチャンクIDの集合)]
    hits = sum(1 for question, expected in cases if expected & set(search(question)[:k]))
    return hits / len(cases)

• 「資料にない質問」も、評価セットに入れる。正しく「分からない」と返せるかを測る
• 質問は、実際の利用で増やす。「役に立たなかった」の記録から、評価セットに足していく
• 数字が下がったら、直前の変更を戻せるようにしておく

運用:動かし続ける

• ログの扱い:質問や答えに、機密や個人情報が入ります。ログも、文書と同じ強さで守る
• 障害時:モデルが止まっても、検索の結果(関連する文書の一覧)だけは返せるようにしておくと、完全に止まらない
• 更新:モデル、埋め込みモデル、検索の設定を変えるときは、評価セットで測ってから切り替える
• バックアップ:索引は文書から作り直せるが、時間がかかる。作り直しにかかる時間を知っておく

確認リスト

• 利用者数、文書量、更新頻度、権限の単位を決めたか
• 文書の変更と削除が、決めた時間内に索引に反映されるか
• 権限での絞り込みを、検索の前に行っているか。モデルのプロンプトに頼っていないか
• 見てはいけない文書が答えに出ないことを、テストで確かめたか
• 評価セットがあり、変更のたびに測っているか。「資料にない質問」を含んでいるか
• 答えの引用を検証し、出典のリンクを出しているか
• ログの保存期間とアクセス権を決めたか
• モデルや検索が止まったときの動作を決めたか
• 誰が運用し、問題が出たときに誰が直すか決めたか