データパターン

本番環境のRAGパイプラインとデータ集約的なLLMアプリケーション構築のための参照パターン。

RAGチャンキング戦略

固定サイズチャンキング

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,        # tokens, not chars — measure with tiktoken
    chunk_overlap=64,      # 10-15% overlap prevents broken context
    separators=["\n\n", "\n", ". ", " ", ""],
    length_function=len,   # replace with token counter for accuracy
)

使用場面: 均一なドキュメント、シンプルな検索。高速で予測可能。

セマンティックチャンキング

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

chunker = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",  # or "standard_deviation", "interquartile"
    breakpoint_threshold_amount=95,
)

使用場面: 複合フォーマットのドキュメント。サイズよりもトピックの境界が重要な場合。

ドキュメント対応チャンキング

# For structured documents (markdown, HTML, code)
from langchain.text_splitter import MarkdownHeaderTextSplitter

headers_to_split_on = [
    ("#", "h1"), ("##", "h2"), ("###", "h3"),
]
splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
# Preserves header hierarchy as metadata on each chunk

使用場面: ドキュメント、技術マニュアル、構造が明確なコードファイル。

チャンキング決定マトリックス

ドキュメント型	戦略	チャンクサイズ	オーバーラップ
散文/記事	再帰的	512-1024トークン	10%
技術ドキュメント	マークダウン対応	自然なセクション	ヘッダーをメタデータとして
コードファイル	言語対応 (tree-sitter)	関数/クラスレベル	インポートをコンテキストとして
チャットログ	メッセージ/ターンごとに固定	1ターンまたは3-5ターン	1ターンのオーバーラップ
法務/契約書	セマンティック	可変	15% (精度が重要)

ベクトルストア選択

各ストアの使い分け

ストア	最適な用途	ホスト型?	フィルタリング	スケール
Chroma	プロトタイピング、小規模データセット (<100Kドキュメント)	ローカル	基本的なメタデータ	シングルノード
pgvector	PostgreSQL を既に使用、ACID が必要	セルフホスト	完全なSQL	中規模 (1M)
Qdrant	本番環境、リッチフィルタリング、ハイブリッド検索	両方	高度なペイロードフィルタ	大規模 (100M+)
Pinecone	マネージド、ゼロオペレーション、エンタープライズ	クラウドのみ	メタデータフィルタ	大規模
Weaviate	マルチモーダル、グラフ的なクエリ	両方	GraphQL	大規模
FAISS	オフライン/バッチ、最大速度	ローカルのみ	なし (自分で追加)	非常に大規模

埋め込みモデル選択

# Small + fast (local, good for prototyping)
# ~384 dimensions, ~50M params
from sentence_transformers import SentenceTransformer
model = SentenceTransformer("all-MiniLM-L6-v2")

# Medium (cloud, balanced cost/quality)
# ~1536 dimensions
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Large (cloud, maximum quality)
# ~3072 dimensions, configurable
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-large",
    dimensions=1024,  # reduce dimensions for cost/speed trade-off
)

コスト最適化: 開発には text-embedding-3-small を、本番環境には次元を削減した text-embedding-3-large を使用します。あなたのデータでベンチマークを実施してください — 汎用的なベンチマークは領域固有のパフォーマンスを反映しません。

データバリデーション

Pydantic でのドキュメント検証

from pydantic import BaseModel, field_validator
from typing import Literal

class ChunkedDocument(BaseModel):
    content: str
    source: str
    chunk_index: int
    total_chunks: int
    metadata: dict

    @field_validator("content")
    @classmethod
    def content_not_empty(cls, v: str) -> str:
        if len(v.strip()) < 10:
            raise ValueError(f"Chunk content too short: {len(v)} chars")
        return v

    @field_validator("chunk_index")
    @classmethod
    def valid_index(cls, v: int, info) -> int:
        if "total_chunks" in info.data and v >= info.data["total_chunks"]:
            raise ValueError(f"chunk_index {v} >= total_chunks {info.data['total_chunks']}")
        return v

Pandera での DataFrame 検証

import pandera as pa
from pandera.typing import DataFrame, Series

class EmbeddingSchema(pa.DataFrameModel):
    document_id: Series[str] = pa.Field(nullable=False, unique=True)
    content: Series[str] = pa.Field(str_length={"min_value": 10})
    embedding: Series[object]  # numpy array
    source: Series[str] = pa.Field(isin=["web", "pdf", "api", "manual"])
    created_at: Series[pa.DateTime]

    class Config:
        strict = True
        coerce = True

@pa.check_types
def process_embeddings(df: DataFrame[EmbeddingSchema]) -> DataFrame[EmbeddingSchema]:
    # Pandera validates input and output automatically
    ...

スキーマ進化

バージョン化された埋め込みスキーマ

from pydantic import BaseModel
from typing import Any
from datetime import datetime

class DocumentSchemaV1(BaseModel):
    """Original schema."""
    content: str
    source: str
    embedding: list[float]

class DocumentSchemaV2(BaseModel):
    """Added metadata and chunk info."""
    content: str
    source: str
    embedding: list[float]
    metadata: dict[str, Any] = {}
    chunk_index: int = 0
    schema_version: int = 2

def migrate_v1_to_v2(doc: DocumentSchemaV1) -> DocumentSchemaV2:
    """Migration function — run as batch job, not on-read."""
    return DocumentSchemaV2(
        content=doc.content,
        source=doc.source,
        embedding=doc.embedding,
        metadata={"migrated_from": "v1", "migrated_at": datetime.now().isoformat()},
        chunk_index=0,
        schema_version=2,
    )

ルール:

1. 常に schema_version フィールドを追加する
2. 新しいフィールドはデフォルト値を持つ必要がある (後方互換性)
3. フィールドを削除しない — マイグレーションで非推奨にする
4. 埋め込みモデルを変更する場合は再埋め込みする (次元を混在させない)
5. マイグレーションはバッチジョブであり、読み取り時の操作ではない

RAG評価

検索メトリクス

def recall_at_k(retrieved_ids: list[str], relevant_ids: set[str], k: int) -> float:
    """What fraction of relevant docs appear in top-k results?"""
    top_k = set(retrieved_ids[:k])
    return len(top_k & relevant_ids) / len(relevant_ids) if relevant_ids else 0.0

def mrr(retrieved_ids: list[str], relevant_ids: set[str]) -> float:
    """Mean Reciprocal Rank — how high is the first relevant result?"""
    for i, doc_id in enumerate(retrieved_ids, 1):
        if doc_id in relevant_ids:
            return 1.0 / i
    return 0.0

def ndcg_at_k(retrieved_ids: list[str], relevance_scores: dict[str, float], k: int) -> float:
    """Normalized Discounted Cumulative Gain — accounts for graded relevance."""
    import math
    dcg = sum(
        relevance_scores.get(doc_id, 0.0) / math.log2(i + 2)
        for i, doc_id in enumerate(retrieved_ids[:k])
    )
    ideal = sorted(relevance_scores.values(), reverse=True)[:k]
    idcg = sum(score / math.log2(i + 2) for i, score in enumerate(ideal))
    return dcg / idcg if idcg > 0 else 0.0

エンドツーエンドのRAG評価

# Using ragas or similar framework
eval_dataset = [
    {
        "question": "What is the refund policy?",
        "ground_truth": "Full refund within 30 days of purchase.",
        "contexts": [...],  # retrieved chunks
        "answer": "...",     # LLM-generated answer
    },
]

# Key metrics:
# - Faithfulness: Does the answer stay faithful to retrieved context?
# - Answer relevance: Does the answer actually address the question?
# - Context precision: Are retrieved chunks relevant to the question?
# - Context recall: Did retrieval find all necessary information?

アンチパターン

1. 測定なしのチャンキング — チャンキング戦略を変更した後は、常に検索品質を評価してください
2. すべてに1つの埋め込みモデル — コード、散文、構造化データには異なる埋め込みアプローチが必要です
3. バリデーションをスキップ — ゴミを入れればゴミが出ます。埋め込む前にドキュメントを検証してください
4. 埋め込み次元を混在させる — 異なるモデルからのベクトルを同じインデックスに格納しないでください
5. 読み取り時のマイグレーション — スキーマ変更はバッチ操作であり、その場での変更ではありません
6. メタデータを無視 — リッチなメタデータは検索スペースを劇的に削減できるフィルタリングを可能にします
7. 過度なチャンキング — より多くのチャンク ≠ より良い検索。チャンク数ではなく recall@k を測定してください