LLM RAGパイプライン：チャンキング戦略とエンベディング最適化実践ガイド 2026

from langchain.text_splitter import CharacterTextSplitter

# 固定サイズチャンキング - 最も基本的な方式
splitter = CharacterTextSplitter(
    separator="\n",
    chunk_size=512,
    chunk_overlap=50,       # 10%オーバーラップで文脈を維持
    length_function=len,
)

documents = splitter.split_text(raw_text)
print(f"総チャンク数: {len(documents)}")
print(f"平均チャンク長: {sum(len(d) for d in documents) / len(documents):.0f}文字")

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 2026年ベンチマーク基準の最適設定
splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,          # 約12%オーバーラップ
    separators=["\n\n", "\n", ". ", " ", ""],
    length_function=len,
    is_separator_regex=False,
)

chunks = splitter.split_text(raw_text)

# チャンク品質の検証
for i, chunk in enumerate(chunks[:3]):
    print(f"[Chunk {i}] 長さ={len(chunk)} | 先頭: {chunk[:80]}...")

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# セマンティックチャンキング - 意味転換点に基づく分割
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
semantic_splitter = SemanticChunker(
    embeddings,
    breakpoint_threshold_type="percentile",  # percentile, standard_deviation, interquartile
    breakpoint_threshold_amount=75,          # 上位25%の類似度差で分割
)

semantic_chunks = semantic_splitter.split_text(raw_text)
print(f"セマンティックチャンク数: {len(semantic_chunks)}")
print(f"平均長さ: {sum(len(c) for c in semantic_chunks) / len(semantic_chunks):.0f}文字")

from langchain.text_splitter import MarkdownHeaderTextSplitter

# Markdown構造ベースチャンキング
headers_to_split_on = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]

md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on,
    strip_headers=False,
)

md_chunks = md_splitter.split_text(markdown_text)

# 各チャンクにメタデータとしてヘッダー階層情報を含む
for chunk in md_chunks[:3]:
    print(f"メタデータ: {chunk.metadata}")
    print(f"内容: {chunk.page_content[:100]}...")
    print("---")

from openai import OpenAI
import numpy as np

client = OpenAI()

def generate_embeddings(
    texts: list[str],
    model: str = "text-embedding-3-large",
    dimensions: int = 1024,    # 次元削減でコスト/速度を最適化
    batch_size: int = 100,
) -> np.ndarray:
    """バッチ単位のエンベディング生成（次元削減対応）"""
    all_embeddings = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        response = client.embeddings.create(
            input=batch,
            model=model,
            dimensions=dimensions,  # text-embedding-3シリーズのみサポート
        )
        batch_embs = [item.embedding for item in response.data]
        all_embeddings.extend(batch_embs)

    return np.array(all_embeddings, dtype=np.float32)


# 使用例
chunks = ["RAGパイプラインの核心は検索品質である。", "チャンキング戦略によって結果が異なる。"]
embeddings = generate_embeddings(chunks, dimensions=1024)
print(f"エンベディングshape: {embeddings.shape}")  # (2, 1024)

from FlagEmbedding import BGEM3FlagModel

# BGE-M3: dense + sparse + colbertを同時サポート
model = BGEM3FlagModel("BAAI/bge-m3", use_fp16=True)

sentences = [
    "LLM RAGパイプラインにおけるチャンキング戦略は検索品質の核心である。",
    "ベクトルデータベースのインデキシングは検索レイテンシに直接影響する。",
]

# dense + sparse + colbertエンベディングを同時生成
output = model.encode(
    sentences,
    batch_size=12,
    max_length=512,
    return_dense=True,
    return_sparse=True,
    return_colbert_vecs=True,
)

dense_embeddings = output["dense_vecs"]       # shape: (2, 1024)
sparse_embeddings = output["lexical_weights"]  # スパースベクトル（BM25代替）
colbert_vecs = output["colbert_vecs"]          # マルチベクトル（精密マッチング）

print(f"Dense shape: {dense_embeddings.shape}")
print(f"Sparse keys例: {list(sparse_embeddings[0].keys())[:5]}")

import chromadb
from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction

# Chromaクライアント初期化（永続ストレージ）
client = chromadb.PersistentClient(path="./chroma_db")

embedding_fn = OpenAIEmbeddingFunction(
    api_key="sk-...",
    model_name="text-embedding-3-large",
)

# コレクション作成（HNSWインデックス自動適用）
collection = client.get_or_create_collection(
    name="rag_knowledge_base",
    embedding_function=embedding_fn,
    metadata={
        "hnsw:space": "cosine",       # 類似度メトリック
        "hnsw:M": 32,                 # HNSW接続数（高いほど正確、メモリ増加）
        "hnsw:ef_construction": 200,  # インデックス構築時の探索幅
    },
)

# ドキュメント追加（バッチ）
collection.add(
    documents=["RAGにおけるチャンキングは検索品質の80%を決定する。", "エンベディングモデルの選択が残りの20%を左右する。"],
    metadatas=[
        {"source": "rag_guide", "section": "chunking", "date": "2026-03"},
        {"source": "rag_guide", "section": "embedding", "date": "2026-03"},
    ],
    ids=["doc_001", "doc_002"],
)

# 検索（メタデータフィルタ + 類似度）
results = collection.query(
    query_texts=["RAGパイプラインで最も重要な要素は？"],
    n_results=5,
    where={"source": "rag_guide"},
    include=["documents", "distances", "metadatas"],
)

for doc, dist, meta in zip(
    results["documents"][0], results["distances"][0], results["metadatas"][0]
):
    print(f"[距離: {dist:.4f}] {meta['section']} | {doc[:80]}")

import numpy as np
from typing import List, Set


def mean_reciprocal_rank(
    retrieved_ids: List[List[str]],
    relevant_ids: List[Set[str]],
) -> float:
    """MRR計算：各クエリにおける最初の関連文書のランク逆数平均"""
    mrr_scores = []
    for retrieved, relevant in zip(retrieved_ids, relevant_ids):
        for rank, doc_id in enumerate(retrieved, 1):
            if doc_id in relevant:
                mrr_scores.append(1.0 / rank)
                break
        else:
            mrr_scores.append(0.0)
    return np.mean(mrr_scores)


def recall_at_k(
    retrieved_ids: List[List[str]],
    relevant_ids: List[Set[str]],
    k: int = 10,
) -> float:
    """Recall@K：上位K件の結果における関連文書の割合"""
    recalls = []
    for retrieved, relevant in zip(retrieved_ids, relevant_ids):
        top_k = set(retrieved[:k])
        if len(relevant) == 0:
            continue
        recalls.append(len(top_k & relevant) / len(relevant))
    return np.mean(recalls)


def ndcg_at_k(
    retrieved_ids: List[List[str]],
    relevant_ids: List[Set[str]],
    k: int = 10,
) -> float:
    """NDCG@K：ランク加重関連性評価"""
    ndcg_scores = []
    for retrieved, relevant in zip(retrieved_ids, relevant_ids):
        # DCG計算
        dcg = 0.0
        for rank, doc_id in enumerate(retrieved[:k], 1):
            if doc_id in relevant:
                dcg += 1.0 / np.log2(rank + 1)

        # Ideal DCG計算
        ideal_hits = min(len(relevant), k)
        idcg = sum(1.0 / np.log2(r + 1) for r in range(1, ideal_hits + 1))

        ndcg_scores.append(dcg / idcg if idcg > 0 else 0.0)
    return np.mean(ndcg_scores)


# 使用例
retrieved = [["d1", "d3", "d5", "d2", "d4"]]
relevant = [{"d1", "d2", "d7"}]

print(f"MRR:       {mean_reciprocal_rank(retrieved, relevant):.4f}")
print(f"Recall@3:  {recall_at_k(retrieved, relevant, k=3):.4f}")
print(f"Recall@5:  {recall_at_k(retrieved, relevant, k=5):.4f}")
print(f"NDCG@5:    {ndcg_at_k(retrieved, relevant, k=5):.4f}")

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

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

# Dense + Sparseベクトルを同時に保存するコレクション作成
client.create_collection(
    collection_name="hybrid_rag",
    vectors_config={
        "dense": VectorParams(size=1024, distance=Distance.COSINE),
    },
    sparse_vectors_config={
        "sparse": SparseVectorParams(),
    },
)

# ドキュメントのインデキシング（dense + sparseベクトルを同時保存）
client.upsert(
    collection_name="hybrid_rag",
    points=[
        models.PointStruct(
            id=1,
            vector={
                "dense": dense_embedding.tolist(),
                "sparse": models.SparseVector(
                    indices=list(sparse_weights.keys()),
                    values=list(sparse_weights.values()),
                ),
            },
            payload={"text": "RAGパイプラインチャンキングガイド", "source": "blog"},
        ),
    ],
)

# Hybrid Search実行（RRFベースのスコア統合）
results = client.query_points(
    collection_name="hybrid_rag",
    prefetch=[
        models.Prefetch(
            query=dense_query_vector.tolist(),
            using="dense",
            limit=20,
        ),
        models.Prefetch(
            query=models.SparseVector(
                indices=list(sparse_query.keys()),
                values=list(sparse_query.values()),
            ),
            using="sparse",
            limit=20,
        ),
    ],
    query=models.FusionQuery(fusion=models.Fusion.RRF),  # Reciprocal Rank Fusion
    limit=10,
)

for point in results.points:
    print(f"[Score: {point.score:.4f}] {point.payload['text']}")