Advanced RAGパイプライン完全ガイド 2025：チャンキング戦略、リランキング、エージェンティックRAG、評価

ユーザーの質問 → [検索(Retrieval)] → [関連文書] → [LLM + 文書コンテキスト] → 回答生成

Naive RAG（2023年初）
├── 固定サイズチャンキング
├── 単一エンベディング検索
├── Top-K結果をそのままLLMに渡す
└── 問題: 検索精度が低い、コンテキストノイズ

Advanced RAG（2024年）
├── セマンティックチャンキング + メタデータ
├── ハイブリッド検索（ベクトル + キーワード）
├── リランキングで検索結果を精製
├── クエリ変換（HyDE、Multi-Query）
└── コンテキスト圧縮

Modular RAG（2025年）
├── エージェンティックRAG（動的ルーティング）
├── Self-RAG（自己反省）
├── CRAG（Corrective RAG）
├── マルチモーダルRAG
├── Knowledge Graph + RAG
└── モジュール組み合わせ型パイプライン

悪いチャンキング: 文の途中で切断 → 意味の損失 → 検索失敗 → 誤った回答
良いチャンキング: 意味単位で分離 → 豊富なコンテキスト → 正確な検索 → 正確な回答

from langchain.text_splitter import CharacterTextSplitter

splitter = CharacterTextSplitter(
    chunk_size=500,      # チャンクサイズ
    chunk_overlap=50,    # オーバーラップ（境界情報の保存）
    separator="\n\n"     # セパレータ
)

chunks = splitter.split_text(document_text)

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""]  # 優先順位
)

chunks = splitter.split_text(document_text)
# まず\n\nで分離を試み、サイズ超過なら\n、次にピリオド...

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# セマンティックチャンキング
chunker = SemanticChunker(
    embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95  # 類似度が95パーセンタイル以下なら分割
)

chunks = chunker.split_text(document_text)

文1: "ベクトルDBはAIインフラの核心だ。" → エンベディング [0.1, 0.3, ...]
文2: "HNSWは最速の検索アルゴリズムだ。" → 類似度 0.85（高い → 同じチャンク）
文3: "一方、今日の天気は晴れだ。"      → 類似度 0.15（低い → 新しいチャンク開始！）

from langchain.text_splitter import MarkdownHeaderTextSplitter

headers_to_split = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]

splitter = MarkdownHeaderTextSplitter(headers_to_split)
chunks = splitter.split_text(markdown_text)

# 各チャンクに見出しメタデータが自動的に含まれる
for chunk in chunks:
    print(f"メタデータ: {chunk.metadata}")
    # {'Header 1': 'Vector Database', 'Header 2': 'インデキシングアルゴリズム'}

from openai import OpenAI

client = OpenAI()

def agentic_chunk(text, max_chunk_size=1500):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """与えられたテキストを意味的に独立したチャンクに分割してください。
各チャンクは自己完結的に理解可能でなければなりません。
JSON配列でチャンクテキストを返してください。"""},
            {"role": "user", "content": text}
        ],
        response_format={"type": "json_object"}
    )
    return response.choices[0].message.content

一般テキスト: 500〜1000トークン（オーバーラップ10〜20%）
技術文書: 800〜1500トークン（セクション単位）
法律/医療文書: 300〜500トークン（精密度重視）
コード: 関数/クラス単位（構造ベース）
会話/QA: 対話ターン単位

コスト重視 + 汎用        → text-embedding-3-small
最高性能 + 無料          → BGE-M3（セルフホスティング）
多言語 + 検索特化        → Cohere embed-v3.0
コード検索              → voyage-code-3
長文書（8K+トークン）    → Jina-embeddings-v3
プライベートデータ + オンプレミス → BGE-M3 or Nomic

from ragatouille import RAGPretrainedModel

rag = RAGPretrainedModel.from_pretrained("colbert-ir/colbertv2.0")

# インデキシング
rag.index(
    collection=[doc1, doc2, doc3],
    index_name="my_index"
)

# 検索（トークンレベルマッチング）
results = rag.search(query="ベクトルデータベース性能比較", k=5)

元のクエリ: "RAG遅い"（曖昧、短い）
→ 変換後: "RAGパイプライン検索レイテンシ最適化方法"（具体的、検索に適切）

from openai import OpenAI

client = OpenAI()

def hyde_search(query, collection):
    # 1. LLMで仮想回答を生成
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "与えられた質問に対する詳細な回答を作成してください。正確でなくても構いません。"},
            {"role": "user", "content": query}
        ]
    )
    hypothetical_doc = response.choices[0].message.content
    
    # 2. 仮想回答をエンベディングして検索
    embedding = get_embedding(hypothetical_doc)
    results = collection.search(query_vector=embedding, limit=5)
    
    return results

def multi_query_transform(original_query):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """与えられた質問を3つの異なる視点から書き換えてください。
各質問は元の意図を維持しつつ、異なるキーワードを使用してください。
改行で区切って3つの質問のみ返してください。"""},
            {"role": "user", "content": original_query}
        ]
    )
    queries = response.choices[0].message.content.strip().split("\n")
    return [original_query] + queries

# 使用例
queries = multi_query_transform("RAGパイプライン性能最適化")
# → ["RAGパイプライン性能最適化",
#    "検索拡張生成システムの応答時間改善方法",
#    "RAGアーキテクチャでの検索精度向上戦略",
#    "LLMベース文書検索システム最適化技法"]

def step_back_prompt(query):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """与えられた質問より一段階より一般的で広い質問を生成してください。
この一般的な質問の回答が元の質問に答えるのに役立つ必要があります。"""},
            {"role": "user", "content": query}
        ]
    )
    return response.choices[0].message.content

def decompose_query(complex_query):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """複雑な質問をシンプルなサブ質問に分解してください。
各サブ質問は独立して回答可能である必要があります。
JSON配列で返してください。"""},
            {"role": "user", "content": complex_query}
        ],
        response_format={"type": "json_object"}
    )
    return response.choices[0].message.content

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import Qdrant

# BM25（キーワード）リトリーバー
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5

# ベクトルリトリーバー
vector_retriever = qdrant_vectorstore.as_retriever(
    search_kwargs={"k": 5}
)

# アンサンブル（ハイブリッド）
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # ベクトルに重み付け
)

results = ensemble_retriever.invoke("RAGパイプライン最適化方法")

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

compressor = LLMChainExtractor.from_llm(llm)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vector_retriever
)

compressed_docs = compression_retriever.invoke("HNSWアルゴリズムのMパラメータの役割")
# 文書全体ではなく、質問に関連する段落のみを返却

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 小さなチャンク: 検索用
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)

# 大きなチャンク: コンテキスト用
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)

store = InMemoryStore()

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

# 200トークンチャンクで精密検索 → 2000トークン親文書を返却
retriever.add_documents(documents)
results = retriever.invoke("HNSWパラメータチューニング")

第1段階（Bi-encoder）: クエリベクトル vs 文書ベクトル → 高速だが不正確 → Top 20
第2段階（Cross-encoder）: (クエリ, 文書)ペアを直接比較 → 遅いが正確 → Top 5

from sentence_transformers import CrossEncoder

model = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

initial_results = vector_search(query, top_k=20)

pairs = [(query, doc.content) for doc in initial_results]
scores = model.predict(pairs)

reranked = sorted(
    zip(initial_results, scores),
    key=lambda x: x[1],
    reverse=True
)[:5]

import cohere

co = cohere.Client("YOUR_API_KEY")

documents = [doc.content for doc in initial_results]

response = co.rerank(
    model="rerank-v3.5",
    query="RAGパイプライン最適化",
    documents=documents,
    top_n=5
)

for result in response.results:
    print(f"Index: {result.index}, Score: {result.relevance_score:.4f}")

従来のRAG: 質問 → 検索 → 生成（固定パイプライン）
エージェンティックRAG: 質問 → [エージェントが判断]
              ├── 検索が必要か？ → 検索 → 結果は十分か？
              │                        ├── Yes → 生成
              │                        └── No → 別のソースを検索 / クエリ修正
              └── 検索なしで回答可能 → 直接生成

def self_rag(query):
    # 1. 検索必要性の判断
    need_retrieval = judge_retrieval_need(query)
    
    if not need_retrieval:
        return generate_without_context(query)
    
    # 2. 検索実行
    documents = retrieve(query)
    
    # 3. 各文書の関連性判断
    relevant_docs = [doc for doc in documents if is_relevant(query, doc)]
    
    if not relevant_docs:
        refined_query = refine_query(query)
        documents = retrieve(refined_query)
        relevant_docs = [d for d in documents if is_relevant(refined_query, d)]
    
    # 4. 回答生成
    answer = generate_with_context(query, relevant_docs)
    
    # 5. 回答品質の自己評価
    if not is_supported(answer, relevant_docs):
        answer = regenerate(query, relevant_docs)
    
    return answer

def corrective_rag(query):
    documents = retrieve(query)
    quality = evaluate_retrieval_quality(query, documents)
    
    if quality == "CORRECT":
        refined_knowledge = refine_knowledge(documents)
        return generate(query, refined_knowledge)
    
    elif quality == "AMBIGUOUS":
        web_results = web_search(query)
        combined = documents + web_results
        refined = refine_knowledge(combined)
        return generate(query, refined)
    
    elif quality == "INCORRECT":
        web_results = web_search(query)
        refined = refine_knowledge(web_results)
        return generate(query, refined)

def adaptive_rag(query):
    complexity = classify_query(query)
    
    if complexity == "SIMPLE":
        docs = simple_retrieve(query, top_k=3)
        return generate(query, docs)
    
    elif complexity == "MODERATE":
        queries = multi_query_transform(query)
        docs = hybrid_search(queries)
        reranked = rerank(query, docs)
        return generate(query, reranked)
    
    elif complexity == "COMPLEX":
        sub_queries = decompose_query(query)
        sub_answers = []
        for sq in sub_queries:
            docs = search(sq)
            sub_answers.append(generate(sq, docs))
        return synthesize(query, sub_answers)

def query_router(query):
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": """クエリを分析して適切なデータソースを選択してください：
- VECTOR_DB: 内部文書検索が必要な場合
- WEB_SEARCH: 最新情報や外部情報が必要な場合
- SQL_DB: 構造化データクエリが必要な場合
- DIRECT: LLMが直接回答可能な場合
一単語のみ返してください。"""},
            {"role": "user", "content": query}
        ]
    )
    route = response.choices[0].message.content.strip()
    
    if route == "VECTOR_DB":
        return vector_db_search(query)
    elif route == "WEB_SEARCH":
        return web_search(query)
    elif route == "SQL_DB":
        return sql_query(query)
    else:
        return direct_answer(query)

from langchain_community.document_loaders import UnstructuredPDFLoader

loader = UnstructuredPDFLoader(
    "document.pdf",
    mode="elements",
    strategy="hi_res"
)
elements = loader.load()

for element in elements:
    if element.metadata.get("type") == "Image":
        description = describe_image_with_vision(element)
        element.page_content = description

def process_table(table_element):
    """表を検索可能な形式に変換"""
    md_table = table_element.metadata.get("text_as_html", "")
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "次の表の内容を自然言語で要約してください。"},
            {"role": "user", "content": md_table}
        ]
    )
    summary = response.choices[0].message.content
    
    return {
        "content": summary,
        "metadata": {"type": "table", "original_html": md_table}
    }

通常のRAG: 文書チャンクを独立的に検索
GraphRAG: エンティティの関係を活用して接続された情報まで検索

例: "HNSWを使用するベクトルデータベースは？"
通常のRAG: HNSWに関するチャンクのみ返却
GraphRAG: HNSW → (使用される) → Qdrant, Weaviate, Milvus
         + 各DBのHNSW設定関連チャンクまで返却

from langchain_community.graphs import Neo4jGraph
from langchain.chains import GraphCypherQAChain

graph = Neo4jGraph(
    url="bolt://localhost:7687",
    username="neo4j",
    password="password"
)

chain = GraphCypherQAChain.from_llm(
    llm=llm,
    graph=graph,
    verbose=True
)

result = chain.invoke("Qdrantがサポートするインデキシングアルゴリズムは？")

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

eval_data = {
    "question": ["RAGとは何か？", "HNSWアルゴリズムの原理は？"],
    "answer": ["RAGは検索拡張生成で...", "HNSWは階層的グラフ..."],
    "contexts": [
        ["RAG（Retrieval-Augmented Generation）は...", "検索ベースの生成技術..."],
        ["HNSWはHierarchical Navigable...", "グラフベースのANNアルゴリズム..."]
    ],
    "ground_truth": ["RAGはLLMが外部知識を...", "HNSWは多層グラフ構造の..."]
}

dataset = Dataset.from_dict(eval_data)

results = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall]
)

print(results)
# {'faithfulness': 0.89, 'answer_relevancy': 0.92,
#  'context_precision': 0.85, 'context_recall': 0.78}

import hashlib
import json

class RAGCache:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.ttl = 3600  # 1時間
    
    def get_cached_response(self, query):
        key = hashlib.md5(query.encode()).hexdigest()
        cached = self.redis.get(f"rag:{key}")
        if cached:
            return json.loads(cached)
        return None

# セマンティックキャッシング: 類似クエリもキャッシュヒット
class SemanticCache:
    def __init__(self, vectorstore, threshold=0.95):
        self.store = vectorstore
        self.threshold = threshold
    
    def get(self, query):
        results = self.store.similarity_search_with_score(query, k=1)
        if results and results[0][1] >= self.threshold:
            return results[0][0].metadata["response"]
        return None

from openai import OpenAI

client = OpenAI()

def stream_rag_response(query, context):
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": f"コンテキスト:\n{context}"},
            {"role": "user", "content": query}
        ],
        stream=True
    )
    
    for chunk in stream:
        if chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content

def rag_with_fallback(query):
    try:
        docs = vector_search(query)
        
        if not docs or max_relevance_score(docs) < 0.5:
            docs = web_search_fallback(query)
        
        if not docs:
            return direct_llm_answer(query)
        
        return generate_with_context(query, docs)
    
    except Exception as e:
        return {
            "answer": "申し訳ございません。一時的なエラーが発生しました。",
            "error": str(e),
            "fallback": True
        }

monitoring = {
    "検索レイテンシ": "目標 p99 200ms",
    "生成レイテンシ": "目標 p99 3s（ストリーミングfirst-token 500ms）",
    "検索適合度（Relevance）": "自動評価 0.85+",
    "回答根拠性（Groundedness）": "自動評価 0.90+",
    "キャッシュヒット率": "目標 30%+",
    "ユーザーフィードバック": "肯定 80%+",
    "トークン使用量": "コスト追跡",
    "エラー率": "目標 1%以下",
}