RAGシステム完全ガイド：検索拡張生成のすべて

GPT-4やClaudeのようなLLMは素晴らしいものですが、根本的な限界があります。学習データのカットオフ以降の知識を持たず、専門的なドメイン知識に不慣れな場合があり、時として説得力はあるが誤った情報を生成することがあります。これは「ハルシネーション」と呼ばれる現象です。RAG（Retrieval-Augmented Generation）はこれらの問題を解決するための最も実用的なアーキテクチャです。

このガイドでは、基本的なRAGから始め、Self-RAG、Corrective RAG、GraphRAGなどの最先端アーキテクチャまで、本番環境で使用可能なコードとともに解説します。

1. RAGとは？

1.1 LLMの知識の限界

LLMは膨大なテキストコーパスで事前学習されていますが、2つの根本的な制限に直面しています。

**知識カットオフ**：LLMは学習が完了した後の情報を知ることができません。GPT-4の学習データは特定の日付までしか含まれていません。

**ハルシネーション**：LLMは確率的な言語モデルです。「わかりません」と言う代わりに、もっともらしいが不正確な情報を生成する傾向があります。これは特定の事実、日付、引用、数字で特によく起こります。

**ドメイン知識の不足**：社内文書、最新の技術仕様、医療や法律などのドメインの専門知識は、汎用LLMに完全にエンコードすることが難しいです。

1.2 RAGのコアアイデア

RAGのコアアイデアはシンプルです：**LLMが回答を生成する前に、関連情報を検索してコンテキストとして提供する。**

ユーザーの質問 → 関連ドキュメントを検索 → [ドキュメント+質問]をLLMに提供 → 回答を生成

それだけです。しかし詳細、つまり「どのように」検索するか、「どのように」ドキュメントを準備するか、「どのように」コンテキストをLLMに渡すかがシステムの品質を決定します。

1.3 RAG vs ファインチューニング

| 基準 | RAG | ファインチューニング |

| ------------------ | ---------------------------------------- | -------------------- |

| 知識の更新 | リアルタイム、ドキュメントの入れ替えのみ | 再学習が必要 |

| コスト | 比較的低い | 高い（GPUが必要） |

| ソースの追跡 | ソースドキュメントを引用可能 | 不透明 |

| ドメイン固有の形式 | 難しい | うまく機能する |

| 最新情報 | 強み | 学習日付まで |

| ハルシネーション | 低い（ドキュメントに基づく） | 依然として発生し得る |

多くの場合RAGの方が実用的です。ただし、出力形式、スタイル、または専門的なドメイン推論が必要な場合は、ファインチューニングを補完的に使用できます。

1.4 RAGシステムアーキテクチャの概要

完全なRAGパイプラインは2つのフェーズに分かれます。

**オフライン（インデックス作成）フェーズ**：

1. ドキュメントの収集（PDF、HTML、DBなど）

2. テキストのチャンキング（分割）

3. 埋め込みの生成

4. ベクトルデータベースへの保存

**オンライン（クエリ）フェーズ**：

1. ユーザークエリを埋め込む

2. ベクトルデータベースから類似チャンクを検索

3. コンテキストを組み立てる

4. LLMで回答を生成

2. テキスト埋め込み

2.1 埋め込みの概念

埋め込みはテキストを高次元の実数値ベクトルに変換します。重要な洞察は、**意味的に類似したテキストがベクトル空間で近くに配置される**ということです。

例えば：

- 「子犬が遊んでいる」

- 「犬が走っている」

これら2つの文は、コサイン類似度が非常に高い埋め込みを持ちます。

2.2 主要な埋め込みモデル

**OpenAI埋め込み**

- `text-embedding-3-small`：1536次元、高速でコスト効率が良い

- `text-embedding-3-large`：3072次元、高品質

- APIベース、使いやすい、有料サービス

**Sentence-Transformers**

- `all-MiniLM-L6-v2`：384次元、高速で汎用的

- `BAAI/bge-large-en-v1.5`：1024次元、高性能

- ローカルで実行可能、無料

**多言語埋め込みモデル**

- `BAAI/bge-m3`：多言語サポート、多くの言語で強力

- `intfloat/multilingual-e5-large`：多言語で優れた性能

- `paraphrase-multilingual-mpnet-base-v2`：クロスリンガル検索に適している

2.3 コサイン類似度による検索

埋め込みベースの検索は、クエリ埋め込みと保存されたドキュメント埋め込みのコサイン類似度を計算します：

$$\text{similarity}(q, d) = \frac{q \cdot d}{\|q\| \|d\|}$$

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')

documents = [

"Pythonはデータサイエンスに広く使われているプログラミング言語です。",

"機械学習はデータからパターンを学習するAIの分野です。",

"パリはフランスの首都です。",

"ディープラーニングはニューラルネットワークを使った機械学習手法です。",

]

doc_embeddings = model.encode(documents)

print(f"埋め込みの形状: {doc_embeddings.shape}") # (4, 384)

query = "人工知能とデータ分析"

query_embedding = model.encode([query])[0]

def cosine_similarity(a, b):

return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

similarities = [cosine_similarity(query_embedding, emb) for emb in doc_embeddings]

ranked = sorted(zip(similarities, documents), reverse=True)

print("\n類似度でランク付け:")

for score, doc in ranked:

print(f" {score:.4f}: {doc}")

2.4 埋め込み品質の評価（MTEB）

MTEB（Massive Text Embedding Benchmark）は、検索、分類、クラスタリングなどのタスクにわたって埋め込みモデルを体系的に評価するベンチマークです。

RAGで埋め込みモデルを選ぶ際の実践的な基準：

1. ターゲット言語のパフォーマンス（言語固有のベンチマークスコアを確認）

2. 埋め込み次元に対するパフォーマンス（次元が高いほどストレージコストが増加）

3. 推論速度（リアルタイムシステムにとって重要）

4. ライセンス（商用利用の可否を確認）

3. チャンキング戦略

チャンキングはRAGのパフォーマンスにとって最も重要な設計上の決断の1つです。大きすぎるチャンクはノイズが多く、小さすぎるチャンクは十分なコンテキストを欠きます。

3.1 固定サイズチャンキング

最もシンプルなアプローチ。指定した文字数またはトークン数で均一に分割します。

from langchain.text_splitter import CharacterTextSplitter

text = """

機械学習は人工知能のサブフィールドであり、明示的にプログラムすることなく

コンピュータがデータから学習できるようにします。手法には教師あり学習、

教師なし学習、強化学習などがあり、それぞれ異なる種類の問題を解きます。

"""

splitter = CharacterTextSplitter(

chunk_size=200,

chunk_overlap=20,

separator="\n"

)

chunks = splitter.split_text(text)

for i, chunk in enumerate(chunks):

print(f"チャンク {i}: {chunk[:80]}...")

3.2 再帰的チャンキング

LangChainの`RecursiveCharacterTextSplitter`は段落、文、単語の順に再帰的に分割し、意味的な境界をできる限り保持します。

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(

chunk_size=500,

chunk_overlap=50,

separators=["\n\n", "\n", ".", "!", "?", ",", " ", ""]

)

with open("document.txt", "r", encoding="utf-8") as f:

text = f.read()

chunks = splitter.split_text(text)

print(f"合計チャンク数: {len(chunks)}")

print(f"平均チャンク長: {sum(len(c) for c in chunks) / len(chunks):.0f} 文字")

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

埋め込みの類似度を使用して意味的な境界で分割します。隣接する文の間で埋め込み類似度が急激に低下する点が分割点になります。

from langchain_experimental.text_splitter import SemanticChunker

from langchain_openai import OpenAIEmbeddings

semantic_splitter = SemanticChunker(

embeddings=OpenAIEmbeddings(),

breakpoint_threshold_type="percentile",

breakpoint_threshold_amount=95 # 上位5%の類似度変化点で分割

)

chunks = semantic_splitter.create_documents([text])

print(f"セマンティックチャンキング結果: {len(chunks)} チャンク")

3.4 親子チャンキング

検索には小さいチャンク（子）を使用し、コンテキストの提供には大きいチャンク（親）を使用する戦略です。

- 子チャンク：小さく精密（検索精度が高い）

- 親チャンク：大きく包括的（LLMに十分なコンテキスト）

from langchain.retrievers import ParentDocumentRetriever

from langchain.storage import InMemoryStore

from langchain_community.vectorstores import Chroma

from langchain.text_splitter import RecursiveCharacterTextSplitter

from langchain_openai import OpenAIEmbeddings

child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)

parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)

vectorstore = Chroma(

collection_name="full_documents",

embedding_function=OpenAIEmbeddings()

)

store = InMemoryStore()

retriever = ParentDocumentRetriever(

vectorstore=vectorstore,

docstore=store,

child_splitter=child_splitter,

parent_splitter=parent_splitter,

)

3.5 チャンクサイズの決定ガイド

| ユースケース | 推奨チャンクサイズ | オーバーラップ |

| ---------------- | ------------------ | -------------- |

| 事実検索（Q&A） | 200-400トークン | 10-20% |

| ドキュメント要約 | 800-1200トークン | 5-10% |

| コード検索 | 関数/クラス単位 | なし |

| 混合コンテンツ | 512トークン | 50-100トークン |

4. ベクトルデータベース

ベクトルデータベースは高次元ベクトルを保存し、類似ベクトルを効率的に検索するための専用データベースです。

4.1 主要なベクトルデータベースの比較

**FAISS（Facebook AI Similarity Search）**

- Metaが開発したライブラリ

- インメモリ処理、非常に高速

- 本番サーバー不要（ライブラリ）

- 大規模バッチ処理に最適化

**Chroma**

- オープンソース、組み込み埋め込み

- Python-native API

- 開発・プロトタイピングに最適

- 永続化サポート（SQLiteベース）

**Pinecone**

- フルマネージドクラウドサービス

- エンタープライズグレードのスケーリング

- 有料サービス、運用が容易

**Weaviate**

- オープンソース + クラウドオプション

- ハイブリッド検索が組み込み

- GraphQL API

**Milvus**

- 高性能オープンソース

- 分散アーキテクチャ

- 数十億ベクトルへのスケール

**pgvector**

- PostgreSQL拡張

- 既存のPostgreSQLインフラを活用

- SQLによるベクトル検索

4.2 ANNアルゴリズム

完全最近傍探索（KNN）はO(n)時間が必要です。大規模では近似アルゴリズム（ANN）が使用されます。

**HNSW（Hierarchical Navigable Small World）**

階層的グラフ構造による高速検索をサポートします。

- 挿入：O(log n)

- 検索：O(log n)

- 高再現率、高速クエリ

- ChromaとWeaviateのデフォルトアルゴリズム

**IVF（Inverted File Index）**

データをクラスターに分割し、関連するクラスターのみ検索します。

- メモリ効率が良い

- nprobeパラメータで再現率と速度のトレードオフ

- FAISSで広く使用

4.3 FAISSとChromaの実装比較

from langchain_community.vectorstores import Chroma

from langchain_openai import OpenAIEmbeddings

from langchain.schema import Document

===== FAISSの直接使用 =====

d = 384 # ベクトル次元

n = 10000 # ドキュメント数

vectors = np.random.randn(n, d).astype('float32')

Flat L2インデックス（完全探索）

index_flat = faiss.IndexFlatL2(d)

index_flat.add(vectors)

print(f"FAISSインデックスサイズ: {index_flat.ntotal}")

検索

query = np.random.randn(1, d).astype('float32')

k = 5

distances, indices = index_flat.search(query, k)

print(f"上位 {k} 件: {indices[0]}")

HNSWインデックス（近似、高速）

index_hnsw = faiss.IndexHNSWFlat(d, 32)

index_hnsw.add(vectors)

distances_hnsw, indices_hnsw = index_hnsw.search(query, k)

print(f"HNSW結果: {indices_hnsw[0]}")

===== LangChain + Chroma =====

documents = [

Document(page_content="Pythonはデータサイエンスに使われています。", metadata={"source": "doc1"}),

Document(page_content="機械学習はデータからパターンを学習します。", metadata={"source": "doc2"}),

Document(page_content="ディープラーニングはニューラルネットワークベースの機械学習です。", metadata={"source": "doc3"}),

Document(page_content="NLPはテキストデータを分析します。", metadata={"source": "doc4"}),

]

embeddings = OpenAIEmbeddings()

chroma_db = Chroma.from_documents(

documents,

embeddings,

persist_directory="./chroma_db"

)

results = chroma_db.similarity_search("AIと機械学習", k=2)

for doc in results:

print(f"ソース: {doc.metadata['source']}, 内容: {doc.page_content}")

5. 基本的なRAGパイプラインの実装

5.1 LangChainを使った完全なRAGパイプライン

from langchain_community.document_loaders import PyPDFLoader, DirectoryLoader

from langchain.text_splitter import RecursiveCharacterTextSplitter

from langchain_community.vectorstores import Chroma

from langchain_openai import OpenAIEmbeddings, ChatOpenAI

from langchain.chains import RetrievalQA

from langchain.prompts import PromptTemplate

===== 1. ドキュメントの読み込み =====

loader = PyPDFLoader("company_handbook.pdf")

pages = loader.load()

print(f"読み込んだページ数: {len(pages)}")

===== 2. テキストの分割 =====

text_splitter = RecursiveCharacterTextSplitter(

chunk_size=500,

chunk_overlap=50,

separators=["\n\n", "\n", ".", "!", "?", ",", " "]

)

chunks = text_splitter.split_documents(pages)

print(f"作成されたチャンク数: {len(chunks)}")

===== 3. 埋め込みとベクトルDB保存 =====

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

vectorstore = Chroma.from_documents(

chunks,

embeddings,

persist_directory="./rag_db"

)

print("ベクトルDB保存完了")

===== 4. RAGチェーンの設定 =====

prompt_template = """あなたは役に立つAIアシスタントです。

提供されたコンテキストのみを使用して質問に回答してください。

コンテキストに回答がない場合は、「提供されたドキュメントには見つかりませんでした。」と言ってください。

コンテキスト:

{context}

質問: {question}

回答:"""

PROMPT = PromptTemplate(

template=prompt_template,

input_variables=["context", "question"]

)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

retriever = vectorstore.as_retriever(

search_type="similarity",

search_kwargs={"k": 4}

)

qa_chain = RetrievalQA.from_chain_type(

llm=llm,

chain_type="stuff",

retriever=retriever,

chain_type_kwargs={"prompt": PROMPT},

return_source_documents=True

)

===== 5. 質問応答 =====

query = "有給休暇のポリシーは何ですか？"

result = qa_chain.invoke({"query": query})

print(f"\n質問: {query}")

print(f"回答: {result['result']}")

print(f"\nソースドキュメント:")

for doc in result['source_documents']:

print(f" - ページ {doc.metadata.get('page', '?')}: {doc.page_content[:100]}...")

5.2 LlamaIndexを使ったRAG

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings

from llama_index.core.node_parser import SentenceSplitter

from llama_index.llms.openai import OpenAI

from llama_index.embeddings.openai import OpenAIEmbedding

Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0)

Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

Settings.node_parser = SentenceSplitter(chunk_size=512, chunk_overlap=20)

documents = SimpleDirectoryReader("./docs/").load_data()

index = VectorStoreIndex.from_documents(documents)

query_engine = index.as_query_engine(

similarity_top_k=4,

response_mode="compact"

)

response = query_engine.query("これらのドキュメントで扱われている主なトピックは何ですか？")

print(f"回答: {response}")

print("\nソースノード:")

for node in response.source_nodes:

print(f" - スコア: {node.score:.4f}")

print(f" テキスト: {node.text[:100]}...")

6. 高度な検索技術

6.1 ハイブリッド検索

ベクトル検索（意味的）とBM25（キーワードベース）を組み合わせることで、両方のアプローチの強みを活かします。

from langchain_community.retrievers import BM25Retriever

from langchain.retrievers import EnsembleRetriever

BM25リトリーバー（キーワードベース）

bm25_retriever = BM25Retriever.from_documents(chunks)

bm25_retriever.k = 4

ベクトルリトリーバー（意味的）

vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

アンサンブル（ハイブリッド）

ensemble_retriever = EnsembleRetriever(

retrievers=[bm25_retriever, vector_retriever],

weights=[0.5, 0.5]

)

results = ensemble_retriever.invoke("Pythonプログラミングチュートリアル")

print(f"ハイブリッド検索結果: {len(results)}")

6.2 マルチクエリ検索

1つの質問を複数の異なる言い回しに書き直して、検索範囲を広げます。

from langchain.retrievers.multi_query import MultiQueryRetriever

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(temperature=0)

multi_query_retriever = MultiQueryRetriever.from_llm(

retriever=vectorstore.as_retriever(),

llm=llm

)

内部でLLMが質問を複数のバージョンに書き直す

例：「RAGの利点は何ですか？」

→ 「検索拡張生成の利点は何ですか？」

→ 「RAGは標準的なLLMよりなぜ優れているのか？」

→ 「RAGが役立つのはなぜか？」

results = multi_query_retriever.invoke("RAGの利点は何ですか？")

print(f"マルチクエリ検索結果: {len(results)}")

6.3 MMR（Maximal Marginal Relevance）

類似度のみを考慮すると冗長なチャンクが選択される可能性があります。MMRは関連性と多様性のバランスを同時に取ります。

$$\text{MMR} = \arg\max_{d_i \in D \setminus R} [\lambda \cdot \text{sim}(d_i, q) - (1-\lambda) \cdot \max_{d_j \in R} \text{sim}(d_i, d_j)]$$

mmr_retriever = vectorstore.as_retriever(

search_type="mmr",

search_kwargs={

"k": 4, # 最終的に返す件数

"fetch_k": 20, # 初期候補プール

"lambda_mult": 0.5 # バランス：0=多様性、1=類似度

}

)

results = mmr_retriever.invoke("機械学習アルゴリズム")

print(f"MMR結果: {len(results)}")

6.4 メタデータフィルタリング

検索にメタデータ条件を追加してスコープを絞り込みます。

from langchain.schema import Document

docs_with_metadata = [

Document(

page_content="2024年Q1の収益は1000万ドルでした。",

metadata={"year": 2024, "quarter": "Q1", "category": "financial"}

),

Document(

page_content="2024年Q2の収益は1200万ドルでした。",

metadata={"year": 2024, "quarter": "Q2", "category": "financial"}

),

Document(

page_content="技術ロードマップ：AI機能の強化を計画中。",

metadata={"year": 2024, "quarter": "Q1", "category": "strategy"}

),

]

メタデータフィルターで検索を絞り込む

filtered_results = vectorstore.similarity_search(

"収益パフォーマンス",

k=2,

filter={"category": "financial", "year": 2024}

)

7. リランキング

より精度の高いモデルで検索結果を並べ替え、ランキングの品質を向上させます。二段階戦略：検索で高再現率、リランキングで高精度。

7.1 クロスエンコーダーリランカー

クロスエンコーダー（両テキストを一緒にエンコード）は、バイエンコーダー（テキストを個別にエンコードしてから類似度を計算）より精度が高いです。

from sentence_transformers import CrossEncoder

cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

query = "機械学習アルゴリズムの種類"

initial_results = vectorstore.similarity_search(query, k=20) # 多く検索

クロスエンコーダーで再スコアリング

pairs = [[query, doc.page_content] for doc in initial_results]

scores = cross_encoder.predict(pairs)

スコアでソート

ranked = sorted(zip(scores, initial_results), reverse=True)

top_k = [doc for _, doc in ranked[:5]]

print("リランキング後の上位結果:")

for score, doc in ranked[:3]:

print(f" スコア {score:.4f}: {doc.page_content[:80]}...")

7.2 Cohere Rerank API

from langchain.retrievers.document_compressors import CohereRerank

from langchain.retrievers import ContextualCompressionRetriever

compressor = CohereRerank(

cohere_api_key="your-api-key",

top_n=3,

model="rerank-multilingual-v3.0"

)

compression_retriever = ContextualCompressionRetriever(

base_compressor=compressor,

base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20})

)

results = compression_retriever.invoke("会社のポリシーについて教えてください")

print(f"リランキング後のドキュメント: {len(results)}")

7.3 BGEリランカー（オープンソース）

from FlagEmbedding import FlagReranker

reranker = FlagReranker('BAAI/bge-reranker-large', use_fp16=True)

query = "RAGとは何ですか？"

passages = [

"RAGはRetrieval-Augmented Generationの略です。",

"ragは掃除に使う布切れのことです。",

"RAGシステムは検索と生成を組み合わせてLLMの回答を向上させます。",

]

scores = reranker.compute_score([[query, p] for p in passages])

ranked = sorted(zip(scores, passages), reverse=True)

for score, passage in ranked:

print(f" {score:.4f}: {passage}")

8. HyDE（仮説的ドキュメント埋め込み）

8.1 HyDEのアイデア

標準的なRAGはクエリ埋め込みをドキュメント埋め込みと直接比較します。しかし、短いクエリの埋め込みは意味空間で長いドキュメントの埋め込みから遠い場合があります。

HyDEの解決策：LLMに仮説的な回答ドキュメントを生成させ、**その仮説的ドキュメントの埋め込み**を検索に使用します。

クエリ → LLMが仮説的回答を生成 → 仮説的回答を埋め込む → 実際のドキュメントを検索

8.2 HyDEの実装

from langchain.chains import HypotheticalDocumentEmbedder

from langchain_openai import OpenAI, OpenAIEmbeddings, ChatOpenAI

llm = OpenAI()

embeddings = OpenAIEmbeddings()

LangChain組み込みのHyDE

hyde_embeddings = HypotheticalDocumentEmbedder.from_llm(

llm=llm,

embeddings=embeddings,

prompt_key="web_search"

)

手動HyDE実装

def manual_hyde(query, llm, embeddings, vectorstore, k=4):

1. 仮説的ドキュメントを生成

hypothetical_doc = llm.invoke(

f"次の質問に対する詳細な回答を書いてください: {query}"

)

2. 仮説的ドキュメントを埋め込む

hyp_embedding = embeddings.embed_query(hypothetical_doc.content)

3. 仮説的埋め込みで検索

results = vectorstore.similarity_search_by_vector(hyp_embedding, k=k)

return results, hypothetical_doc.content

chat_llm = ChatOpenAI(temperature=0.7)

results, hyp_doc = manual_hyde(

"ディープラーニングの歴史", chat_llm, embeddings, vectorstore

)

print(f"仮説的ドキュメント: {hyp_doc[:200]}...")

print(f"検索された実際のドキュメント: {len(results)}")

9. 高度なRAGアーキテクチャ

9.1 Self-RAG

Self-RAG（2023年、Asaiら）は、LLMが検索が必要かどうかを判断し、検索されたドキュメントの関連性と回答品質を批判的に評価できるようにします。

4つの特殊トークンを使用します：

- `[Retrieve]`：検索が必要か？（Yes/No）

- `[IsRel]`：検索されたドキュメントは関連性があるか？（Relevant/Irrelevant）

- `[IsSup]`：回答はドキュメントに支持されているか？（Supported/Partially/Not）

- `[IsUse]`：回答は有用か？（スコア1-5）

from langchain_openai import ChatOpenAI

from langchain.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class SelfRAGSimulator:

"""Self-RAGの動作をシミュレート（実際のSelf-RAGには特別に学習されたモデルが必要）"""

def __init__(self, retriever, llm):

self.retriever = retriever

self.llm = llm

def should_retrieve(self, query: str) -> bool:

prompt = ChatPromptTemplate.from_template("""

次の質問に回答するために外部ドキュメントを検索する必要があるかどうか判断してください。

一般的な知識や推論だけで回答できる場合は「NO」と答えてください。

特定の事実や専門知識が必要な場合は「YES」と答えてください。

YES か NO のみで答えてください。

質問: {query}

判断:""")

response = self.llm.invoke(prompt.format_messages(query=query))

return "YES" in response.content.upper()

def is_relevant(self, query: str, doc_content: str) -> bool:

prompt = ChatPromptTemplate.from_template("""

次のドキュメントが質問に関連しているかどうか判断してください。

RELEVANT か IRRELEVANT のみで答えてください。

質問: {query}

ドキュメント: {doc}

判断:""")

response = self.llm.invoke(

prompt.format_messages(query=query, doc=doc_content[:500])

)

return "RELEVANT" in response.content.upper()

def generate_with_reflection(self, query: str) -> str:

1. 検索が必要かどうかを判断

need_retrieve = self.should_retrieve(query)

print(f"検索が必要: {need_retrieve}")

if not need_retrieve:

response = self.llm.invoke(query)

return response.content

2. ドキュメントを検索

docs = self.retriever.invoke(query)

3. 関連性でフィルタリング

relevant_docs = [d for d in docs if self.is_relevant(query, d.page_content)]

print(f"関連ドキュメント: {len(relevant_docs)}/{len(docs)}")

if not relevant_docs:

return "関連ドキュメントが見つかりませんでした。一般的な知識で回答します: " + \

self.llm.invoke(query).content

4. コンテキストで回答を生成

context = "\n\n".join([d.page_content for d in relevant_docs[:3]])

prompt = f"""コンテキストを使って質問に答えてください。

コンテキスト: {context}

質問: {query}

回答:"""

return self.llm.invoke(prompt).content

9.2 Corrective RAG（CRAG）

CRAGは検索されたドキュメントの品質を評価し、品質が低い場合はウェブ検索で補完します。

from langchain_community.tools.tavily_search import TavilySearchResults

from typing import List, Tuple

class CorrectiveRAG:

def __init__(self, retriever, llm):

self.retriever = retriever

self.llm = llm

self.web_search = TavilySearchResults(max_results=3)

def evaluate_documents(self, query: str, docs: list) -> Tuple[str, List]:

"""

ドキュメントの関連性を評価します。

戻り値: ("CORRECT"|"INCORRECT"|"AMBIGUOUS", フィルタリングされたドキュメント)

"""

evaluation_prompt = """検索されたドキュメントが質問にどれだけ関連しているか評価してください。

- CORRECT: ドキュメントが質問に直接答えられる

- INCORRECT: ドキュメントが質問と関係ない

- AMBIGUOUS: 部分的に関連しているが不完全

質問: {query}

ドキュメント:

{docs}

評価 (CORRECT/INCORRECT/AMBIGUOUS):"""

docs_text = "\n---\n".join([d.page_content[:300] for d in docs[:4]])

response = self.llm.invoke(

evaluation_prompt.format(query=query, docs=docs_text)

)

evaluation = response.content.strip().upper()

if "CORRECT" in evaluation:

return "CORRECT", docs

elif "INCORRECT" in evaluation:

return "INCORRECT", []

else:

return "AMBIGUOUS", docs

def run(self, query: str) -> str:

1. 初期検索

docs = self.retriever.invoke(query)

2. ドキュメント品質を評価

status, filtered_docs = self.evaluate_documents(query, docs)

print(f"ドキュメント評価: {status}")

3. 各ケースを処理

if status == "INCORRECT":

print("ウェブ検索で補完中...")

web_results = self.web_search.invoke(query)

context = "\n".join([r['content'] for r in web_results])

elif status == "AMBIGUOUS":

web_results = self.web_search.invoke(query)

web_context = "\n".join([r['content'] for r in web_results])

doc_context = "\n".join([d.page_content for d in filtered_docs[:2]])

context = doc_context + "\n\n[ウェブ検索補完]\n" + web_context

else:

context = "\n\n".join([d.page_content for d in filtered_docs[:4]])

4. 最終回答を生成

response = self.llm.invoke(

f"コンテキスト:\n{context}\n\n質問: {query}\n回答:"

)

return response.content

9.3 Adaptive RAG

クエリの複雑さに基づいて検索戦略を動的に選択します。

class AdaptiveRAG:

def __init__(self, simple_retriever, advanced_retriever, llm):

self.simple_retriever = simple_retriever # 基本的なベクトル検索

self.advanced_retriever = advanced_retriever # ハイブリッド + リランキング

self.llm = llm

def classify_query(self, query: str) -> str:

prompt = f"""次の質問の複雑さを分類してください:

- simple: 単純な事実確認または直接答えられる質問

- complex: 複数のソースを組み合わせるか、多段階の推論が必要な質問

質問: {query}

分類 (simple/complex):"""

response = self.llm.invoke(prompt)

return "complex" if "complex" in response.content.lower() else "simple"

def run(self, query: str) -> str:

query_type = self.classify_query(query)

print(f"クエリタイプ: {query_type}")

if query_type == "simple":

docs = self.simple_retriever.invoke(query)

else:

docs = self.advanced_retriever.invoke(query)

context = "\n\n".join([d.page_content for d in docs])

return self.llm.invoke(

f"コンテキスト:\n{context}\n\n質問: {query}\n回答:"

).content

9.4 GraphRAG（Microsoft）

MicrosoftのGraphRAGはドキュメントからナレッジグラフを構築し、グラフ構造を使って検索します。

主要なアイデア：

1. ドキュメントからエンティティ（人物、場所、概念）と関係を抽出

2. コミュニティ検出アルゴリズムで関連するエンティティをグループ化

3. 各コミュニティのサマリーを生成

4. グローバルクエリにはコミュニティサマリーを使用；ローカルクエリにはグラフトラバーサルを使用

GraphRAGのインストールと初期化

pip install graphrag

プロジェクトの初期化

python -m graphrag.index --init --root ./ragtest

設定を編集してインデックス作成

python -m graphrag.index --root ./ragtest

グローバル検索（全ドキュメントセットの理解が必要）

python -m graphrag.query --root ./ragtest --method global "主なテーマは何ですか？"

ローカル検索（特定のエンティティに焦点）

python -m graphrag.query --root ./ragtest --method local "会社Xについて教えてください"

10. RAG評価メトリクス

RAGシステムの品質を客観的に測定することは改善に不可欠です。

10.1 RAGAS（RAG評価）

RAGASはRAGパイプラインを自動的に評価するためのフレームワークです。

**主要メトリクス**：

- **Faithfulness（忠実度）**：回答はコンテキストにどれだけ忠実か？（ハルシネーションを測定）

- **Answer Relevancy（回答関連性）**：回答は質問にどれだけ関連しているか？

- **Context Recall（コンテキスト再現率）**：関連するコンテキストがどれだけ検索されたか？

- **Context Precision（コンテキスト精度）**：検索されたコンテキストのうち実際に有用だった割合は？

from ragas import evaluate

from ragas.metrics import (

faithfulness,

answer_relevancy,

context_recall,

context_precision

)

from datasets import Dataset

evaluation_data = {

"question": [

"会社の年次有給休暇ポリシーは何ですか？",

"リモートワークポリシーは何ですか？",

],

"answer": [

"従業員は1年後に15日の有給休暇を受け取り、毎年1日追加されます。",

"従業員は週3日リモートワークできます。",

],

"contexts": [

["従業員は1年勤務後に15日の有給休暇が付与されます。その後毎年1日追加されます。"],

["従業員は週最大2日リモートワークできます。追加日数は許可を得れば承認される場合があります。"],

],

"ground_truth": [

"1年後に15日、毎年1日追加",

"デフォルトで2日リモートワーク、承認があれば追加可能",

]

}

dataset = Dataset.from_dict(evaluation_data)

result = evaluate(

dataset,

metrics=[faithfulness, answer_relevancy, context_recall, context_precision]

)

print(result)

faithfulness: 0.75（回答とコンテキストの不一致を検出）

answer_relevancy: 0.92

context_recall: 0.85

context_precision: 0.78

11. 本番RAGシステム

11.1 キャッシング戦略

class CachedRAGSystem:

def __init__(self, rag_chain, redis_client=None, ttl=3600):

self.rag_chain = rag_chain

self.redis = redis_client

self.ttl = ttl

def _get_cache_key(self, query: str) -> str:

return f"rag:{hashlib.md5(query.encode()).hexdigest()}"

def query(self, query: str) -> dict:

cache_key = self._get_cache_key(query)

キャッシュを確認

if self.redis:

cached = self.redis.get(cache_key)

if cached:

print("キャッシュヒット!")

return json.loads(cached)

RAGを実行

result = self.rag_chain.invoke({"query": query})

response = {

"answer": result["result"],

"sources": [d.metadata for d in result["source_documents"]]

}

キャッシュに保存

if self.redis:

self.redis.setex(cache_key, self.ttl, json.dumps(response))

return response

11.2 ストリーミングレスポンス

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

from langchain_openai import ChatOpenAI

from fastapi import FastAPI

from fastapi.responses import StreamingResponse

app = FastAPI()

streaming_llm = ChatOpenAI(

model="gpt-4o-mini",

streaming=True,

callbacks=[StreamingStdOutCallbackHandler()]

)

async def generate_rag_stream(query: str):

docs = retriever.invoke(query)

context = "\n\n".join([d.page_content for d in docs])

async for chunk in streaming_llm.astream(

f"コンテキスト:\n{context}\n\n質問: {query}\n回答:"

):

if chunk.content:

yield f"data: {chunk.content}\n\n"

@app.get("/rag/stream")

async def rag_stream_endpoint(query: str):

return StreamingResponse(

generate_rag_stream(query),

media_type="text/event-stream"

)

12. 本番RAGチェックリスト

本番RAGシステムを構築する際の重要な考慮事項。

**ドキュメント処理**

- 多様なファイル形式のサポート（PDF、Word、HTML、Markdown）

- メタデータの保持（ソース、日付、著者）

- 画像とテーブルの戦略

- インクリメンタル更新のサポート

**検索品質**

- ドメインと言語に適した埋め込みモデルを選択

- ハイブリッド検索（キーワード + 意味的）を検討

- チャンクサイズとオーバーラップを適切に調整

- リランキングで精度を向上

**LLM統合**

- 明確なシステムプロンプト（コンテキストのみを使用するよう強調）

- ソースの引用を要求

- 不確実性の表現を許可

**運用**

- コスト削減のためのレスポンスキャッシング

- トークン使用量の監視

- チャンキングと検索パラメータのA/Bテスト

- 自動評価パイプライン（RAGAS）

まとめ

RAGはLLMの限界を克服するための最も実用的なアプローチです。まとめると：

1. **基本的なRAG**：チャンキング → 埋め込み → ベクトルDB → 検索 → 生成

2. **検索品質の向上**：ハイブリッド検索、MMR、リランキング

3. **高度なアーキテクチャ**：Self-RAG、CRAG、HyDE、GraphRAG

4. **評価**：RAGASで忠実度と関連性を測定

5. **本番**：キャッシング、監視、コスト最適化

RAGシステムのパフォーマンスは、単一のコンポーネントではなく、パイプライン全体の調和に依存します。チャンキング戦略と埋め込みモデルの選択が検索品質の80%を決定します。これらの2つの要素に集中することが最高のROIをもたらします。

参考文献

- Lewis et al. (2020), "Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks" (arXiv:2005.11401)

- Asai et al. (2023), "Self-RAG: Learning to Retrieve, Generate, and Critique through Self-Reflection"

- Shi et al. (2023), "REPLUG: Retrieval-Augmented Language Model Pre-Training"

- [LangChain Documentation](https://docs.langchain.com/)

- [LlamaIndex Documentation](https://docs.llamaindex.ai/)

- [RAGAS Documentation](https://docs.ragas.io/)

- [Microsoft GraphRAG](https://github.com/microsoft/graphrag)