pgvectorを本番環境で使う: PostgreSQLでの高スケールベクトル検索とハイブリッドRRF

[The Dual-Database Trap]
  User Action ──► Write to Primary Database (PostgreSQL)
                        │
                        ▼ (CDC Pipeline / Debezium / Celery Worker)
                  Dual-Write Lag & Potential Desync!
                        ▼
                  Write to Vector DB (Pinecone / Qdrant)

[The Unified pgvector Architecture]
  User Action ──► Write to PostgreSQL (Atomic Transaction)
                        ├── Relational Metadata (user_id, tenant_id, created_at)
                        └── High-Dimensional Vector (embedding vector(1536))
                  ✓ Zero Synchronization Lag
                  ✓ ACID Guarantees & Atomic Rollbacks
                  ✓ Unified Row-Level Security (RLS)

-- Enable the vector extension
CREATE EXTENSION IF NOT EXISTS vector;

-- Create production knowledge base table
CREATE TABLE document_chunks (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    document_id UUID NOT NULL,
    tenant_id UUID NOT NULL,
    content TEXT NOT NULL,
    metadata JSONB DEFAULT '{}'::jsonb,
    embedding vector(1536) NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Production HNSW Index Configuration
CREATE INDEX idx_document_chunks_hnsw_cosine 
ON document_chunks 
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- Tune per-session or per-query search depth
SET hnsw.ef_search = 100; -- Default is 40; higher values yield higher recall

-- Hybrid Search using Reciprocal Rank Fusion in pure PostgreSQL
WITH semantic_search AS (
    SELECT id, content, RANK() OVER (ORDER BY embedding <=> '[0.012, -0.045, ...]') AS rank
    FROM document_chunks
    WHERE tenant_id = 'a1b2c3d4-e5f6-7890-abcd-1234567890ab'
    ORDER BY embedding <=> '[0.012, -0.045, ...]'
    LIMIT 20
),
keyword_search AS (
    SELECT id, content, RANK() OVER (ORDER BY ts_rank_cd(to_tsvector('english', content), query) DESC) AS rank
    FROM document_chunks, plainto_tsquery('english', 'database connection pooling') query
    WHERE tenant_id = 'a1b2c3d4-e5f6-7890-abcd-1234567890ab'
      AND to_tsvector('english', content) @@ query
    LIMIT 20
)
SELECT 
    COALESCE(s.id, k.id) AS id,
    COALESCE(s.content, k.content) AS content,
    COALESCE(1.0 / (60 + s.rank), 0.0) + 
    COALESCE(1.0 / (60 + k.rank), 0.0) AS rrf_score
FROM semantic_search s
FULL OUTER JOIN keyword_search k ON s.id = k.id
ORDER BY rrf_score DESC
LIMIT 10;

# db_vector_client.py
import os
from sqlalchemy import create_engine, select, text
from sqlalchemy.orm import declarative_base, Session
from pgvector.sqlalchemy import Vector

Base = declarative_base()

class DocumentChunk(Base):
    __tablename__ = 'document_chunks'
    
    id = Column(UUID, primary_key=True)
    tenant_id = Column(UUID, nullable=False)
    content = Column(Text, nullable=False)
    embedding = Column(Vector(1536), nullable=False)

def query_similar_chunks(tenant_id: str, query_embedding: list[float], limit: int = 5):
    engine = create_engine(os.getenv("DATABASE_URL"))
    with Session(engine) as session:
        # Cosine distance ordering using native pgvector operator
        stmt = (
            select(DocumentChunk)
            .filter(DocumentChunk.tenant_id == tenant_id)
            .order_by(DocumentChunk.embedding.cosine_distance(query_embedding))
            .limit(limit)
        )
        return session.scalars(stmt).all()