RAGパイプライン高度化戦略：チャンキング、リランキング、ハイブリッド検索最適化

[インデクシングフェーズ]
文書収集 → 前処理 → チャンキング → エンベディング → ベクトルDB保存 + 転置インデックス保存

[クエリフェーズ]
質問 → クエリ変換 → ハイブリッド検索 (Dense + Sparse)
     → リランキング → 上位K件選択 → プロンプト構成 → LLM生成

from langchain_text_splitters import CharacterTextSplitter

# 固定サイズチャンキング - 最も基本的な方式
fixed_splitter = CharacterTextSplitter(
    separator="\n",
    chunk_size=512,
    chunk_overlap=50,       # 10%オーバーラップ推奨
    length_function=len,
    is_separator_regex=False,
)

chunks = fixed_splitter.split_text(document_text)
print(f"合計 {len(chunks)}個のチャンク作成")

from langchain_text_splitters import RecursiveCharacterTextSplitter

# 再帰的チャンキング - プロダクション基本推奨
recursive_splitter = RecursiveCharacterTextSplitter(
    separators=["\n\n", "\n", ". ", " ", ""],
    chunk_size=512,
    chunk_overlap=64,
    length_function=len,
    add_start_index=True,   # 原文位置追跡用
)

chunks = recursive_splitter.split_documents(documents)

# 各チャンクにメタデータが自動的に含まれる
for chunk in chunks[:3]:
    print(f"サイズ: {len(chunk.page_content)}, "
          f"開始位置: {chunk.metadata.get('start_index')}")

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# セマンティックチャンキング - 意味単位での分割
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

semantic_splitter = SemanticChunker(
    embeddings=embeddings,
    breakpoint_threshold_type="percentile",  # percentile, standard_deviation, interquartile
    breakpoint_threshold_amount=75,          # 75パーセンタイル以上の差で分割
)

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

from openai import OpenAI

client = OpenAI()

def agentic_chunk(text: str, max_chunks: int = 20) -> list[dict]:
    """LLMベースのエージェンティックチャンキング"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "与えられたテキストを論理的単位に分割してください。"
                    "各チャンクは1つの完結したトピックを扱う必要があります。"
                    "JSON配列で返してください: "
                    '[{"title": "チャンクタイトル", "content": "チャンク内容", "summary": "一行要約"}]'
                ),
            },
            {"role": "user", "content": text[:8000]},  # トークン制限に注意
        ],
        response_format={"type": "json_object"},
        temperature=0,
    )
    import json
    result = json.loads(response.choices[0].message.content)
    return result.get("chunks", [])

# 使用例
chunks = agentic_chunk(long_document_text)
for chunk in chunks:
    print(f"[{chunk['title']}] {chunk['summary']}")

from langchain_openai import OpenAIEmbeddings

# OpenAIエンベディング - 次元削減の活用
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-large",
    dimensions=1024,  # 3072 -> 1024に削減してコスト/ストレージを節約
)

# BGE-M3: Dense + Sparse同時生成（ハイブリッド検索に最適）
from FlagEmbedding import BGEM3FlagModel

bge_model = BGEM3FlagModel("BAAI/bge-m3", use_fp16=True)

# DenseとSparseベクトルを同時生成
output = bge_model.encode(
    ["RAGパイプライン最適化方法論"],
    return_dense=True,
    return_sparse=True,
)

dense_vector = output["dense_vecs"][0]    # (1024,) floatベクトル
sparse_vector = output["lexical_weights"][0]  # スパースベクトル（単語別重み）

print(f"Dense次元: {len(dense_vector)}")
print(f"Sparseアクティブトークン数: {len(sparse_vector)}")

質問: "Kubernetes HPAの設定でCPU閾値は？"

Dense検索（ベクトル類似度）:
  → "コンテナオーケストレーションでの自動スケーリング設定方法" (意味的に類似)

Sparse検索（BM25キーワードマッチング）:
  → "HPAのtargetCPUUtilizationPercentageを80に設定" (キーワード正確マッチング)

ハイブリッド結合（RRFまたは加重和）:
  → 両方の結果を統合して最適な文書を返却

import weaviate, { WeaviateClient } from 'weaviate-client'

// Weaviateクライアント接続
const client: WeaviateClient = await weaviate.connectToLocal({
  host: 'localhost',
  port: 8080,
})

// ハイブリッド検索の実行
const collection = client.collections.get('Documents')

const result = await collection.query.hybrid('Kubernetes HPA CPU 閾値', {
  alpha: 0.7, // 0 = 純粋BM25, 1 = 純粋ベクトル, 0.7 = ベクトル70%
  limit: 20, // リランキング前の候補数
  fusionType: 'RelativeScore', // RelativeScoreまたはRanked
  returnMetadata: ['score', 'explainScore'],
  returnProperties: ['title', 'content', 'source'],
})

for (const item of result.objects) {
  console.log(`[${item.metadata?.score?.toFixed(3)}] ${item.properties.title}`)
}

from rank_bm25 import BM25Okapi
import numpy as np
from typing import List, Tuple

def hybrid_search(
    query: str,
    documents: list[dict],
    dense_scores: np.ndarray,
    k: int = 10,
    alpha: float = 0.7,
    rrf_k: int = 60,
) -> list[dict]:
    """
    RRFベースのハイブリッド検索
    alpha: Dense検索の重み（1-alphaがSparseの重み）
    rrf_k: RRF定数（デフォルト60、論文推奨）
    """
    # Sparse検索（BM25）
    tokenized_docs = [doc["content"].split() for doc in documents]
    bm25 = BM25Okapi(tokenized_docs)
    sparse_scores = bm25.get_scores(query.split())

    # Denseランク計算
    dense_ranks = np.argsort(-dense_scores) + 1  # 1-indexedランク
    sparse_ranks = np.argsort(-sparse_scores) + 1

    # RRFスコア計算
    rrf_scores = []
    for i in range(len(documents)):
        dense_rrf = alpha / (rrf_k + dense_ranks[i])
        sparse_rrf = (1 - alpha) / (rrf_k + sparse_ranks[i])
        rrf_scores.append(dense_rrf + sparse_rrf)

    # 上位K件返却
    top_indices = np.argsort(rrf_scores)[::-1][:k]
    return [
        {**documents[i], "hybrid_score": rrf_scores[i]}
        for i in top_indices
    ]

# 使用例
results = hybrid_search(
    query="Kubernetes HPA CPU threshold",
    documents=all_documents,
    dense_scores=embedding_similarity_scores,
    k=20,       # リランキング前なので余裕を持って
    alpha=0.7,  # Dense 70%, Sparse 30%
)

import cohere

co = cohere.ClientV2(api_key="your-cohere-api-key")

def rerank_with_cohere(
    query: str,
    documents: list[str],
    top_n: int = 5,
) -> list[dict]:
    """Cohere Rerank 4で文書を再順位付け"""
    response = co.rerank(
        model="rerank-v3.5",
        query=query,
        documents=documents,
        top_n=top_n,
        return_documents=True,
    )

    results = []
    for item in response.results:
        results.append({
            "index": item.index,
            "score": item.relevance_score,
            "text": item.document.text if item.document else documents[item.index],
        })
    return results

# 使用例：ハイブリッド検索結果20件をリランキングして上位5件を選択
hybrid_results = hybrid_search(query, documents, dense_scores, k=20)
reranked = rerank_with_cohere(
    query="Kubernetes HPA CPUの閾値設定方法",
    documents=[r["content"] for r in hybrid_results],
    top_n=5,
)

for r in reranked:
    print(f"[{r['score']:.4f}] {r['text'][:80]}...")

from FlagEmbedding import FlagReranker

# BGE Reranker v2 M3 - 多言語対応の軽量リランカー
reranker = FlagReranker(
    "BAAI/bge-reranker-v2-m3",
    use_fp16=True,  # GPUメモリ節約
)

def rerank_with_bge(
    query: str,
    documents: list[str],
    top_n: int = 5,
) -> list[dict]:
    """BGE Rerankerで文書を再順位付け"""
    # クエリ-文書ペアの生成
    pairs = [[query, doc] for doc in documents]

    # 関連度スコアの計算
    scores = reranker.compute_score(pairs, normalize=True)

    # スコア順にソート
    scored_docs = [
        {"index": i, "score": s, "text": d}
        for i, (s, d) in enumerate(zip(scores, documents))
    ]
    scored_docs.sort(key=lambda x: x["score"], reverse=True)

    return scored_docs[:top_n]

# 使用例
results = rerank_with_bge(
    query="RAGパイプラインでチャンキングサイズを決める基準は？",
    documents=candidate_documents,
    top_n=5,
)

for r in results:
    print(f"[{r['score']:.4f}] {r['text'][:100]}...")

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

class OptimizedRAGPipeline:
    """高度化されたRAGパイプライン"""

    def __init__(self):
        self.splitter = RecursiveCharacterTextSplitter(
            chunk_size=512, chunk_overlap=64
        )
        self.embeddings = OpenAIEmbeddings(
            model="text-embedding-3-large", dimensions=1024
        )
        self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
        self.reranker = FlagReranker("BAAI/bge-reranker-v2-m3", use_fp16=True)

    def query(self, question: str, top_k: int = 5) -> str:
        # 1. ハイブリッド検索：候補20件
        candidates = self._hybrid_search(question, k=20)

        # 2. リランキング：上位5件を選択
        reranked = self._rerank(question, candidates, top_n=top_k)

        # 3. プロンプト構成とLLM生成
        context = "\n\n---\n\n".join([doc["text"] for doc in reranked])
        prompt = ChatPromptTemplate.from_messages([
            ("system", "以下のコンテキストに基づいて質問に回答してください。\n\n{context}"),
            ("human", "{question}"),
        ])
        chain = prompt | self.llm
        response = chain.invoke({"context": context, "question": question})
        return response.content

    def _hybrid_search(self, query: str, k: int = 20) -> list[dict]:
        # Dense + Sparseハイブリッド検索（上記コード参照）
        ...

    def _rerank(self, query: str, docs: list[dict], top_n: int) -> list[dict]:
        pairs = [[query, doc["text"]] for doc in docs]
        scores = self.reranker.compute_score(pairs, normalize=True)
        for doc, score in zip(docs, scores):
            doc["rerank_score"] = score
        docs.sort(key=lambda x: x["rerank_score"], reverse=True)
        return docs[:top_n]

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

# 評価データセットの構成
eval_data = {
    "question": [
        "Kubernetes HPAのCPU閾値のデフォルト値は？",
        "RAGでチャンキングサイズを決める基準は？",
    ],
    "answer": [
        "HPAのCPU閾値のデフォルト値は80%です。",
        "クエリタイプに応じて256-1024トークンの範囲で決定します。",
    ],
    "contexts": [
        ["HPAはtargetCPUUtilizationPercentageのデフォルト値80を使用します。"],
        ["ファクトイドクエリは256-512、分析型は1024+トークンを推奨します。"],
    ],
    "ground_truth": [
        "デフォルト値は80%である。",
        "クエリタイプと文書特性に応じて決定する。",
    ],
}

dataset = Dataset.from_dict(eval_data)

# 評価の実行
results = evaluate(
    dataset=dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)

print(results)
# faithfulness: 0.92, answer_relevancy: 0.88,
# context_precision: 0.85, context_recall: 0.90

from deepeval import assert_test
from deepeval.test_case import LLMTestCase
from deepeval.metrics import (
    FaithfulnessMetric,
    ContextualRelevancyMetric,
    AnswerRelevancyMetric,
)

def test_rag_faithfulness():
    """RAG応答がコンテキストに忠実かテスト"""
    test_case = LLMTestCase(
        input="Kubernetes HPA CPUの閾値は？",
        actual_output="HPAのCPU閾値のデフォルト値は80%です。",
        retrieval_context=[
            "HPAはtargetCPUUtilizationPercentageのデフォルト値80を使用します。"
        ],
    )

    faithfulness = FaithfulnessMetric(threshold=0.8)
    relevancy = ContextualRelevancyMetric(threshold=0.7)
    answer_rel = AnswerRelevancyMetric(threshold=0.7)

    assert_test(test_case, [faithfulness, relevancy, answer_rel])

# pytestで実行: pytest test_rag.py -v

# Blue-Greenインデックス切り替え例
import time

def reindex_with_blue_green(
    old_collection: str,
    new_collection: str,
    new_embedding_model: str,
):
    """無停止再インデクシング"""
    # 1. 新コレクションにインデクシング（既存サービスはold_collectionを使用中）
    print(f"新コレクション '{new_collection}' のインデクシング開始...")
    create_and_populate_collection(new_collection, new_embedding_model)

    # 2. 検証：新コレクションに対してテストクエリを実行
    test_results = run_evaluation_suite(new_collection)
    if test_results["context_precision"] < 0.8:
        raise ValueError(
            f"新インデックスの品質が基準未満: {test_results['context_precision']:.2f}"
        )

    # 3. トラフィック切り替え：aliasまたは設定変更
    update_active_collection(new_collection)
    print(f"トラフィック切り替え完了: {old_collection} -> {new_collection}")

    # 4. 旧コレクションはロールバック用に一定期間保持
    schedule_cleanup(old_collection, delay_days=7)

import tiktoken

enc = tiktoken.encoding_for_model("text-embedding-3-small")

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    length_function=lambda text: len(enc.encode(text)),  # トークン数基準
)

import asyncio

async def rerank_with_fallback(
    query: str,
    documents: list[str],
    top_n: int = 5,
    timeout: float = 2.0,
) -> list[dict]:
    """フォールバック付きリランキング"""
    try:
        # Primary: Cohere Rerank（タイムアウト2秒）
        result = await asyncio.wait_for(
            cohere_rerank_async(query, documents, top_n),
            timeout=timeout,
        )
        return result
    except (asyncio.TimeoutError, Exception) as e:
        print(f"Cohereリランキング失敗、BGEフォールバック: {e}")
        try:
            # Fallback: セルフホスティングBGE Reranker
            return rerank_with_bge(query, documents, top_n)
        except Exception as e2:
            print(f"BGEリランキングも失敗、元の順位を返却: {e2}")
            # 最終フォールバック：ハイブリッド検索結果をそのまま返却
            return [
                {"index": i, "score": 1.0 - i * 0.05, "text": d}
                for i, d in enumerate(documents[:top_n])
            ]