1. RAG のフレームワーク一覧

---

RAG（Retrieval-Augmented Generation）は、LLM に外部知識を与えて回答精度を上げる手法。基本フローは以下の通り。

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

主要フレームワーク比較

なぜ自前実装 + pgvector か

フレームワークは便利だが、依存が増えてアップデートに追従するコストが高い。RAG のコアロジック（チャンク → Embedding → 検索 → LLM）は実はシンプルで、pgvector + SQLAlchemy で十分に実装できる。既に PostgreSQL を使っているなら、専用の Vector DB を追加する必要がなくなる。

---

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;

これは専用 Vector DB では不可能。

---

4. Python pgvector + SQLAlchemy + 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()

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)

コサイン類似度検索

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 なら正規化済み

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

設定まとめ

---

参考リンク

• 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