rag-implementation

RAG の実装

検索拡張生成（RAG）を習得して、外部のナレッジソースを使い、正確で根拠に基づいた応答を提供する LLM アプリケーションを構築します。

このスキルを使う場面

• 独自のドキュメントを対象とする Q&A システムを構築する場合
• 最新かつ事実に基づいた情報を提供するチャットボットを作成する場合
• 自然言語クエリによるセマンティック検索を実装する場合
• 根拠に基づいた応答によってハルシネーションを減らす場合
• LLM がドメイン固有のナレッジにアクセスできるようにする場合
• ドキュメントアシスタントを構築する場合
• 出典を引用するリサーチツールを作成する場合

このスキルを使わない場面

• 検索を必要としない純粋な生成型の文章作成だけが必要な場合
• データセットが小さすぎて、埋め込みを使うメリットがない場合
• ソースデータを安全に保存または処理できない場合

手順

1. コーパス、更新頻度、評価目標を定義します。
2. 規模に基づいて埋め込みモデルとベクトルストアを選びます。
3. 再ランキングを含む取り込み、チャンク化、検索を構築します。
4. 根拠に基づく QA 指標で評価し、ドリフトを監視します。

安全性

• 機密データをマスキングし、アクセス制御を徹底します。
• 制限対象の場合は、応答でソースドキュメントを公開しないようにします。

主なコンポーネント

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

目的：ドキュメントの埋め込みを効率的に保存、取得します。

選択肢：

• Pinecone：マネージド型、スケーラブル、高速なクエリ
• Weaviate：オープンソース、ハイブリッド検索
• Milvus：高性能、オンプレミス
• Chroma：軽量で使いやすい
• Qdrant：高速でフィルター検索に対応
• FAISS：Meta のライブラリ、ローカルでのデプロイ

2. 埋め込み

目的：テキストを類似度検索用の数値ベクトルに変換します。

モデル：

• text-embedding-ada-002 (OpenAI)：汎用、1536 次元
• all-MiniLM-L6-v2 (Sentence Transformers)：高速、軽量
• e5-large-v2：高品質、多言語対応
• Instructor：タスク固有の指示に対応
• bge-large-en-v1.5：最先端の性能

3. 検索戦略

アプローチ：

• 高密度検索：埋め込みによるセマンティック類似度検索
• 疎検索：キーワードマッチング（BM25、TF-IDF）
• ハイブリッド検索：高密度検索と疎検索を組み合わせます
• マルチクエリ：複数のクエリ候補を生成します
• HyDE：仮説上のドキュメントを生成します

4. 再ランキング

目的：検索結果の順序を並べ替えて、検索品質を向上させます。

手法：

• クロスエンコーダー：BERT ベースの再ランキング
• Cohere Rerank：API ベースの再ランキング
• 最大限界関連性（MMR）：多様性と関連性
• LLM ベース：LLM を使って関連性をスコアリングします

クイックスタート

from langchain.document_loaders import DirectoryLoader
from langchain.text_splitters import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

# 1. Load documents
loader = DirectoryLoader('./docs', glob="**/*.txt")
documents = loader.load()

# 2. Split into chunks
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len
)
chunks = text_splitter.split_documents(documents)

# 3. Create embeddings and vector store
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)

# 4. Create retrieval chain
qa_chain = RetrievalQA.from_chain_type(
    llm=OpenAI(),
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
    return_source_documents=True
)

# 5. Query
result = qa_chain({"query": "What are the main features?"})
print(result['result'])
print(result['source_documents'])

高度な RAG パターン

パターン 1：ハイブリッド検索

from langchain.retrievers import BM25Retriever, EnsembleRetriever

# Sparse retriever (BM25)
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

# Dense retriever (embeddings)
embedding_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# Combine with weights
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, embedding_retriever],
    weights=[0.3, 0.7]
)

パターン 2：マルチクエリ検索

from langchain.retrievers.multi_query import MultiQueryRetriever

# Generate multiple query perspectives
retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=OpenAI()
)

# Single query → multiple variations → combined results
results = retriever.get_relevant_documents("What is the main topic?")

パターン 3：コンテキスト圧縮

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=vectorstore.as_retriever()
)

# Returns only relevant parts of documents
compressed_docs = compression_retriever.get_relevant_documents("query")

パターン 4：親ドキュメント検索

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

# Store for parent documents
store = InMemoryStore()

# Small chunks for retrieval, large chunks for context
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)

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

ドキュメントのチャンク化戦略

再帰的文字列テキスト分割

from langchain.text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
    separators=["\n\n", "\n", " ", ""]  # Try these in order
)

トークンベースの分割

from langchain.text_splitters import TokenTextSplitter

splitter = TokenTextSplitter(
    chunk_size=512,
    chunk_overlap=50
)

セマンティックチャンク化

from langchain.text_splitters import SemanticChunker

splitter = SemanticChunker(
    embeddings=OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile"
)

Markdown ヘッダー分割

from langchain.text_splitters import MarkdownHeaderTextSplitter

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

splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)

ベクトルストアの設定

Pinecone

import pinecone
from langchain.vectorstores import Pinecone

pinecone.init(api_key="your-api-key", environment="us-west1-gcp")

index = pinecone.Index("your-index-name")

vectorstore = Pinecone(index, embeddings.embed_query, "text")

Weaviate

import weaviate
from langchain.vectorstores import Weaviate

client = weaviate.Client("http://localhost:8080")

vectorstore = Weaviate(client, "Document", "content", embeddings)

Chroma（ローカル）

from langchain.vectorstores import Chroma

vectorstore = Chroma(
    collection_name="my_collection",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

検索の最適化

1. メタデータフィルタリング

# Add metadata during indexing
chunks_with_metadata = []
for i, chunk in enumerate(chunks):
    chunk.metadata = {
        "source": chunk.metadata.get("source"),
        "page": i,
        "category": determine_category(chunk.page_content)
    }
    chunks_with_metadata.append(chunk)

# Filter during retrieval
results = vectorstore.similarity_search(
    "query",
    filter={"category": "technical"},
    k=5
)

2. 最大限界関連性

# Balance relevance with diversity
results = vectorstore.max_marginal_relevance_search(
    "query",
    k=5,
    fetch_k=20,  # Fetch 20, return top 5 diverse
    lambda_mult=0.5  # 0=max diversity, 1=max relevance
)

3. クロスエンコーダーによる再ランキング

from sentence_transformers import CrossEncoder

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

# Get initial results
candidates = vectorstore.similarity_search("query", k=20)

# Rerank
pairs = [[query, doc.page_content] for doc in candidates]
scores = reranker.predict(pairs)

# Sort by score and take top k
reranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)[:5]

RAG のプロンプトエンジニアリング

コンテキスト付きプロンプト

prompt_template = """Use the following context to answer the question. If you cannot answer based on the context, say "I don't have enough information."

Context:
{context}

Question: {question}

Answer:"""

引用付き

prompt_template = """Answer the question based on the context below. Include citations using [1], [2], etc.

Context:
{context}

Question: {question}

Answer (with citations):"""

信頼度付き

prompt_template = """Answer the question using the context. Provide a confidence score (0-100%) for your answer.

Context:
{context}

Question: {question}

Answer:
Confidence:"""

評価指標

def evaluate_rag_system(qa_chain, test_cases):
    metrics = {
        'accuracy': [],
        'retrieval_quality': [],
        'groundedness': []
    }

    for test in test_cases:
        result = qa_chain({"query": test['question']})

        # Check if answer matches expected
        accuracy = calculate_accuracy(result['result'], test['expected'])
        metrics['accuracy'].append(accuracy)

        # Check if relevant docs were retrieved
        retrieval_quality = evaluate_retrieved_docs(
            result['source_documents'],
            test['relevant_docs']
        )
        metrics['retrieval_quality'].append(retrieval_quality)

        # Check if answer is grounded in context
        groundedness = check_groundedness(
            result['result'],
            result['source_documents']
        )
        metrics['groundedness'].append(groundedness)

    return {k: sum(v)/len(v) for k, v in metrics.items()}

リソース

• references/vector-databases.md：ベクトル DB の詳細な比較
• references/embeddings.md：埋め込みモデルの選定ガイド
• references/retrieval-strategies.md：高度な検索手法
• references/reranking.md：再ランキング手法と使用場面
• references/context-window.md：コンテキスト制限の管理
• assets/vector-store-config.yaml：設定テンプレート
• assets/retriever-pipeline.py：完全な RAG パイプライン
• assets/embedding-models.md：モデルの比較とベンチマーク

ベストプラクティス

1. チャンクサイズ：コンテキストと具体性のバランスを取ります（500～1000 トークン）
2. オーバーラップ：境界でコンテキストを保つため、10～20% オーバーラップさせます
3. メタデータ：フィルタリングやデバッグのために、ソース、ページ、タイムスタンプを含めます
4. ハイブリッド検索：最適な結果を得るため、セマンティック検索とキーワード検索を組み合わせます
5. 再ランキング：クロスエンコーダーを使って上位の結果を改善します
6. 引用：透明性を確保するため、常にソース文書を返します
7. 評価：検索品質と回答精度を継続的にテストします
8. モニタリング：本番環境で検索メトリクスを追跡します

よくある問題

• 検索精度が低い：埋め込みの品質、チャンクサイズ、クエリの組み立て方を確認します
• 関連性の低い結果：メタデータフィルタリングを追加し、ハイブリッド検索を使い、再ランキングします
• 情報が見つからない：文書が適切にインデックスされていることを確認します
• クエリが遅い：ベクトルストアを最適化し、キャッシュを使い、k を小さくします
• ハルシネーション：グラウンディング用プロンプトを改善し、検証ステップを追加します