pgvectorで構築するRAG基盤｜PostgreSQL拡張によるベクトル検索の設計と実装

概要

RAG（Retrieval-Augmented Generation）を社内に導入する際、ベクトル検索の実装基盤として専用のVector DBを採用するか、既存のPostgreSQLを拡張するかは、運用コストとアーキテクチャの複雑さを左右する重要な選択です。

本稿では、PostgreSQLの拡張機能である pgvector を採用し、SQLAlchemy + Pydantic で型安全なベクトル検索基盤を構築する手法を、DigitalBaseでの実装知見に基づいて整理します。専用Vector DBを追加せず、既存DBの運用フローやACID特性、通常テーブルとのJOINをそのまま活かせる点に焦点を当てます。

---

1. RAGのフレームワークと自前実装の位置づけ

RAGは、LLMに外部知識を与えて回答精度を高める手法です。基本フローは以下のとおりです。

ドキュメント → チャンク分割 → Embedding → ベクトルDB に格納
                                                    ↓
ユーザー質問 → Embedding → 類似検索 → コンテキスト付きで LLM に渡す

主要フレームワークの比較

なぜ自前実装 + pgvector を選ぶのか

フレームワークは強力ですが、依存関係が増えるほどアップデートへの追従コストが高くなります。一方、RAGのコアロジック（チャンク分割 → Embedding → 検索 → LLM）は本質的にシンプルであり、pgvector + SQLAlchemy で十分に実装可能です。すでにPostgreSQLを運用している環境であれば、専用Vector DBを別途追加する必要がなくなり、インフラと運用をシンプルに保てます。

DigitalBaseでは、社内データが既存のRDBに集約されている製造業・業務システム連携の案件で、この自前実装 + pgvector の構成を標準的な選択肢の一つとして評価しています。

---

2. pgvectorとは

pgvector は PostgreSQL の拡張機能で、ベクトル型（vector）をカラムとして扱えるようにします。これにより、PostgreSQL 上でベクトル類似度検索が可能になります。

基本概念

-- 拡張を有効化
CREATE EXTENSION vector;

-- ベクトルカラムを持つテーブル
CREATE TABLE items (
    id SERIAL PRIMARY KEY,
    content TEXT,
    embedding vector(1024)  -- 1024次元のベクトル
);

-- コサイン距離で類似検索
SELECT content, 1 - (embedding <=> query_vector) AS similarity
FROM items
ORDER BY embedding <=> query_vector
LIMIT 5;

対応する距離関数

インデックスの種類

バージョンと機能（v0.8.1）

• Postgres 13+ 対応
• HNSW の iterative scan（WHERE フィルタ併用時の精度改善）
• halfvec（半精度）、sparsevec（スパース）対応
• ベクトルは最大 16,000 次元まで

---

3. pgvectorのメリット

専用Vector DBが不要

Pythonパッケージで完結

# これだけで SQLAlchemy 統合が使える
uv add pgvector

PostgreSQLサーバー側の拡張は、Dockerイメージ（pgvector/pgvector:pg17）で解決できます。apt install は不要で、docker compose up だけで環境が立ち上がります。

既存データとの統合

-- ベクトル検索結果をユーザーテーブルと JOIN できる
SELECT e.content, e.metadata, u.name
FROM pgvector.embeddings e
JOIN "User" u ON e.user_id = u.id
WHERE e.bot_id = 'bot-123'
ORDER BY e.embedding <=> query_vector
LIMIT 5;

ベクトル検索の結果を業務テーブルと直接JOINできる点は、データが別DBに分離される専用Vector DBでは実現できない利点です。

---

4. SQLAlchemy + Pydanticによる型安全な実装

ここでは、生SQLではなくSQLAlchemy ORMとPydanticを用いることで、ベクトル操作を型安全に扱う実装パターンを示します。

依存パッケージ

# pyproject.toml
dependencies = [
    "sqlalchemy>=2.0.44",
    "psycopg2-binary>=2.9.11",
    "pgvector>=0.4.2",
    "pydantic>=2.0.0",
]

SQLAlchemyモデル定義

from sqlalchemy import (
    Column, Integer, String, Text, DateTime, Index,
    func, text, delete, literal,
)
from sqlalchemy.dialects.postgresql import JSONB
from pgvector.sqlalchemy import Vector
from common.database import Base, SessionLocal

class EmbeddingModel(Base):
    """pgvector embeddings テーブル"""
    __tablename__ = "embeddings"
    __table_args__ = (
        Index("idx_bot_user", "bot_id", "user_id"),
        Index("idx_document", "document_id"),
        {"schema": "pgvector"},
    )

    id = Column(Integer, primary_key=True, autoincrement=True)
    bot_id = Column(String(255), nullable=False)
    user_id = Column(String(255), nullable=False)
    document_id = Column(String(255), nullable=False)
    chunk_id = Column(Integer, nullable=False)
    content = Column(Text, nullable=False)
    embedding = Column(Vector())  # 次元数は動的（モデルに依存）
    doc_metadata = Column("metadata", JSONB)
    created_at = Column(DateTime, server_default=text("NOW()"))

設計上のポイント:

• Vector() は次元数を指定しないことで、異なる Embedding モデル（768d, 1024d など）に対応できます。
• doc_metadata = Column("metadata", JSONB) — SQLAlchemy の予約語 metadata を避けて属性名をマッピングしています。
• スキーマを pgvector に分離し、アプリケーションテーブルと混在させない構成としています。

Pydanticスキーマ

from pydantic import BaseModel, ConfigDict
from typing import Dict, Any

class SimilarChunk(BaseModel):
    """類似チャンク検索結果"""
    model_config = ConfigDict(from_attributes=True)

    text: str
    metadata: Dict[str, Any] | None = None
    similarity: float
    document_id: str

class DocumentInfo(BaseModel):
    """ドキュメント情報"""
    model_config = ConfigDict(from_attributes=True)

    id: str
    filename: str
    embedding_model: str
    chunks: int
    uploaded_at: str | None = None

設計上のポイント:

• ConfigDict(from_attributes=True) により、SQLAlchemy の ORM オブジェクトから直接変換できます（Pydantic v2）。
• APIレスポンスの型安全性を保証します。

Embeddingの保存

def store_embeddings_pgvector(
    bot_id: str,
    user_id: str,
    document_id: str,
    chunks: list[str],
    embeddings: list[list[float]],
    metadata: dict,
) -> int:
    db = SessionLocal()
    try:
        count = 0
        for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
            if not embedding:
                continue

            record = EmbeddingModel(
                bot_id=bot_id,
                user_id=user_id,
                document_id=document_id,
                chunk_id=i,
                content=chunk,
                embedding=embedding,       # list[float] をそのまま渡せる
                doc_metadata=metadata,      # dict をそのまま JSONB に
            )
            db.add(record)
            count += 1

        db.commit()
        return count
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

生SQL（asyncpg）では、ベクトル文字列やJSONを手動で構築する必要がありました。

Before（生 SQL + asyncpg）:

# 手動でベクトル文字列を構築
vec_str = "[" + ",".join(map(str, embedding)) + "]"
metadata_json = json.dumps(metadata)

await conn.execute(
    "INSERT INTO pgvector.embeddings ... VALUES ($1, ..., $6::vector, $7::jsonb)",
    bot_id, ..., vec_str, metadata_json
)

After（SQLAlchemy ORM）:

# Python のリストと dict をそのまま渡すだけ
record = EmbeddingModel(embedding=embedding, doc_metadata=metadata)
db.add(record)

ORMを介することで、ベクトルやJSONの直列化が自動化され、SQLインジェクションのリスクも抑えられます。

コサイン類似度検索

def retrieve_similar_chunks(
    query_embedding: list[float],
    bot_id: str,
    user_id: str,
    top_k: int = 5,
    similarity_threshold: float = 0.50,
) -> list[dict]:
    db = SessionLocal()
    try:
        # コサイン距離を計算（0 = 同一、2 = 正反対）
        distance = EmbeddingModel.embedding.cosine_distance(query_embedding)
        # 類似度に変換（1 - distance）
        similarity = (literal(1) - distance).label("similarity")

        rows = (
            db.query(
                EmbeddingModel.document_id,
                EmbeddingModel.content,
                EmbeddingModel.doc_metadata,
                similarity,
            )
            .filter(
                EmbeddingModel.bot_id == bot_id,
                EmbeddingModel.user_id == user_id,
            )
            .order_by(distance)
            .limit(top_k)
            .all()
        )

        # アプリケーション側で閾値フィルタ
        return [
            {
                "text": r.content,
                "metadata": r.doc_metadata if isinstance(r.doc_metadata, dict) else {},
                "similarity": float(r.similarity),
                "document_id": r.document_id,
            }
            for r in rows
            if float(r.similarity) >= similarity_threshold
        ]
    finally:
        db.close()

設計上のポイント:

• cosine_distance() は pgvector の <=> 演算子にマッピングされます。
• 閾値フィルタはSQLのWHEREではなくPython側で適用しています。HNSWインデックスの探索効率を落とさないための判断です。
• literal(1) - distance でSQL上の計算式を生成しています。

ドキュメント一覧取得（GROUP BY + JSONB）

def get_bot_documents(bot_id: str, user_id: str) -> list[dict]:
    db = SessionLocal()
    try:
        # JSONB からフィールドを抽出
        filename_expr = EmbeddingModel.doc_metadata['filename'].astext
        model_expr = EmbeddingModel.doc_metadata['embedding_model'].astext

        rows = (
            db.query(
                EmbeddingModel.document_id,
                filename_expr.label('filename'),
                model_expr.label('embedding_model'),
                func.count().label('chunks'),
                func.min(EmbeddingModel.created_at).label('uploaded_at'),
            )
            .filter(
                EmbeddingModel.bot_id == bot_id,
                EmbeddingModel.user_id == user_id,
            )
            .group_by(
                EmbeddingModel.document_id,
                filename_expr,
                model_expr,
            )
            .order_by(func.min(EmbeddingModel.created_at).desc())
            .all()
        )

        # Pydantic で型安全に変換
        return [
            DocumentInfo(
                id=r.document_id,
                filename=r.filename or "Unknown",
                embedding_model=r.embedding_model or "unknown",
                chunks=r.chunks,
                uploaded_at=r.uploaded_at.isoformat() if r.uploaded_at else None,
            ).model_dump()
            for r in rows
        ]
    finally:
        db.close()

設計上のポイント:

• doc_metadata['filename'].astext はSQLの metadata->>'filename' に変換されます。
• Pydantic の model_dump() でレスポンス型を保証します。

削除

def delete_document_embeddings(document_id: str, bot_id: str, user_id: str) -> int:
    db = SessionLocal()
    try:
        result = db.execute(
            delete(EmbeddingModel).where(
                EmbeddingModel.document_id == document_id,
                EmbeddingModel.bot_id == bot_id,
                EmbeddingModel.user_id == user_id,
            )
        )
        count = result.rowcount
        db.commit()
        return count
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

コネクションプールによる効率化

生asyncpgでは、処理のたびにTCP接続を確立・切断していました。

# Before: 毎回接続
conn = await asyncpg.connect(DATABASE_URL)
try:
    await conn.execute(...)
finally:
    await conn.close()  # 切断

SQLAlchemy の SessionLocal は、デフォルトで pool_size=5 のコネクションプールを持ちます。

# After: プールから取得・返却（TCP ハンドシェイク不要）
db = SessionLocal()
try:
    db.add(record)
    db.commit()
finally:
    db.close()  # プールに返却（切断ではない）

これにより、2回目以降のDBアクセスで接続確立コストが不要となり、検索処理のレイテンシ低減に寄与します。

---

5. パラメータの推奨値

similarity_threshold（類似度閾値）

類似度は 1 - コサイン距離 で表されます。値が大きいほど似ており、1.0が完全一致です。

注意: スコア分布はEmbeddingモデルに依存します。embeddinggemma や mxbai-embed-large（1024d）はスコアが低めに出る傾向があるため、0.45〜0.50 が現実的です。

スコア分布の確認方法:

SELECT
    1 - (embedding <=> (SELECT embedding FROM pgvector.embeddings LIMIT 1)) as similarity
FROM pgvector.embeddings
ORDER BY similarity DESC
LIMIT 20;

top_k（取得件数）

推奨: リランカーを使わない場合は top_k=5 を基本とします。

HNSWインデックス

データ量が数千行を超えたら、HNSWインデックスを作成します。インデックスがない場合は全行スキャン（exact search）となり、データ量に比例して検索が遅くなります。

-- HNSW インデックスの作成（コサイン距離用）
CREATE INDEX idx_embeddings_hnsw
ON pgvector.embeddings
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 128);

ビルド時の注意:

-- メモリを十分に確保してからビルド
SET maintenance_work_mem = '1GB';

クエリ時パラメータ

-- 検索時の候補リストサイズ（大きいほど再現率向上、速度低下）
SET hnsw.ef_search = 100;  -- デフォルト: 40

-- WHERE フィルタ併用時は iterative scan を有効化
SET hnsw.iterative_scan = strict_order;

ef_search は top_k 以上である必要があります。top_k=5 に対してデフォルトの 40 は条件を満たしますが、RAGでは再現率が重要なため、100 に引き上げるのが安全です。

iterative_scan は、WHERE bot_id = X AND user_id = Y のようなフィルタクエリで、インデックスの候補がフィルタにより減りすぎた場合に、自動的にスキャン範囲を広げる機能です。pgvector 0.8.0 以降で利用できます。

距離関数の選び方

Embedding が正規化済み（norm ≈ 1.0）?
├── Yes → inner product (<#>) が最速
│         vector_ip_ops でインデックス作成
└── No  → cosine distance (<=>) を使用
          vector_cosine_ops でインデックス作成

正規化の確認:

SELECT vector_norm(embedding) FROM pgvector.embeddings LIMIT 5;
-- すべて ≈ 1.0 なら正規化済み

正規化済みベクトルではコサイン距離と内積の順序が一致しますが、内積は正規化ステップを省略するため計算が高速です。

設定まとめ

---

まとめ

RAGのベクトル検索に専用Vector DBは必須ではありません。すでにPostgreSQLを運用している環境であれば、pgvectorとSQLAlchemy + Pydanticの組み合わせにより、インフラを追加せず、型安全で既存テーブルとJOIN可能なベクトル検索基盤を構築できます。コスト削減と運用のシンプル化を両立できる点が、この構成の最大の利点です。

実装上は、HNSWインデックスのパラメータ、類似度閾値、ef_search といった設定を、使用するEmbeddingモデルとデータ規模に合わせて調整することが、検索品質と性能を両立させる鍵となります。

---

参考リンク

• pgvector GitHub — PostgreSQL 拡張本体（v0.8.1）
• pgvector-python — Python パッケージ（v0.4.2）
• pgvector Docker — pgvector/pgvector:pg17
• SQLAlchemy AsyncIO — 非同期が必要な場合
• Pydantic v2 ConfigDict — from_attributes=True