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

概要

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

PostgreSQLの拡張機能である pgvector とSQLAlchemy + Pydanticによるベクトル検索基盤の設計と実装を、DigitalBaseでの実装知見に基づいて解説します。専用Vector DBを追加せず、既存DBの運用フローやACID特性、通常テーブルとのJOINを活用できる構成です。

1. RAGの基本構成とpgvectorの採用理由

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()

コサイン類似度検索

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上の計算式を生成しています。

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可能なベクトル検索基盤を構築できます。既存DBと検索基盤をまとめて管理できるため、運用の簡素化につながります。

実装上は、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

下記のお悩みなら、ご相談ください

PoC から商用化に進むには、次の 3 点が必要になります。統合型 AI エージェント基盤「DigitalBase」は、 これらを 1 つの製品で提供しています。お気軽にご相談ください。