社内向けのRAGを作るとき、ベクトルの置き場所を決める必要があります。専用のベクトルDBを契約するか、すでに動いているPostgreSQLに pgvector を入れて済ませるか。後者を選べる条件は思ったより広く、そして日本語でDjangoからこれを扱った実装記事はほとんどありません。

この記事では、モデル定義からインデックス設計、そしていちばん厄介な「再インデックスの運用」までを順に書きます。前回の権限フィルタの記事と同じ構成を前提にしています。

そもそも自前で持つべきか

先に判断基準を出します。マネージドのベクトルDBが向いている場面は確かにあるので、無理に自前にする必要はありません。

実務でいちばん効くのは最後の行です。ベクトル検索の結果を、そのまま権限テーブルや業務テーブルと結合できるのは、同じDBに置いているからこそできることです。別サービスに分けると、検索結果のIDを持ち帰ってから照合する処理を自分で書くことになり、そこが必ずバグります。

導入は拡張1つ

-- PostgreSQL 側
CREATE EXTENSION IF NOT EXISTS vector;

Djangoからは pgvector-python を入れます。マイグレーションで拡張を有効化しておくと、環境を作り直したときに手作業が要りません。

# migrations/0002_vector.py
from pgvector.django import VectorExtension

class Migration(migrations.Migration):
    operations = [VectorExtension()]

次元数は、あとから変えられない

モデル定義で最初に決めるのがベクトルの次元数です。これは使う埋め込みモデルによって決まります。

# models.py
from pgvector.django import VectorField, HnswIndex

class Chunk(models.Model):
    document   = models.ForeignKey(Document, on_delete=models.CASCADE)
    ordinal    = models.IntegerField()              # 文書内の何番目か
    body       = models.TextField()
    body_hash  = models.CharField(max_length=64)  # 再計算の要否判定に使う
    embedding  = VectorField(dimensions=384)       # multilingual-e5-small

dimensions=384 は multilingual-e5-small の出力次元です。ここを変えるということは、全チャンクを作り直すということです。カラム定義だけ変えても既存のデータは合いません。

だからモデル選定は、性能の比較というより後戻りのコストを見積もる作業だと考えたほうが正確です。

e5系を使うなら、プレフィックスを忘れない

実装で最も静かに効くのがこれです。e5系のモデルは、質問側と文書側で別の接頭辞を付ける前提で学習されています。

# 正しい
q = model.encode("query: " + question)
d = model.encode("passage: " + chunk_body)

# 付け忘れ → エラーにはならない。静かに精度が落ちるだけ
q = model.encode(question)

付け忘れても例外は出ません。検索は動き、それらしい結果も返ります。「なんとなく精度が悪い」の原因がこれだった、というのは十分あり得ます。

本文とベクトルは分けるか、同じ行に持つか

小規模なら同じテーブルで構いません。ただしモデルを差し替える予定があるなら、最初から分けたほうが楽です。

分ける場合は Embedding(chunk, model_name, vector) のような形にして、model_name でどの世代のベクトルかを持たせます。後述の入れ替え作業が、これがあるかないかで難易度が変わります。

インデックスは HNSW か IVFFlat か

pgvectorには2種類の近似インデックスがあります。迷ったらHNSWでよいのですが、性質が違うので押さえておきます。

# models.py の Meta
class Meta:
    indexes = [
        HnswIndex(
            name="chunk_emb_hnsw",
            fields=["embedding"],
            m=16,                 # 1ノードあたりの接続数
            ef_construction=64,   # 構築時の探索幅。大きいほど精度が上がり遅くなる
            opclasses=["vector_cosine_ops"],
        ),
    ]

構築が遅いと感じたら、セッション単位で作業メモリを上げると変わります。

SET maintenance_work_mem = '2GB';
SET max_parallel_maintenance_workers = 4;

インデックスがメモリに乗り切らないと構築が急激に遅くなります。件数が増えたときに「昨日までと同じ作業なのに終わらない」と感じたら、まずここを疑ってください。

コサインか、内積か

埋め込みを正規化してあるなら、コサイン距離と内積は順位が一致します。内積のほうが計算が軽いので、正規化を保証できるなら vector_ip_ops を選ぶ余地があります。

ただし「正規化しているつもり」が最も危険です。モデルを差し替えたときに正規化の有無が変わることがあります。確信が持てないうちはコサインにしておくほうが安全です。

検索はDjangoのORMでそのまま書ける

# services/search.py
from pgvector.django import CosineDistance

def search(question, user, k=5):
    vec = model.encode("query: " + question)
    return (
        Chunk.objects
        .filter(allowed_groups__overlap=user.group_ids)  # 権限は先に絞る
        .annotate(dist=CosineDistance("embedding", vec))
        .order_by("dist")[:k]
    )

権限の絞り込みを annotate() より前に書く理由は前回の記事に書きました。あとから落とすと件数が消えるうえ、文書の存在自体が漏れます。

類似度のしきい値を、固定値で決めない

「距離が0.3以下なら採用」のような固定しきい値は、たいてい機能しません。質問の長さや語彙によって、距離の絶対値がまるごとずれるからです。

私たちが社内の検索で採っているのはギャップ方式です。上位から順に並べたとき、隣り合う結果の距離の差が一定以上（実測では 0.010）開いた場所を切れ目とみなす。絶対値ではなく落差を見ます。

# 上位から見ていき、落差の大きいところで打ち切る
GAP = 0.010
def cut(results):
    for i in range(1, len(results)):
        if results[i].dist - results[i-1].dist > GAP:
            return results[:i]
    return results

これなら「よく似たものが3件しかない質問」と「10件ある質問」を、同じ設定で扱えます。

ここからが本題：モデルを変えるときの入れ替え

RAGの運用でいちばん重いのは、最初の構築ではありません。埋め込みモデルを差し替える日です。

これは必ず来ます。より精度の高い日本語モデルが出る、コストの安いものに寄せたい、次元数を落としてストレージを減らしたい。理由はいくらでもあります。そしてモデルを変えるということは、全チャンクを計算し直すということです。数万件でも数時間、数十万件なら日単位の作業になります。

止めずにやるなら、並存させて切り替える

素直にカラムを上書きすると、変換中は新旧のベクトルが混ざった状態になります。この状態の検索結果は意味がありません。距離の尺度が違うものを一列に並べているからです。

手順はこうなります。

1. 新しい置き場所を作る（embedding_v2 カラム、または model_name 付きの別テーブルの新しい行）
2. 裏で埋めていく。バッチで少しずつ。この間、検索は旧ベクトルで動き続ける
3. 埋め終わったら新しい側にインデックスを張る
4. 参照先を切り替える。設定値ひとつで切り替わるようにしておく
5. 問題がなければ旧側を落とす。問題が出たら設定を戻すだけで復旧できる

文書が更新されたとき、どこを計算し直すか

全件やり直すのは高くつくので、変わったチャンクだけを再計算したくなります。ここで body_hash が効きます。

# 取り込み時
h = hashlib.sha256(body.encode("utf-8")).hexdigest()
if chunk.body_hash != h:
    chunk.body      = body
    chunk.body_hash = h
    chunk.embedding = model.encode("passage: " + body)
    chunk.save()
# 変わっていなければ何もしない

注意点がひとつあります。チャンクの切り方を変えると、本文が同じでも全チャンクのハッシュが変わります。分割ロジックを触ったときは、部分更新ではなく全件再計算になると考えてください。

モデル選定でボツにしたもの

参考までに、実際に試して採用しなかった例を書いておきます。

軽量な蒸留モデルは、一語のクエリに弱いという傾向がありました。速度とサイズは魅力的でしたが、「請求書」「有給」のような単語ひとつで投げられる質問——社内検索ではこれが非常に多い——で、期待した文書が上位に来ませんでした。

文章として成立している質問なら悪くない。ただ社内検索の実際の入力は、文章より単語のほうが多い。ベンチマークの数字ではなく、自社の実際のクエリで比べないと判断を誤ります。

最終的に multilingual-e5-small に落ち着きました。384次元でストレージも軽く、日本語の単語クエリでも崩れませんでした。

まとめ

• チャンクが数十万までなら pgvector で足りる。専用DBを増やす前に検討する価値がある
• いちばんの利点は速度ではなく、権限や業務データと同じDBで結合できること
• 次元数は後から変えられない。モデル選定は後戻りコストの見積もり
• e5系は query: / passage: の付け忘れが静かに効く
• インデックスは迷ったらHNSW。IVFFlatは空のテーブルに作らない
• しきい値は固定値ではなくギャップ（落差）で切る
• モデル差し替えは並存させて切り替える。切り戻せる形が実質のバックアップ
• 部分更新は body_hash で。ただし分割ロジックを変えたら全件やり直し