pgvectorチュートリアル：PostgreSQLにベクトル検索を追加してRAGとセマンティック検索を実現する

RAGパイプラインやセマンティック検索が必要なアプリケーションを構築する際、最初に決める必要があるのは「Embeddingをどこに保存するか」です。PineconeやQdrant、Weaviateといった専用ベクトルデータベースも選択肢の一つですが、すでにPostgreSQLを運用しているチームにとって、pgvectorはより速く、安価で、運用が簡単な方法です。

pgvectorはPostgreSQL用のオープンソースExtensionで、vectorデータ型・類似検索演算子・HNSW/IVFFlatインデックスを既存のPostgresデータベースに追加します。EmbeddingはアプリケーションデータとともにPostgresのACID保証のもと、標準SQLで検索できます。

このチュートリアルは、インストールから本番対応のRAGクエリまでをカバーし、すべてのステップでコードを提供します。

構築するもの： PostgreSQLにDocument Embeddingを保存し、クエリに対して最も関連性の高いChunkを取得するセマンティック検索システム（RAGパイプラインのRetrieval層）

スタック：

PostgreSQL 15+（pgvector 0.7+）
Python 3.11
Embedding用にopenaiまたはollama
DB接続にpsycopg2 / asyncpg

pgvectorとは何か、なぜ使うのか

Vector Embeddingは、テキストや画像などのデータの意味的内容をエンコードした浮動小数点数のリスト（通常768〜3,072次元）です。意味が近い2つのテキストは、その高次元空間で近くに位置するベクトルを生成します。

pgvectorはこれらのベクトルをPostgresに保存し、距離計算を効率的に実行する機能を追加します：

演算	SQL
コサイン類似度	`embedding <=> query_vector`
L2距離	`embedding <-> query_vector`
内積	`embedding <#> query_vector`

pgvectorが適しているケース：

すでにPostgreSQLを運用している
ベクトル数が500万件以下
EmbeddingとリレーショナルデータをJOINしたい（ユーザーID、文書メタデータ、アクセス制御）
アプリデータとEmbeddingにまたがるACIDトランザクションが必要
追加のインフラコンポーネントを管理したくない

専用Vector DBを使うべきケース：

1,000万件超のベクトルで10ms未満のレイテンシが必要
大規模なAdvanced Filteringが必要
Postgresインフラがない

日本の企業向けRAGプロジェクト（社内文書検索、契約書検索、J-SOX準拠の監査ログ分析など）では、pgvectorで十分なケースがほとんどであり、運用が格段にシンプルです。勘定奉行や弥生などのレガシーシステムからエクスポートしたデータとEmbeddingをPostgres上で一元管理できる点も大きなメリットです。

アーキテクチャ概要

flowchart TD
    A["ドキュメント\n(PDF、テキスト、DBレコード)"] --> B["Chunking\n(約500トークンに分割)"]
    B --> C["Embeddingモデル\n(OpenAI / Ollama / ローカル)"]
    C --> D["pgvector\n(PostgreSQL)"]
    E["ユーザーのクエリ"] --> F["クエリをEmbeddingに変換"]
    F --> G["類似検索\n(コサイン / L2)"]
    D --> G
    G --> H["上位k件のChunk"]
    H --> I["LLM\n(GPT / Claude / ローカル)"]
    I --> J["引用付きの回答"]

ステップ1 — pgvectorのインストール

オプションA：Docker（開発環境に推奨）

docker run -d \
  --name pgvector-dev \
  -e POSTGRES_PASSWORD=secret \
  -e POSTGRES_DB=ragdb \
  -p 5432:5432 \
  pgvector/pgvector:pg16

このイメージにはpgvectorがプリインストールされています。

オプションB：既存PostgreSQLへのインストール

Ubuntu / Debian：

sudo apt install postgresql-16-pgvector

macOS（Homebrew）：

brew install pgvector

データベースでExtensionを有効化

CREATE EXTENSION IF NOT EXISTS vector;

-- インストール確認
SELECT * FROM pg_extension WHERE extname = 'vector';

ステップ2 — スキーマの作成

-- documentsテーブル：ソーステキストとメタデータを保存
CREATE TABLE documents (
    id          BIGSERIAL PRIMARY KEY,
    title       TEXT NOT NULL,
    source_url  TEXT,
    lang        VARCHAR(5) DEFAULT 'ja',
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

-- document_chunksテーブル：テキスト断片とEmbeddingを保存
CREATE TABLE document_chunks (
    id          BIGSERIAL PRIMARY KEY,
    document_id BIGINT REFERENCES documents(id) ON DELETE CASCADE,
    chunk_index INTEGER NOT NULL,
    content     TEXT NOT NULL,
    token_count INTEGER,
    embedding   vector(1536),  -- 使用するEmbeddingモデルの次元数に合わせる
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

モデル別Embedding次元数：

モデル	次元数
OpenAI text-embedding-3-small	1536
OpenAI text-embedding-3-large	3072
Ollama nomic-embed-text	768
Ollama mxbai-embed-large	1024
BGE-M3（多言語対応・日本語も高品質）	1024

vector(1536)のカラム型は完全に一致している必要があります。同一カラムで次元数を混在させることはできません。

ステップ3 — ドキュメントの取り込みとEmbeddingの生成

import os
import psycopg2
from openai import OpenAI
import tiktoken

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
DB_URL = "postgresql://postgres:secret@localhost:5432/ragdb"

def chunk_text(text: str, max_tokens: int = 500, overlap: int = 50) -> list[str]:
    """トークン数でテキストをオーバーラップ付きChunkに分割"""
    enc = tiktoken.get_encoding("cl100k_base")
    tokens = enc.encode(text)
    chunks = []
    start = 0
    while start < len(tokens):
        end = min(start + max_tokens, len(tokens))
        chunks.append(enc.decode(tokens[start:end]))
        start += max_tokens - overlap
    return chunks

def get_embedding(text: str) -> list[float]:
    """OpenAIからEmbeddingベクトルを取得"""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text.replace("\n", " ")
    )
    return response.data[0].embedding

def ingest_document(title: str, content: str, source_url: str = None, lang: str = "ja"):
    """ドキュメントをChunk分割・Embedding生成してPostgreSQLに保存"""
    conn = psycopg2.connect(DB_URL)
    cur = conn.cursor()
    try:
        cur.execute(
            "INSERT INTO documents (title, source_url, lang) VALUES (%s, %s, %s) RETURNING id",
            (title, source_url, lang)
        )
        doc_id = cur.fetchone()[0]
        chunks = chunk_text(content)
        print(f"'{title}'を取り込み中：{len(chunks)}チャンク")
        for i, chunk in enumerate(chunks):
            embedding = get_embedding(chunk)
            cur.execute(
                """INSERT INTO document_chunks
                   (document_id, chunk_index, content, token_count, embedding)
                   VALUES (%s, %s, %s, %s, %s)""",
                (doc_id, i, chunk, len(chunk.split()), embedding)
            )
        conn.commit()
        print(f"完了：document_id={doc_id}")
        return doc_id
    except Exception as e:
        conn.rollback()
        raise e
    finally:
        cur.close()
        conn.close()

# 使用例
ingest_document(
    title="情報セキュリティポリシー v3.2（J-SOX対応）",
    content=open("security_policy.txt").read(),
    source_url="https://internal.company.co.jp/policies/security",
    lang="ja"
)

OllamaによるローカルEmbedding（APIキー不要）

import httpx

def get_embedding_ollama(text: str, model: str = "nomic-embed-text") -> list[float]:
    """ローカルOllamaインスタンスからEmbeddingを取得"""
    response = httpx.post(
        "http://localhost:11434/api/embeddings",
        json={"model": model, "prompt": text}
    )
    return response.json()["embedding"]

get_embedding()をget_embedding_ollama()に置き換え、nomic-embed-textを使う場合はスキーマの次元数を768に変更してください。

ステップ4 — 高速類似検索のためのインデックス作成

HNSW（ほとんどのケースに推奨）

CREATE INDEX ON document_chunks
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

パラメータ	デフォルト	効果
`m`	16	ノードあたりの接続数。高いほどRecall向上・メモリ使用量増加
`ef_construction`	64	ビルド時の探索幅。高いほどRecall向上・ビルド低速化

IVFFlat（大規模データセット向け）

CREATE INDEX ON document_chunks
USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);

比較：

	HNSW	IVFFlat
Recall	高い	低い（調整可能）
ビルド時間	遅い	速い
メモリ	多い	少ない
適したケース	200万件以下	200万件以上

ステップ5 — 類似検索クエリの実行

def semantic_search(
    query: str,
    top_k: int = 5,
    lang_filter: str = None,
    min_similarity: float = 0.7
) -> list[dict]:
    """クエリに最も意味的に近いChunkを検索"""
    query_embedding = get_embedding(query)
    conn = psycopg2.connect(DB_URL)
    cur = conn.cursor()
    try:
        cur.execute("SET hnsw.ef_search = 100")
        sql = """
            SELECT dc.id, dc.content, dc.chunk_index,
                   d.title, d.source_url, d.lang,
                   1 - (dc.embedding <=> %s::vector) AS similarity
            FROM document_chunks dc
            JOIN documents d ON d.id = dc.document_id
            WHERE 1 - (dc.embedding <=> %s::vector) >= %s
            {lang_clause}
            ORDER BY dc.embedding <=> %s::vector
            LIMIT %s
        """.format(lang_clause="AND d.lang = %s" if lang_filter else "")
        params = [query_embedding, query_embedding, min_similarity]
        if lang_filter:
            params.append(lang_filter)
        params.extend([query_embedding, top_k])
        cur.execute(sql, params)
        return [
            {"id": r[0], "content": r[1], "chunk_index": r[2],
             "title": r[3], "source_url": r[4], "lang": r[5],
             "similarity": float(r[6])}
            for r in cur.fetchall()
        ]
    finally:
        cur.close()
        conn.close()

# 使用例
results = semantic_search(
    query="特権アカウントのパスワードポリシーは何ですか？",
    top_k=5,
    lang_filter="ja",
    min_similarity=0.75
)

ステップ6 — 完全なRAGパイプラインの構築

def rag_query(question: str, lang: str = "ja", top_k: int = 5) -> dict:
    """完全なRAGパイプライン：関連Chunkを取得して根拠のある回答を生成"""
    chunks = semantic_search(question, top_k=top_k, lang_filter=lang)
    if not chunks:
        return {"answer": "この質問に関連するドキュメントが見つかりませんでした。", "sources": []}

    context = "\n\n".join([f"[{i+1}] {c['content']}" for i, c in enumerate(chunks)])
    sources = [{"title": c["title"], "url": c["source_url"], "similarity": c["similarity"]}
               for c in chunks]

    system_prompt = """あなたは役立つアシスタントです。提供されたContextのみを使用して質問に答えてください。
Contextに答えが含まれていない場合は「利用可能なドキュメントにこの情報は見つかりません」と回答してください。
情報を参照する際は必ず出典番号[1]、[2]などを引用してください。
回答は日本語でお願いします。"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"Context:\n{context}\n\n質問：{question}\n\n回答："}
        ],
        temperature=0.1
    )
    return {"answer": response.choices[0].message.content, "sources": sources}

# 使用例
result = rag_query("管理者アカウントのパスワード要件は何ですか？", lang="ja")
print(result["answer"])

ステップ7 — ハイブリッド検索（ベクトル + 全文検索）

純粋なベクトル検索は、完全なキーワードマッチを見逃すことがあります。ハイブリッド検索はベクトル類似度とPostgreSQL全文検索を組み合わせ、より高いRecallを実現します：

ALTER TABLE document_chunks ADD COLUMN content_tsv tsvector
    GENERATED ALWAYS AS (to_tsvector('english', content)) STORED;

CREATE INDEX ON document_chunks USING gin(content_tsv);

日本語テキストにはpg_bigm拡張を検討してください：

CREATE EXTENSION IF NOT EXISTS pg_bigm;
-- 日本語用のインデックス
CREATE INDEX ON document_chunks USING gin(content gin_bigm_ops);

ハイブリッド検索はドメイン固有のCorpusでRecallを15〜30%改善します。製品コード、法令番号、固有名詞が含まれる文書（J-SOXの条文、APPI規制文書）に特に効果的です。

パフォーマンスチューニング

-- インデックスビルドを高速化するためのメモリ増加
SET maintenance_work_mem = '1GB';
CREATE INDEX ON document_chunks USING hnsw (embedding vector_cosine_ops);

-- Recall向上のためef_searchを高く設定（デフォルト40）
SET hnsw.ef_search = 100;

-- 言語別のPartialインデックス（日本語ドキュメント用）
CREATE INDEX ON document_chunks USING hnsw (embedding vector_cosine_ops)
WHERE document_id IN (SELECT id FROM documents WHERE lang = 'ja');

-- インデックスが使用されているか確認
EXPLAIN (ANALYZE, BUFFERS)
SELECT id, content, embedding <=> '[0.1, 0.2, ...]'::vector AS distance
FROM document_chunks ORDER BY distance LIMIT 5;

よくある質問

pgvectorは何件のベクトルを処理できますか？

実際には、標準的なハードウェア（RAM 16GB、SSD）で100〜500万件のベクトルを快適に処理できます。それを超えるとクエリのレイテンシが上昇し始めます。1,000万件以上ではQdrantやWeaviateを検討してください。

コサイン類似度とL2距離、どちらを使うべきですか？

テキストEmbeddingにはコサイン類似度（<=>）が適しています。大きさを無視してベクトル間の角度を測定します。RAGのテキスト用途では常にコサイン類似度を使用してください。

日本語と英語のEmbeddingを同じテーブルに保存できますか？

できます。ただしEmbeddingモデルが多言語対応である必要があります。BGE-M3とmultilingual-e5-largeは日本語で高品質な結果を出します。langカラムで言語を記録し、クエリ時にフィルタリングしてください。

J-SOXやAPPIへの対応でpgvectorを使う際の注意点は？

pgvectorのデータはPostgreSQLに保存されるため、通常のデータベースのセキュリティ対策がそのまま適用されます。暗号化はPostgreSQLのpgcryptoや透過的なディスク暗号化で対応できます。アクセスログはPostgreSQLの標準監査機能またはpgauditExtensionで記録でき、J-SOXのアクセス管理要件を満たします。APPIに基づく個人情報については、Embeddingに個人識別情報が含まれないようChunkingの設計で配慮してください。

チャンクサイズはいくつが適切ですか？

オーバーラップ50トークン付きの500トークンが出発点として適切です。小さすぎる（100トークン未満）とContextが失われ、大きすぎる（1,000トークン超）と関連性が薄れます。法律文書や技術仕様書は、固定トークン数ではなくセクション境界でChunkingすることをお勧めします。

本番環境チェックリスト

[ ] Embeddingカラムにオーバーラップ付きHNSWまたはIVFFlatインデックスを作成済み
[ ] インデックスビルド用にmaintenance_work_memを最低512MBに設定済み
[ ] hnsw.ef_searchを調整済み（100から始めてRecallとレイテンシをベンチマーク）
[ ] フィルタリングが多い場合、言語またはテナント別のPartialインデックスを作成済み
[ ] PostgreSQLの前にコネクションプーリング（PgBouncer）を設置済み
[ ] Embeddingの生成はキューベースの非同期処理（ユーザーリクエストとインライン化しない）
[ ] 低品質マッチをフィルタリングするmin_similarity閾値を設定済み（0.7〜0.8）
[ ] 低速なベクトルクエリに対してpg_stat_activityを監視中
[ ] バルクインジェスト後にVACUUM ANALYZEを定期的に実行

次のステップ

このチュートリアルはRetrieval層をカバーしています。完全なRAGスタックのために：

日本語ドキュメントの取り込みと多言語Embeddingの詳細 — ガイドを参照：LlamaIndex + pgvector: Production RAG for Thai and Japanese Business Documents
ローカルEmbeddingモデルの実行ハードウェア選定（OpenAI不要） — 参照：Choosing Hardware for Local LLMs in 2026
Private AI全体のアーキテクチャにおけるRAGの位置づけ — 参照：Private AI vs ChatGPT

社内文書、J-SOX準拠のワークフロー、または顧客サポート向けのRAGシステムを構築していますか？ hello@simplico.net までお問い合わせください。タイと日本の企業向けに本番RAGスタックのデプロイ支援を行っています。

関連サービス