pgvectorのメリットと実践構築方法
pgvectorのメリットと実践構築方法 RAG実装に専用Vector DB。PostgreSQLの拡張機能pgvectorを使えば、既存のデータベースでベクトル検索が可能に。SQLAlchemy + Pydanticで型安全な実装ができ、通常のテーブルとJOINも自由自在。インフラ追加なし、コスト削減、運用シンプル化を実現する実践的な構築方法を解説します。
1. RAG のフレームワーク一覧
RAG（Retrieval-Augmented Generation）は、LLM に外部知識を与えて回答精度を上げる手法。基本フローは以下の通り。
ドキュメント → チャンク分割 → Embedding → ベクトルDB に格納
↓
ユーザー質問 → Embedding → 類似検索 → コンテキスト付きで LLM に渡す
主要フレームワーク比較
フレームワーク 特徴 向いているケース LangChain 最も広く使われる。チェーン構造でパイプラインを組める。プラグインが豊富 複雑な RAG パイプライン、エージェント構築 LlamaIndex データ接続に特化。100以上のデータソースコネクタ 多様なデータソースの統合
ご紹介 弊社では下記プロダクトを取り扱いしております。ご興味ありましたらご覧ください。
Haystack Deepset 製。パイプラインベースの設計。本番運用向き エンタープライズ RAG
自前実装 フレームワーク依存なし。pgvector + SQLAlchemy で直接構築 シンプルな RAG、既存 PostgreSQL の活用
なぜ自前実装 + 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 ;
対応する距離関数 演算子 距離関数 用途 <->L2（ユークリッド）距離 一般的な距離計算 <=>コサイン距離 テキスト Embedding の類似度（最も一般的） <#>負の内積 正規化済みベクトルで高速な類似度計算 <+>L1（マンハッタン）距離 特殊な用途
インデックスの種類 インデックス 特徴 推奨場面 HNSW 高速なクエリ、高い再現率。メモリ使用量大 本番環境のデフォルト IVFFlat 高速なビルド、低メモリ。学習ステップが必要 大量データの初期インデックス なし（exact） 完全な再現率。全行スキャン 数千行以下の小規模データ
最新バージョン（v0.8.1）
Postgres 13+ 対応
HNSW の iterative scan（WHERE フィルタ併用時の改善）
halfvec（半精度）、sparsevec（スパース）対応
ベクトルは最大 16,000 次元まで
3. pgvector のメリット
専用 Vector DB が不要 比較項目 pgvector Pinecone / Weaviate / Milvus インフラ 既存の PostgreSQL に追加 別サービスの管理が必要 コスト 無料（OSS） 有料 or セルフホスト ACID PostgreSQL の ACID 準拠 製品によって異なる JOIN 通常のテーブルと JOIN 可能 不可（別DBなので） フィルタ WHERE 句でそのまま絞り込み メタデータフィルタ（制約あり） バックアップ pg_dump で一括 別途バックアップ戦略が必要
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 ;
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 ( )
# 手動でベクトル文字列を構築
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
)
# 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 = 完全一致）。
値 挙動 評価 0.3〜0.4 ほぼ無関係なチャンクも拾う 低すぎる 0.45〜0.55 ローカル Embedding モデル向き（embeddinggemma, mxbai-embed-large） 推奨 0.55〜0.65 OpenAI ada-002 等の高品質モデル向き モデル次第 0.7+ 非常に厳格。関連チャンクも落とすリスク 高すぎる
注意: スコア分布は 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（取得件数） 値 用途 3〜5 リランカーなし。LLM のコンテキストを圧迫しない 10〜20 リランカーあり。大量に取得して上位を選別 20+ ノイズが増えて LLM の精度が下がる
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 ) ;
パラメータ デフォルト 推奨 説明 m16 16 レイヤーあたりの最大接続数。大きいほど再現率が上がるがメモリ増加 ef_construction64 128 グラフ構築時の候補リストサイズ。RAG では品質重視で高めに設定
-- メモリを十分に確保してからビルド
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 なら正規化済み
正規化済みベクトルではコサイン距離と内積の結果は同一順序になるが、内積は正規化ステップを省略するため計算が速い。
設定まとめ 設定 推奨値 備考 similarity_threshold0.45〜0.55 ローカル Embedding モデル使用時 top_k5 リランカーなしの場合 HNSW m 16 デフォルトで十分 HNSW ef_construction 128 デフォルト(64)より高めで品質確保 hnsw.ef_search100 デフォルト(40)より高めで再現率確保 hnsw.iterative_scanstrict_orderWHERE フィルタ併用時に必須 距離関数 cosine (<=>) 正規化済みなら inner product (<#>)
参考リンク 