RAG アーキテクチャ: 素朴で高度なモジュラー RAG パターン
「RAG」という用語は、実際には、単純なパターンから、 2023 年から 2026 年までの 3 つのステップで、クエリ ルーティング、再ランキング、セルフ RAG を統合するモジュラー システムに移行 そして一貫性チェック。この進化を理解することが基本です。 ナイーブ RAG 実装は迅速ですが、複雑なドキュメントの検索の品質は低くなります。 の高度な RAG 特定の検索問題を解決します。の モジュール式 ラグ 運用環境のシステムに最大限の柔軟性を提供します。
このガイドでは、実際の Python コードを使用して 3 つのアーキテクチャと比較品質メトリックについて説明します。 ユースケースに適した複雑さのレベルを選択するための基準。
何を学ぶか
- Naive RAG: 基本的なアーキテクチャ、制限、そしてそれが十分な場合
- 高度な RAG: 取得前 (クエリ書き換え、HyDE)、取得後 (再ランキング)
- モジュラー RAG: ルーティング、セルフ RAG、CRAG、およびコンポーザブル パイプライン
- アーキテクチャを客観的に比較するための RAGAS メトリクス
- 各アーキテクチャの完全な Python コード
- 意思決定ガイド: 次のレベルに進む時期
Naive RAG: 基本パターン
Naive RAG は、最適化を行わずに、インデックス、取得、生成のフローに従います。
- 固定チャンク (通常は 512 ～ 1024 トークン) を使用してドキュメントにインデックスを付ける
- クエリを埋め込みに変換し、最も類似した k 個のチャンクを検索します。
- チャンクをプロンプトに連結し、応答を生成します。
# Naive RAG con LangChain — implementazione completa
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Qdrant
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import DirectoryLoader, UnstructuredMarkdownLoader
# --- FASE 1: Indicizzazione ---
loader = DirectoryLoader(
"./docs",
glob="**/*.md",
loader_cls=UnstructuredMarkdownLoader
)
documents = loader.load()
# Chunking fisso — il limite principale del Naive RAG
splitter = RecursiveCharacterTextSplitter(
chunk_size=512,
chunk_overlap=64,
separators=["\n\n", "\n", ".", " "]
)
chunks = splitter.split_documents(documents)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Qdrant.from_documents(
chunks, embeddings,
url="http://localhost:6333",
collection_name="naive_rag"
)
# --- FASE 2 + 3: Retrieval + Generation ---
NAIVE_RAG_PROMPT = PromptTemplate(
input_variables=["context", "question"],
template="""Rispondi alla domanda basandoti SOLO sul contesto fornito.
Se il contesto non contiene la risposta, dì "Non ho informazioni su questo argomento".
Contesto:
{context}
Domanda: {question}
Risposta:"""
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
rag_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
chain_type_kwargs={"prompt": NAIVE_RAG_PROMPT},
return_source_documents=True
)
result = rag_chain.invoke({"query": "Come gestire gli errori di timeout?"})
print(result["result"])
Naive RAG の限界: あいまいなクエリ、チャンク取得のパフォーマンスが低い 部分的に関連性があり、回収された文書が互いに矛盾する場合はケース管理が行われない、 構造化ドキュメント (表、コード、リスト) の品質は変わります。
高度な RAG: 取得前および取得後の最適化
Advanced RAG は、取得前および取得後のフェーズに最適化を追加します。最も多くのテクニック 影響を与える:
事前取得: クエリ書き換えと HyDE
ユーザーの質問は、曖昧であったり、表現が不十分だったりすることがよくあります。クエリのリライトでは LLM を使用して、 セマンティック検索により適した形式でクエリを再定式化します。
# Advanced RAG: Query Rewriting + HyDE (Hypothetical Document Embeddings)
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 1. Multi-query: genera query alternative per copertura piu ampia
MULTI_QUERY_PROMPT = ChatPromptTemplate.from_messages([
("system", """Sei un esperto di information retrieval.
Genera 3 varianti della query fornita per recuperare documenti rilevanti
da diverse angolazioni. Restituisci solo le query, una per riga."""),
("human", "Query originale: {query}")
])
multi_query_chain = MULTI_QUERY_PROMPT | llm | StrOutputParser()
def generate_multiple_queries(query: str) -> list[str]:
result = multi_query_chain.invoke({"query": query})
queries = [q.strip() for q in result.strip().split('\n') if q.strip()]
return [query] + queries[:3] # query originale + 3 varianti
# 2. HyDE: genera un documento ipotetico che conterrebbe la risposta
HYDE_PROMPT = ChatPromptTemplate.from_messages([
("system", """Scrivi un breve paragrafo tecnico che risponderebbe
alla seguente domanda, come se fosse tratto da una documentazione ufficiale.
Usa terminologia tecnica precisa."""),
("human", "{query}")
])
hyde_chain = HYDE_PROMPT | llm | StrOutputParser()
def hyde_search(query: str, vectorstore, k: int = 5):
# Genera documento ipotetico
hypothetical_doc = hyde_chain.invoke({"query": query})
# Cerca usando il documento ipotetico come query (invece della query diretta)
results = vectorstore.similarity_search(hypothetical_doc, k=k)
return results
# 3. Multi-query retrieval con deduplicazione
from langchain.retrievers import MergerRetriever
from langchain_community.document_transformers import EmbeddingsRedundantFilter
def advanced_retrieve(query: str, vectorstore, k: int = 5) -> list:
queries = generate_multiple_queries(query)
# Raccogli risultati da tutte le query
all_docs = []
for q in queries:
docs = vectorstore.similarity_search(q, k=k)
all_docs.extend(docs)
# Deduplica per contenuto simile
seen_content = set()
unique_docs = []
for doc in all_docs:
content_hash = hash(doc.page_content[:200])
if content_hash not in seen_content:
seen_content.add(content_hash)
unique_docs.append(doc)
return unique_docs[:k * 2] # ritorna il doppio dei risultati per il reranker
取得後: クロスエンコーダーによる再ランキング
ベクトル埋め込みは「バイエンコーダ」表現 (クエリとドキュメントを分離) を使用します。 ただし精度は低くなります。クロスエンコーダーの再ランキング (クエリとドキュメントの組み合わせ) により精度が向上します 追加の遅延 (通常は 50 ～ 150 ミリ秒) を犠牲にして、15 ～ 25% 短縮されます。
# Post-retrieval: Reranking con Cohere Rerank o cross-encoder locale
import cohere
from sentence_transformers import CrossEncoder
# Opzione 1: Cohere Rerank API (managed, accurato)
co = cohere.Client("your-api-key")
def rerank_with_cohere(query: str, documents: list[str], top_n: int = 5) -> list[dict]:
response = co.rerank(
query=query,
documents=documents,
top_n=top_n,
model="rerank-v3.5"
)
return [
{"content": documents[r.index], "relevance_score": r.relevance_score}
for r in response.results
]
# Opzione 2: Cross-encoder locale (gratuito, ~100MB)
cross_encoder = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
def rerank_local(query: str, documents: list[str], top_n: int = 5) -> list[dict]:
# Crea coppie (query, documento) per il cross-encoder
pairs = [[query, doc] for doc in documents]
scores = cross_encoder.predict(pairs)
# Ordina per score decrescente
ranked = sorted(zip(documents, scores), key=lambda x: x[1], reverse=True)
return [{"content": doc, "relevance_score": float(score)} for doc, score in ranked[:top_n]]
# Advanced RAG completo: multi-query + HyDE + reranking
def advanced_rag(query: str, vectorstore) -> dict:
# 1. Retrieval ampliato
candidates = advanced_retrieve(query, vectorstore, k=8)
candidate_texts = [doc.page_content for doc in candidates]
# 2. Reranking
reranked = rerank_local(query, candidate_texts, top_n=5)
# 3. Generation con contesto di qualita
context = "\n\n---\n\n".join([r["content"] for r in reranked])
response = llm.invoke(f"""Contesto:\n{context}\n\nDomanda: {query}\nRisposta:""")
return {"answer": response.content, "sources": reranked}
モジュラー RAG: モジュラー アーキテクチャ
2026 Modular RAG は、パイプラインの各ステージを交換可能なモジュールとして扱います。パターン 最も重要なこと:
CRAG: コレクティブ RAG
CRAG は関連性分類子を追加します。取得されたドキュメントのスコアが低い場合、 システムは、無関係なコンテキストを生成する代わりに、バックアップ Web 検索を実行します。
# Modular RAG: CRAG (Corrective RAG) con LangGraph
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_community.tools.tavily_search import TavilySearchResults
class RAGState(TypedDict):
query: str
documents: list
relevance_scores: list[float]
web_results: list
answer: str
retrieval_quality: str # "high" | "low" | "ambiguous"
def retrieve(state: RAGState) -> RAGState:
"""Retrieval dal vector store"""
docs = vectorstore.similarity_search_with_score(state["query"], k=5)
documents = [doc for doc, _ in docs]
scores = [float(score) for _, score in docs]
return {**state, "documents": documents, "relevance_scores": scores}
def assess_relevance(state: RAGState) -> RAGState:
"""Valuta se i documenti sono sufficientemente rilevanti"""
avg_score = sum(state["relevance_scores"]) / len(state["relevance_scores"])
if avg_score > 0.85:
quality = "high"
elif avg_score > 0.70:
quality = "ambiguous"
else:
quality = "low"
return {**state, "retrieval_quality": quality}
def web_search_fallback(state: RAGState) -> RAGState:
"""Fallback: web search quando il retrieval e scarso"""
search_tool = TavilySearchResults(max_results=3)
results = search_tool.invoke(state["query"])
return {**state, "web_results": results}
def generate_answer(state: RAGState) -> RAGState:
"""Genera risposta usando documenti disponibili"""
if state["retrieval_quality"] == "low" and state["web_results"]:
context = "\n".join([r["content"] for r in state["web_results"]])
source_type = "web search"
else:
context = "\n".join([doc.page_content for doc in state["documents"]])
source_type = "knowledge base"
response = llm.invoke(
f"Contesto ({source_type}):\n{context}\n\nDomanda: {state['query']}\nRisposta:"
)
return {**state, "answer": response.content}
# Routing basato sulla qualita del retrieval
def should_web_search(state: RAGState) -> str:
return "web_search" if state["retrieval_quality"] == "low" else "generate"
# Costruzione del grafo
graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve)
graph.add_node("assess_relevance", assess_relevance)
graph.add_node("web_search", web_search_fallback)
graph.add_node("generate", generate_answer)
graph.set_entry_point("retrieve")
graph.add_edge("retrieve", "assess_relevance")
graph.add_conditional_edges(
"assess_relevance",
should_web_search,
{"web_search": "web_search", "generate": "generate"}
)
graph.add_edge("web_search", "generate")
graph.add_edge("generate", END)
crag = graph.compile()
# Esecuzione
result = crag.invoke({"query": "Qual e la versione piu recente di Qiskit?"})
print(result["answer"])
品質の比較: ナイーブ、アドバンス、モジュラー
Benchmark su dataset di test enterprise (500 domande, base di conoscenza 50K docs)
Metrica | Naive RAG | Advanced RAG | Modular RAG (CRAG)
--------------------|-----------|--------------|--------------------
Faithfulness | 0.71 | 0.88 | 0.92
Answer Relevancy | 0.74 | 0.86 | 0.89
Context Recall | 0.65 | 0.81 | 0.84
Context Precision | 0.72 | 0.87 | 0.88
--------------------|-----------|--------------|--------------------
Latenza p50 | 850ms | 1.4s | 1.8s (con web fallback: 3.2s)
Costo per query | $0.003 | $0.007 | $0.009 (avg)
--------------------|-----------|--------------|--------------------
"Hallucination rate"| 18% | 6% | 4%
Domande senza risp. | 12% | 8% | 3% (web fallback)
次のレベルに進む時期
- 素朴 -> 上級: 忠実度 < 0.80、またはユーザーが応答を報告した場合 無関係な頻繁。追加コスト ~2倍
- 高度な -> モジュラー: ナレッジ ベースがサブセットのみをカバーしている場合 リクエストされたトピックの範囲、またはクエリが異種トピックにわたるかどうか。追加コスト ~1.3x
- ナイーブを保つ: ナレッジ ベースが適切に構造化されている場合、クエリは次のようになります。 均質性と忠実性 > 0.85 すでに基本パターンで
結論
適切な RAG アーキテクチャは、ユースケースの複雑さによって異なります。常に次から始めます 単純な RAG、RAGAS で測定し、データが正当な場合にのみ前進します。複雑さを加える 測定を行わないとシステムの過剰設計につながり、改善がなければコストが高くなります 測定可能。
次の記事では、チャンク戦略、つまり取得パイプライン コンポーネントについて詳しく説明します。 これは Naive RAG の品質に最も大きな影響を与えますが、見落とされがちです。