RAG完全ガイド2026:RAGパイプラインの設計・実装・評価まで【実践編】
検索拡張生成(RAG: Retrieval-Augmented Generation)は、LLMの知識制限を克服する最も実践的な手法です。本記事では、2026年現在の最新RAGアーキテクチャから実装、評価ツールまで徹底解説します。
なぜRAGが必要か?
LLMが抱える3つの根本的な限界:
| 限界 | 症状 | RAGの解決策 |
|---|---|---|
| 知識のカットオフ | 最新情報に答えられない | リアルタイム検索で補完 |
| ハルシネーション | 自信満々に間違いを言う | 根拠文書を参照して生成 |
| コンテキスト不足 | 社内ドキュメントを知らない | プライベートデータを注入 |
RAGの進化:Naive RAG → Advanced RAG → Agentic RAG
Naive RAG(2023年以前)
質問 → 埋め込み検索 → チャンク取得 → LLM生成
シンプルだが精度に限界。
Advanced RAG(2024年主流)
質問 → クエリ変換 → ハイブリッド検索 → リランキング → LLM生成
HyDE、クエリ分解、MMRなどの技術で精度向上。
Agentic RAG(2025〜2026年最新)
質問 → エージェントが戦略を立案
→ 複数データソースを並列検索
→ 結果を統合・検証
→ 必要に応じて再検索
→ 最終回答生成
エージェントが動的に検索戦略を決定。複雑な質問に対応。
実装:LangChainでRAGパイプラインを構築
Step 1:文書の読み込みとチャンキング
from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
# PDFの読み込み
loader = PyPDFLoader("technical_manual.pdf")
documents = loader.load()
# Webページの読み込み
web_loader = WebBaseLoader(["https://docs.example.com"])
web_docs = web_loader.load()
# チャンキング
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 1チャンク = 1000文字
chunk_overlap=200, # 前後のチャンクと200文字重複(コンテキスト保持)
separators=["\n\n", "\n", "。", ".", " ", ""]
)
all_docs = documents + web_docs
chunks = splitter.split_documents(all_docs)
print(f"チャンク数: {len(chunks)}")
Step 2:ベクトルストアへの格納
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
import chromadb
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
# Chromaに格納(永続化あり)
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db",
collection_name="knowledge_base"
)
print(f"格納完了: {vectorstore._collection.count()} チャンク")
Step 3:Advanced検索(ハイブリッド + リランク)
from langchain.retrievers import EnsembleRetriever, ContextualCompressionRetriever
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
# ベクトル検索(意味的類似性)
vector_retriever = vectorstore.as_retriever(
search_type="mmr", # 多様性を確保(Maximal Marginal Relevance)
search_kwargs={"k": 10, "fetch_k": 30}
)
# BM25検索(キーワードマッチ)
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 10
# ハイブリッド検索(ベクトル60% + BM25 40%)
hybrid_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.6, 0.4]
)
# クロスエンコーダーでリランキング
reranker_model = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-v2-m3")
reranker = CrossEncoderReranker(model=reranker_model, top_n=5)
final_retriever = ContextualCompressionRetriever(
base_compressor=reranker,
base_retriever=hybrid_retriever
)
Step 4:RAGチェーンの構築
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_template("""
あなたは丁寧で正確な技術アシスタントです。
以下の参考文書のみを使って質問に答えてください。
文書に答えがない場合は「この質問に答えるための情報が文書中に見つかりませんでした」と正直に伝えてください。
参考文書:
{context}
質問: {question}
回答(日本語で、根拠を示しながら):
""")
def format_docs(docs):
return "\n\n---\n\n".join([
f"[出典: {doc.metadata.get('source', '不明')}]\n{doc.page_content}"
for doc in docs
])
rag_chain = (
{"context": final_retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
# 実行
response = rag_chain.invoke("RAGのハルシネーション対策について教えてください")
print(response)
Agentic RAGの実装(LangGraph版)
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage
from typing import TypedDict, List, Optional
class RAGState(TypedDict):
question: str
search_queries: List[str]
retrieved_docs: List[str]
answer: Optional[str]
needs_more_search: bool
iteration: int
def query_decomposer(state: RAGState) -> RAGState:
"""複雑な質問を複数のサブクエリに分解"""
decompose_prompt = f"""
以下の質問を、効率的に検索するための2〜4つのサブクエリに分解してください。
各サブクエリは独立して検索できる具体的なものにしてください。
質問: {state['question']}
サブクエリ(JSON配列で返してください):
"""
result = llm.invoke(decompose_prompt)
# 実際にはJSONパースが必要
queries = [state['question']] # 簡略化
return {"search_queries": queries}
def parallel_retriever(state: RAGState) -> RAGState:
"""複数クエリを並列検索"""
all_docs = []
for query in state['search_queries']:
docs = final_retriever.get_relevant_documents(query)
all_docs.extend([doc.page_content for doc in docs])
# 重複除去
unique_docs = list(dict.fromkeys(all_docs))
return {"retrieved_docs": unique_docs[:10]}
def answer_generator(state: RAGState) -> RAGState:
"""回答生成 + 十分かどうか評価"""
context = "\n\n".join(state['retrieved_docs'])
answer_prompt = f"""
文書: {context}
質問: {state['question']}
回答を生成し、最後に以下のJSONを付けてください:
{{"needs_more_search": true/false, "reason": "理由"}}
十分な情報があれば false、情報が不足していれば true を設定してください。
"""
response = llm.invoke(answer_prompt)
# 実際にはJSONパースが必要
return {
"answer": response.content,
"needs_more_search": False,
"iteration": state.get('iteration', 0) + 1
}
def should_continue(state: RAGState) -> str:
if state['needs_more_search'] and state['iteration'] < 3:
return "retry"
return "end"
# グラフ構築
workflow = StateGraph(RAGState)
workflow.add_node("decompose", query_decomposer)
workflow.add_node("retrieve", parallel_retriever)
workflow.add_node("generate", answer_generator)
workflow.set_entry_point("decompose")
workflow.add_edge("decompose", "retrieve")
workflow.add_edge("retrieve", "generate")
workflow.add_conditional_edges("generate", should_continue, {
"retry": "retrieve",
"end": END
})
agentic_rag = workflow.compile()
result = agentic_rag.invoke({"question": "RAGとFine-tuningの使い分けは?", "iteration": 0})
RAGの評価:2026年おすすめツール比較
| ツール | 特徴 | 主なメトリクス |
|---|---|---|
| Ragas | RAG特化の評価フレームワーク | Faithfulness, Answer Relevancy, Context Recall |
| DeepEval | LLMアウトプット全般の評価 | G-Eval, Hallucination, Contextual Precision |
| LangSmith | LangChainとの深い統合 | トレース + 評価 + フィードバック |
| GAIA | 難易度の高いベンチマーク | エージェントの汎用タスク達成率 |
Ragasで評価する
from ragas import evaluate
from ragas.metrics import (
faithfulness, # 生成された回答が文書に忠実か
answer_relevancy, # 回答が質問に関連しているか
context_recall, # 正解を得るために必要な情報が取得できているか
context_precision, # 取得した文書の中で有用なものの割合
)
from datasets import Dataset
# 評価データセット
eval_data = {
"question": ["RAGとは何ですか?", "ベクトル検索の仕組みを説明してください"],
"answer": [rag_chain.invoke(q) for q in ["RAGとは何ですか?", "ベクトル検索の仕組みを説明してください"]],
"contexts": [
[doc.page_content for doc in final_retriever.get_relevant_documents("RAGとは何ですか?")],
[doc.page_content for doc in final_retriever.get_relevant_documents("ベクトル検索の仕組みを説明してください")],
],
"ground_truth": ["検索拡張生成(RAG)は...", "ベクトル検索は..."]
}
dataset = Dataset.from_dict(eval_data)
result = evaluate(
dataset=dataset,
metrics=[faithfulness, answer_relevancy, context_recall, context_precision],
llm=llm,
embeddings=embeddings
)
print(result.to_pandas())
理想的なスコアの目安:
- Faithfulness: > 0.9(ハルシネーション対策の要)
- Answer Relevancy: > 0.85
- Context Recall: > 0.8
- Context Precision: > 0.7
よくある問題と解決策
1. チャンクの粒度が合わない
症状:長すぎると関連性が薄まり、短すぎるとコンテキストが失われる
解決:ドキュメントの性質に応じてサイズを調整(技術文書: 500〜1000、対話ログ: 200〜500)
2. クエリと文書の表現がずれる
症状:同じ意味でも言い回しが違うと検索ヒットしない
解決:HyDE(仮想的な文書生成でクエリを拡張)+ ハイブリッド検索
3. 古い情報が混ざる
症状:更新前の情報が回答に含まれる
解決:メタデータでフィルタリング(date >= 2025-01-01)+ 定期的な再インデックス
4. 長文書でコンテキストが分断
症状:重要な情報がチャンクの境界で切れる
解決:Hierarchical RAG(文書→セクション→パラグラフの階層管理)
2026年のRAGトレンド
- GraphRAG:知識グラフを活用した関係性を考慮した検索(Microsoft GraphRAG)
- Multi-modal RAG:テキストだけでなく画像・表・グラフも検索対象に
- Adaptive RAG:質問の複雑さに応じて検索戦略を動的に選択
- Long-context RAG:128k〜1Mトークンモデルによる大量文書の直接処理
RAGは「使えれば終わり」ではなく、継続的な評価と改善が必要な技術です。Ragasなどのツールで定期的に品質を測定し、チャンキング戦略や検索パラメータを最適化し続けることが、プロダクションRAGの成功の鍵です。
AIエージェント・RAGツールの総合ディレクトリは AgDex.ai で!460以上のツールを収録中。