rag-patterns

RAG のパターン

標準的な RAG パイプライン

Documents → Chunk → Embed → Store (vector DB)
Query → Embed → Retrieve → Augment prompt → Generate answer

チャンク分割戦略

from langchain_text_splitters import RecursiveCharacterTextSplitter

# Recommended defaults
splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,      # chars (not tokens)
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""],
)
chunks = splitter.split_documents(docs)

戦略	最適な用途	チャンクサイズ
固定サイズ（オーバーラップあり）	一般的なテキスト	500～1000 文字
再帰的な文字分割	構造化されたドキュメント	500～1000 文字
セマンティック（意味に基づく）	長文コンテンツ	可変
ドキュメント構造対応（Markdown ヘッダー）	技術ドキュメント	セクション単位

メタデータの拡充

for chunk in chunks:
    chunk.metadata.update({
        "source": doc.metadata["source"],
        "section": extract_section_title(chunk),
        "doc_id": doc.metadata["id"],
        "chunk_index": i,
    })

検索戦略

ハイブリッド検索（キーワード＋セマンティック）

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

bm25 = BM25Retriever.from_documents(docs, k=5)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

hybrid = EnsembleRetriever(
    retrievers=[bm25, vector_retriever],
    weights=[0.3, 0.7],
)

再ランキング

from cohere import Client

cohere = Client(api_key=COHERE_API_KEY)

def rerank(query: str, documents: list[str], top_n: int = 5):
    response = cohere.rerank(
        model="rerank-english-v3.0",
        query=query,
        documents=documents,
        top_n=top_n,
    )
    return [documents[r.index] for r in response.results]

マルチクエリ検索

# Generate multiple query variations for better recall
prompt = """Generate 3 different versions of this question
to retrieve relevant documents: {question}"""

queries = llm.invoke(prompt).split("\n")
all_docs = set()
for q in queries:
    all_docs.update(retriever.invoke(q))

プロンプトの構築

SYSTEM_PROMPT = """Answer based only on the provided context.
If the context doesn't contain the answer, say "I don't have enough information."
Cite sources using [Source: filename] format.

Context:
{context}"""

def format_context(docs, max_tokens=3000):
    context_parts = []
    for doc in docs:
        source = doc.metadata.get("source", "unknown")
        context_parts.append(f"[Source: {source}]\n{doc.page_content}")
    return "\n\n---\n\n".join(context_parts)

評価

指標	測定内容	ツール
コンテキストの関連性	検索されたドキュメントは関連性がありますか？	RAGAS、手動
忠実性	回答はコンテキストと一致していますか？	RAGAS
回答の関連性	回答は質問に答えていますか？	RAGAS
検索再現率	正しいドキュメントが検索されていますか？	カスタム評価セット

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision

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

アンチパターン

アンチパターン	対策
チャンクが大きすぎる（>1500 文字）	500～1000 文字のチャンクを使い、200 文字をオーバーラップさせます
チャンクにメタデータがない	ソース、セクション、ページ番号を保存します
検索を評価していない	評価セットを作成し、再現率と適合率を測定します
すべてのチャンクをプロンプトに詰め込む	上位 K 件（3～5 件）に制限し、再ランキングを使います
ハイブリッド検索を無視する	BM25 とベクトルを組み合わせて再現率を高めます
引用元／ソースを追跡していない	パイプラインを通してメタデータを渡します

本番環境向けチェックリスト

• 評価セットを使ってチャンク分割戦略を調整済み
• ハイブリッド検索（BM25＋ベクトル）を有効化済み
• 検索結果に再ランキングを適用済み
• 回答に出典を明記済み
• 対象範囲外の質問に対するガードレールを設定済み
• モニタリング：検索レイテンシ、回答品質スコア
• 新しいドキュメントの増分インデックス作成