RAG実装パターン完全ガイド - ベクトル検索からハイブリッド検索まで

1. ドキュメントをチャンク分割
2. 各チャンクをベクトル化
3. Vector DBに保存
4. ユーザークエリをベクトル化
5. 類似ベクトル検索（Top-K）
6. 取得したチャンクをLLMのコンテキストに追加
7. LLMが回答生成

from langchain.vectorstores import Qdrant
from langchain.embeddings import OpenAIEmbeddings
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA

# ベクトルストア
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Qdrant.from_documents(
    documents,
    embeddings,
    url="http://localhost:6333",
    collection_name="docs"
)

# RAGチェーン
llm = ChatOpenAI(model="gpt-4")
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    return_source_documents=True
)

# 質問
result = qa_chain("2025年のAI技術トレンドは？")
print(result["result"])

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

# BM25リトリーバー
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5

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

# ハイブリッドリトリーバー
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # BM25: 40%, Vector: 60%
)

# RAGチェーン
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=ensemble_retriever
)

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank

# Cohereリランカー
compressor = CohereRerank(model="rerank-english-v2.0", top_n=3)

# 圧縮リトリーバー
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vector_retriever
)

# RAGチェーン
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=compression_retriever
)

from sentence_transformers import CrossEncoder

# モデル読み込み
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

# リランキング
def rerank_documents(query: str, docs: list, top_k: int = 3):
    pairs = [[query, doc.page_content] for doc in docs]
    scores = reranker.predict(pairs)
    
    # スコアでソート
    ranked_docs = sorted(zip(docs, scores), key=lambda x: x[1], reverse=True)
    return [doc for doc, score in ranked_docs[:top_k]]

# 使用例
retrieved_docs = vector_retriever.get_relevant_documents("AIエージェント")
reranked_docs = rerank_documents("AIエージェント", retrieved_docs, top_k=3)

from langchain.text_splitter import CharacterTextSplitter

splitter = CharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
chunks = splitter.split_documents(documents)

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", "、", " "]
)
chunks = splitter.split_documents(documents)

# Markdownドキュメント
from langchain.text_splitter import MarkdownHeaderTextSplitter

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

splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)
chunks = splitter.split_text(markdown_document)

# ドキュメントにメタデータ追加
documents = [
    Document(
        page_content="AIエージェントの最新動向",
        metadata={"category": "AI", "date": "2025-01-15", "author": "John"}
    ),
    Document(
        page_content="ブロックチェーン技術の進化",
        metadata={"category": "Blockchain", "date": "2025-02-01", "author": "Alice"}
    )
]

# Vector DBに保存
vectorstore = Qdrant.from_documents(
    documents,
    embeddings,
    url="http://localhost:6333",
    collection_name="filtered_docs"
)

# フィルタリング検索
retriever = vectorstore.as_retriever(
    search_kwargs={
        "k": 5,
        "filter": {"category": "AI", "date": {"$gte": "2025-01-01"}}
    }
)

results = retriever.get_relevant_documents("最新のAI技術")

from ragas import evaluate
from ragas.metrics import context_recall, context_precision

# 評価データ
eval_data = {
    "question": ["AIエージェントとは？"],
    "contexts": [retrieved_contexts],
    "ground_truths": [["AIエージェントは自律的に..."]]
}

# 評価
results = evaluate(
    eval_data,
    metrics=[context_recall, context_precision]
)
print(results)

from ragas.metrics import answer_relevancy, faithfulness

results = evaluate(
    eval_data,
    metrics=[answer_relevancy, faithfulness]
)

from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

# キャッシュ有効化
set_llm_cache(InMemoryCache())

# 同一クエリは再計算せず、キャッシュから返却

def compress_context(docs: list, max_tokens: int = 3000) -> str:
    """コンテキストをトークン数制限内に圧縮"""
    context = ""
    token_count = 0
    
    for doc in docs:
        doc_tokens = len(doc.page_content.split())
        if token_count + doc_tokens > max_tokens:
            break
        context += doc.page_content + "\n\n"
        token_count += doc_tokens
    
    return context

def safe_rag_query(query: str) -> str:
    try:
        result = qa_chain(query)
        return result["result"]
    except Exception as e:
        logger.error(f"RAG error: {e}")
        return "申し訳ございません。エラーが発生しました。"