ベクターDB完全比較2025：Pinecone、Weaviate、Chroma、pgvectorから選ぶ方法

-- 通常のDB：正確なマッチング
SELECT * FROM products WHERE id = 123;
SELECT * FROM documents WHERE title = 'AIガイド';

-- ベクターDB：意味的類似度検索
-- 「AIについて教えて」クエリ → 最も関連するドキュメント5件を返す
SELECT id, content, 1 - (embedding <=> '[0.1, 0.2, ...]') AS similarity
FROM documents
ORDER BY embedding <=> '[0.1, 0.2, ...]'
LIMIT 5;
-- 「=」演算ではなく「<=>（コサイン距離）」演算を使用

HNSW (Hierarchical Navigable Small World):
──────────────────────────────────────────
構造: 階層的グラフ（地図のズームレベルのように）
ビルド時間: O(n log n)
クエリ時間: O(log n) 近似
メモリ: 高い（グラフ全体がRAMに常駐）
精度: 高い、調整可能

メリット: 高速クエリ、高いRecall、動的挿入サポート
デメリット: メモリ使用量大、大規模データセットのビルドが遅い
適合: 100万〜1億件、高Recall要件

IVF (Inverted File Index):
───────────────────────────
構造: k-meansクラスタリング後、各クラスターを別インデックスとして管理
ビルド時間: 速い
クエリ時間: nprobeパラメータに依存
メモリ: HNSWより低い
精度: nprobe増加で向上

メリット: メモリ効率的、数十億件の処理が可能
デメリット: nprobeのチューニングが必要、精度-速度のトレードオフ
適合: 数十億件以上の超大規模データセット

from pinecone import Pinecone, ServerlessSpec

# 初期化
pc = Pinecone(api_key="your-api-key")

# インデックス作成（一度だけ）
pc.create_index(
    name="my-rag-index",
    dimension=1536,          # 使用する埋め込みモデルの次元数と一致させること
    metric="cosine",         # cosine, euclidean, dotproductから選択
    spec=ServerlessSpec(
        cloud="aws",
        region="us-east-1"
    )
)

index = pc.Index("my-rag-index")

# ベクターのアップサート（挿入/更新）
index.upsert(vectors=[
    {
        "id": "doc1",
        "values": embedding_vector,
        "metadata": {
            "text": "元のドキュメントテキスト",
            "source": "document.pdf",
            "page": 1
        }
    }
])

# メタデータフィルタリングを使った類似度検索
results = index.query(
    vector=query_embedding,
    top_k=5,
    include_metadata=True,
    filter={"source": {"$eq": "document.pdf"}}  # メタデータによる事前フィルタ
)

for match in results.matches:
    print(f"スコア: {match.score:.3f} | {match.metadata['text'][:100]}")

# 名前空間によるマルチテナンシー
index.upsert(
    vectors=[{"id": "doc1", "values": embedding, "metadata": {...}}],
    namespace="customer-123"  # 顧客ごとに分離
)

import weaviate
from weaviate.classes.config import Configure, Property, DataType, VectorDistances

# ローカルDocker接続 または Weaviate Cloud Services
client = weaviate.connect_to_local()

# スキーマ定義
client.collections.create(
    "Document",
    vectorizer_config=Configure.Vectorizer.text2vec_openai(),  # 自動ベクトル化
    vector_index_config=Configure.VectorIndex.hnsw(
        distance_metric=VectorDistances.COSINE
    ),
    properties=[
        Property(name="content", data_type=DataType.TEXT),
        Property(name="source", data_type=DataType.TEXT),
        Property(name="page", data_type=DataType.INT)
    ]
)

collection = client.collections.get("Document")

# ドキュメント挿入（vectorizerが設定されていれば自動でベクトル化）
collection.data.insert({
    "content": "RAGは検索拡張生成技術です",
    "source": "guide.pdf",
    "page": 1
})

# セマンティック検索
results = collection.query.near_text(
    query="検索拡張生成の仕組み",
    limit=5
)

# ハイブリッド検索：ベクター + BM25キーワード（Weaviateの強み）
hybrid_results = collection.query.hybrid(
    query="RAG検索システム",
    alpha=0.5,  # 0.0 = 純粋BM25, 1.0 = 純粋ベクター, 0.5 = バランス
    limit=5
)

# フィルター + セマンティック検索
import weaviate.classes.query as wq
filtered = collection.query.near_text(
    query="AI技術",
    filters=wq.Filter.by_property("page").greater_than(0),
    limit=10
)

import chromadb
from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction

# インメモリ（開発/テスト用）
client = chromadb.Client()

# 永続的なローカルストレージ
client = chromadb.PersistentClient(path="./chroma_db")

# 埋め込み関数の設定（挿入時に自動で埋め込み）
embedding_fn = OpenAIEmbeddingFunction(
    api_key="your-openai-key",
    model_name="text-embedding-3-small"
)

collection = client.create_collection(
    name="my_documents",
    embedding_function=embedding_fn,
    metadata={"hnsw:space": "cosine"}
)

# ドキュメントの追加（テキストを提供するだけで自動埋め込み）
collection.add(
    documents=[
        "RAGは検索拡張生成技術です",
        "埋め込みはテキストをベクターに変換します",
        "ベクターDBは類似度検索に最適化されています"
    ],
    ids=["doc1", "doc2", "doc3"],
    metadatas=[
        {"source": "guide.pdf", "page": 1},
        {"source": "guide.pdf", "page": 2},
        {"source": "guide.pdf", "page": 3}
    ]
)

# クエリ（テキストを渡すだけでOK）
results = collection.query(
    query_texts=["テキスト検索技術について教えてください"],
    n_results=3,
    where={"source": "guide.pdf"}  # メタデータフィルター
)

print(results["documents"])
print(results["distances"])

-- 拡張機能のインストール
CREATE EXTENSION vector;

-- ベクターカラムを持つテーブル作成
CREATE TABLE documents (
    id BIGSERIAL PRIMARY KEY,
    content TEXT NOT NULL,
    source VARCHAR(255),
    page INTEGER,
    embedding vector(1536),          -- 埋め込みを格納するカラム
    created_at TIMESTAMP DEFAULT NOW()
);

-- HNSWインデックスの作成（パフォーマンスの要）
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- コサイン類似度検索
SELECT
    id,
    content,
    source,
    1 - (embedding <=> $1) AS similarity
FROM documents
ORDER BY embedding <=> $1
LIMIT 5;

-- 通常のPostgreSQLフィルターとの組み合わせ
SELECT id, content, similarity
FROM (
    SELECT
        id,
        content,
        1 - (embedding <=> $1) AS similarity
    FROM documents
    WHERE source = 'guide.pdf'
      AND created_at > NOW() - INTERVAL '30 days'
) sub
WHERE similarity > 0.75
ORDER BY similarity DESC
LIMIT 10;

import psycopg2
from openai import OpenAI

openai_client = OpenAI()

def index_document(content: str, source: str, page: int, conn) -> None:
    """ドキュメントを埋め込んで保存する"""
    embedding = openai_client.embeddings.create(
        input=content,
        model="text-embedding-3-small"
    ).data[0].embedding

    with conn.cursor() as cur:
        cur.execute(
            "INSERT INTO documents (content, source, page, embedding) VALUES (%s, %s, %s, %s::vector)",
            (content, source, page, str(embedding))
        )
    conn.commit()

def semantic_search(query: str, conn, top_k: int = 5) -> list:
    """意味的類似度によるドキュメント検索"""
    query_embedding = openai_client.embeddings.create(
        input=query,
        model="text-embedding-3-small"
    ).data[0].embedding

    with conn.cursor() as cur:
        cur.execute("""
            SELECT id, content, source,
                   1 - (embedding <=> %s::vector) AS similarity
            FROM documents
            ORDER BY embedding <=> %s::vector
            LIMIT %s
        """, (str(query_embedding), str(query_embedding), top_k))
        rows = cur.fetchall()

    return [
        {"id": r[0], "content": r[1], "source": r[2], "similarity": r[3]}
        for r in rows
    ]

テスト条件: 100万件 1536次元ベクター、コサイン類似度、top-5検索

クエリ処理能力（QPS）:
────────────────────────────────────────
Pinecone Serverless:  ~2,000 QPS (自動スケール)
Weaviate (HNSW):      ~1,500 QPS (単一インスタンス)
pgvector (HNSW):      ~800 QPS (インデックスがRAMに完全ロード時)
Chroma:               ~200 QPS (シングルスレッドの限界)

レイテンシ p99 (100万ベクター時):
────────────────────────────────────────
Pinecone:    ~50ms（ネットワーク込み）
Weaviate:    ~30ms（ローカル）
pgvector:    ~20ms（インデックスがメモリに常駐時）
Chroma:      ~100ms以上

from abc import ABC, abstractmethod

class VectorStoreBase(ABC):
    @abstractmethod
    def upsert(self, ids: list, embeddings: list, metadatas: list) -> None: ...

    @abstractmethod
    def search(self, query_embedding: list, top_k: int = 5) -> list: ...

    @abstractmethod
    def delete(self, ids: list) -> None: ...


class ChromaVectorStore(VectorStoreBase):
    def __init__(self, collection_name: str):
        import chromadb
        self.collection = chromadb.PersistentClient("./chroma").get_or_create_collection(collection_name)

    def upsert(self, ids, embeddings, metadatas):
        self.collection.upsert(ids=ids, embeddings=embeddings, metadatas=metadatas)

    def search(self, query_embedding, top_k=5):
        r = self.collection.query(query_embeddings=[query_embedding], n_results=top_k)
        return [{"id": i, "score": 1-d} for i, d in zip(r["ids"][0], r["distances"][0])]

    def delete(self, ids):
        self.collection.delete(ids=ids)


class PineconeVectorStore(VectorStoreBase):
    def __init__(self, index_name: str):
        from pinecone import Pinecone
        self.index = Pinecone(api_key="your-key").Index(index_name)

    def upsert(self, ids, embeddings, metadatas):
        self.index.upsert([{"id": i, "values": e, "metadata": m}
                           for i, e, m in zip(ids, embeddings, metadatas)])

    def search(self, query_embedding, top_k=5):
        r = self.index.query(vector=query_embedding, top_k=top_k)
        return [{"id": m.id, "score": m.score} for m in r.matches]

    def delete(self, ids):
        self.index.delete(ids=ids)


# ビジネスロジックはインターフェースだけを見る
def build_rag_pipeline(provider: str) -> VectorStoreBase:
    stores = {"chroma": ChromaVectorStore, "pinecone": PineconeVectorStore}
    return stores[provider]("my_collection")