Vector Databaseエンジニアキャリアガイド：RAG時代の必須インフラ完全比較と必要スキル

ユーザーの質問
    |
質問をベクトルに変換（Embedding Model）
    |
ベクトルDBで類似文書ベクトルを検索（ANN Search）
    |
検索された文書をコンテキストとしてLLMに渡す
    |
LLMがコンテキストに基づいて回答を生成

from openai import OpenAI

client = OpenAI()

# テキストをベクトルに変換
response = client.embeddings.create(
    model="text-embedding-3-small",
    input="Vector databases are essential for RAG applications"
)

# 結果：1536次元のfloat配列
embedding = response.data[0].embedding
print(f"Dimension: {len(embedding)}")  # 1536
print(f"Sample: {embedding[:5]}")  # [0.0123, -0.0456, 0.0789, ...]

import numpy as np

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

def euclidean_distance(a, b):
    return np.linalg.norm(a - b)

def dot_product(a, b):
    return np.dot(a, b)

# 例
v1 = np.array([1.0, 2.0, 3.0])
v2 = np.array([1.1, 2.1, 2.9])
v3 = np.array([-1.0, -2.0, -3.0])

print(f"v1-v2 cosine: {cosine_similarity(v1, v2):.4f}")   # 0.9998（非常に類似）
print(f"v1-v3 cosine: {cosine_similarity(v1, v3):.4f}")   # -1.0000（正反対）
print(f"v1-v2 L2: {euclidean_distance(v1, v2):.4f}")       # 0.1732

# HNSW主要パラメータ
hnsw_params = {
    "M": 16,              # 各ノードの最大接続数（高いほど正確、メモリ増加）
    "ef_construction": 200, # インデックス構築時の探索範囲（高いほど品質向上）
    "ef_search": 100       # 検索時の探索範囲（高いほど正確、遅い）
}

# IVF-Flat主要パラメータ
ivf_params = {
    "nlist": 1024,    # クラスタ数（通常 sqrt(N) ~ 4*sqrt(N)）
    "nprobe": 16      # 検索するクラスタ数（高いほど正確、遅い）
}

# 例：1536次元ベクトル -> PQ圧縮
# 元：1536 * 4 bytes = 6,144 bytes
# PQ(m=96, nbits=8): 96 bytes（64倍圧縮）

from pinecone import Pinecone

pc = Pinecone(api_key="your-api-key")

# インデックス作成
pc.create_index(
    name="my-index",
    dimension=1536,
    metric="cosine",
    spec={
        "serverless": {
            "cloud": "aws",
            "region": "us-east-1"
        }
    }
)

# ベクトルアップサート
index = pc.Index("my-index")
index.upsert(
    vectors=[
        ("id1", [0.1, 0.2, ...], {"text": "example document"}),
        ("id2", [0.3, 0.4, ...], {"text": "another document"})
    ],
    namespace="my-namespace"
)

# 検索
results = index.query(
    vector=[0.15, 0.25, ...],
    top_k=10,
    include_metadata=True,
    namespace="my-namespace"
)

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

client = weaviate.connect_to_local()

# コレクション作成（ハイブリッド検索サポート）
collection = client.collections.create(
    name="Document",
    vectorizer_config=Configure.Vectorizer.text2vec_openai(),
    properties=[
        Property(name="text", data_type=DataType.TEXT),
        Property(name="source", data_type=DataType.TEXT),
    ]
)

# データ挿入（自動ベクトル化）
collection.data.insert(
    properties={
        "text": "Vector databases power modern AI",
        "source": "blog"
    }
)

# ハイブリッド検索
results = collection.query.hybrid(
    query="vector search for AI",
    alpha=0.75,  # 0=BM25のみ, 1=ベクトルのみ
    limit=10
)

client.close()

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

# 接続
connections.connect("default", host="localhost", port="19530")

# スキーマ定義
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1536)
]
schema = CollectionSchema(fields)

# コレクション作成
collection = Collection("documents", schema)

# インデックス作成
index_params = {
    "metric_type": "COSINE",
    "index_type": "HNSW",
    "params": {"M": 16, "efConstruction": 256}
}
collection.create_index("embedding", index_params)

# 検索
collection.load()
results = collection.search(
    data=[[0.1, 0.2, ...]],
    anns_field="embedding",
    param={"metric_type": "COSINE", "params": {"ef": 128}},
    limit=10,
    output_fields=["text"]
)

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct

client = QdrantClient(host="localhost", port=6333)

# コレクション作成
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(size=1536, distance=Distance.COSINE)
)

# データ挿入
client.upsert(
    collection_name="documents",
    points=[
        PointStruct(
            id=1,
            vector=[0.1, 0.2, ...],
            payload={"text": "example document", "source": "blog"}
        )
    ]
)

# 検索（フィルタリング付き）
results = client.search(
    collection_name="documents",
    query_vector=[0.15, 0.25, ...],
    query_filter={
        "must": [{"key": "source", "match": {"value": "blog"}}]
    },
    limit=10
)

-- pgvector拡張インストール
CREATE EXTENSION vector;

-- ベクトルカラムを持つテーブル作成
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    text TEXT,
    embedding vector(1536)
);

-- HNSWインデックス作成
CREATE INDEX ON documents
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 200);

-- ベクトル検索（コサイン類似度）
SELECT id, text, 1 - (embedding <=> query_embedding) AS similarity
FROM documents
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector
LIMIT 10;

import chromadb

client = chromadb.Client()

# コレクション作成
collection = client.create_collection(
    name="documents",
    metadata={"hnsw:space": "cosine"}
)

# データ挿入（自動エンベディング含む）
collection.add(
    documents=["Vector databases are essential", "RAG needs vector search"],
    metadatas=[{"source": "blog"}, {"source": "paper"}],
    ids=["doc1", "doc2"]
)

# 検索
results = collection.query(
    query_texts=["AI infrastructure"],
    n_results=5,
    where={"source": "blog"}
)

# 次元削減の例
response = client.embeddings.create(
    model="text-embedding-3-small",
    input="Vector databases are essential",
    dimensions=256  # 1536 -> 256に削減（メモリ6倍節約）
)

# Weaviateハイブリッド検索の例
results = collection.query.hybrid(
    query="kubernetes pod autoscaling",
    alpha=0.75,  # 0.75 = ベクトル75% + BM25 25%
    limit=10,
    return_metadata=["score", "explain_score"]
)

# alpha値ガイド：
# alpha=1.0: ベクトル検索のみ（意味的検索）
# alpha=0.0: BM25のみ（キーワード検索）
# alpha=0.5: 均等混合
# alpha=0.7-0.8: 大多数のRAGシステムで最適

from sentence_transformers import CrossEncoder

# Cross-Encoderモデルロード
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-12-v2")

# 初期検索結果のリランキング
query = "How does HNSW algorithm work?"
passages = [doc["text"] for doc in initial_results]

# クエリ-文書ペアのスコア計算
pairs = [[query, passage] for passage in passages]
scores = reranker.predict(pairs)

# スコア基準で再ソート
reranked = sorted(zip(passages, scores), key=lambda x: x[1], reverse=True)

# データ規模別推奨インデックス戦略
indexing_strategy = {
    "small": {  # < 10万ベクトル
        "index": "HNSW",
        "params": {"M": 16, "ef_construction": 200},
        "reason": "小規模ではHNSWが最高の精度を提供"
    },
    "medium": {  # 10万 - 1000万ベクトル
        "index": "HNSW",
        "params": {"M": 32, "ef_construction": 256},
        "reason": "メモリが許せばHNSW、そうでなければIVF-Flat"
    },
    "large": {  # 1000万 - 10億ベクトル
        "index": "IVF-PQ",
        "params": {"nlist": 4096, "m": 96, "nbits": 8},
        "reason": "メモリ効率が鍵、PQ圧縮必須"
    },
    "massive": {  # > 10億ベクトル
        "index": "IVF-PQ + 分散",
        "params": {"shards": 8, "replicas": 3},
        "reason": "単一ノードでは不可能、分散クラスタ必須"
    }
}

# ベクトルDB核心モニタリング指標
monitoring_metrics = {
    "performance": {
        "query_latency_p50": "< 10ms",
        "query_latency_p95": "< 50ms",
        "query_latency_p99": "< 100ms",
        "queries_per_second": "> 1000 QPS",
        "index_build_time": "モニタリング"
    },
    "quality": {
        "recall_at_10": "> 95%",
        "recall_at_100": "> 99%",
        "embedding_drift": "定期的に測定"
    },
    "operational": {
        "memory_usage": "< 80%容量",
        "disk_usage": "< 70%容量",
        "cpu_utilization": "< 70%",
        "index_freshness": "最新データ反映遅延"
    }
}

# マルチテナンシー実装方式別比較
multi_tenancy = {
    "collection_per_tenant": {
        "isolation": "強い",
        "overhead": "高い（テナントごとにコレクション作成）",
        "use_case": "少数の大型テナント"
    },
    "namespace_per_tenant": {
        "isolation": "中程度",
        "overhead": "中程度",
        "use_case": "中規模テナント"
    },
    "metadata_filter": {
        "isolation": "弱い",
        "overhead": "低い",
        "use_case": "多数の小型テナント"
    }
}

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 推奨チャンキング戦略
splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,       # チャンクサイズ（トークン基準）
    chunk_overlap=50,     # チャンク間のオーバーラップ（文脈維持）
    separators=["\n\n", "\n", ". ", " ", ""],
    length_function=len
)

chunks = splitter.split_text(document_text)

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall
)

# RAGパイプライン評価
results = evaluate(
    dataset=eval_dataset,
    metrics=[
        faithfulness,        # 回答はコンテキストに忠実か？
        answer_relevancy,    # 回答は質問に関連しているか？
        context_precision,   # 検索されたコンテキストは正確か？
        context_recall       # 必要な情報はすべて検索されたか？
    ]
)

print(results)
# faithfulness: 0.92
# answer_relevancy: 0.89
# context_precision: 0.85
# context_recall: 0.91