RAG-architectuur: naïeve, geavanceerde en modulaire RAG-patronen
De term "RAG" omvat eigenlijk een zeer breed spectrum van architecturen, van het eenvoudige patroon tot drie stappen vanaf 2023 naar modulaire systemen vanaf 2026 die queryroutering, herrangschikking en zelf-RAG integreren en consistentiecontroles. Het begrijpen van deze evolutie is van fundamenteel belang: de Naïeve RAG het is snel te implementeren, maar levert opvragingen van complexe documenten van lage kwaliteit op; deGeavanceerde RAG lost specifieke ophaalproblemen op; De Modulair RAG biedt maximale flexibiliteit voor systemen in productie.
Deze gids behandelt de drie architecturen met echte Python-code en vergelijkende kwaliteitsstatistieken en criteria voor het kiezen van het juiste complexiteitsniveau voor uw gebruiksscenario.
Wat je gaat leren
- Naïeve RAG: basisarchitectuur, grenzen en wanneer het genoeg is
- Geavanceerde RAG: vóór het ophalen (herschrijven van zoekopdrachten, HyDE), na het ophalen (herrangschikken)
- Modulaire RAG: Routing, zelf-RAG, CRAG en samenstelbare pijpleidingen
- RAGAS-statistieken om architecturen objectief te vergelijken
- Volledige Python-code voor elke architectuur
- Beslissingsgids: wanneer door te gaan naar het volgende niveau
Naïeve RAG: het basispatroon
De Naive RAG volgt de stroom voor het ophalen en genereren van indexen zonder optimalisaties:
- Indexdocumenten met vaste segmenten (doorgaans 512-1024 tokens)
- Converteert de query naar insluiting en zoekt naar de k meest vergelijkbare segmenten
- Voeg de segmenten samen in de prompt en genereer het antwoord
# 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"])
Grenzen van de naïeve RAG: Slechte prestaties bij dubbelzinnige zoekopdrachten, ophalen van stukken gedeeltelijk relevant, geen casemanagement waarbij de teruggevonden documenten elkaar tegenspreken, variabele kwaliteit met gestructureerde documenten (tabellen, code, lijsten).
Geavanceerde RAG: optimalisaties vóór en na het ophalen
Geavanceerde RAG voegt optimalisaties toe in de pre- en post-retrievalfasen. De meeste technieken van invloed op:
Pre-retrieval: herschrijven van zoekopdrachten en HyDE
Gebruikersvragen zijn vaak dubbelzinnig of slecht geformuleerd. Het herschrijven van zoekopdrachten gebruikt de LLM om herformuleer de zoekopdracht in vormen die geschikter zijn voor semantisch zoeken.
# 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
Na het ophalen: opnieuw rangschikken met Cross-Encoder
Vector-insluitingen gebruiken een "bi-encoder"-representatie (afzonderlijke query en document): en snel maar minder nauwkeurig. Cross-encoder herschikking (query + document samen) verbetert de nauwkeurigheid met 15-25% ten koste van extra latentie (doorgaans 50-150 ms).
# 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}
Modulaire RAG: modulaire architectuur
De 2026 Modular RAG behandelt elke fase van de pijpleiding als een uitwisselbare module. De patronen belangrijkste:
CRAG: Corrigerende RAG
CRAG voegt een relevantieclassifier toe: als de opgehaalde documenten een lage score hebben, het systeem voert een back-upzoekopdracht op internet uit in plaats van te genereren met irrelevante context.
# 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"])
Kwaliteitsvergelijking: naïef versus geavanceerd versus modulair
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)
Wanneer moet u doorgaan naar het volgende niveau?
- Naïef -> Geavanceerd: als betrouwbaarheid < 0,80 of gebruikers reacties rapporteren irrelevant frequent; meerprijs ~2x
- Geavanceerd -> Modulair: Als uw kennisbank slechts een subset omvat van de gevraagde onderwerpen, of als de vragen betrekking hebben op heterogene onderwerpen; extra kosten ~1,3x
- Blijf naïef: als uw kennisbank goed gestructureerd is, zijn de zoekopdrachten dat ook homogeen en trouw > 0,85 al bij het basispatroon
Conclusies
De juiste RAG-architectuur hangt af van de complexiteit van uw use case. Begin altijd met Naïeve RAG, meet met RAGAS en ga alleen verder als de gegevens dit rechtvaardigen. Voeg complexiteit toe zonder metingen leidt tot overontwikkelde systemen die zonder verbeteringen meer kosten meetbaar.
Het volgende artikel gaat dieper in op chunking-strategieën: de component voor het ophalen van pijplijnen die de grootste impact heeft op de kwaliteit van de Naive RAG en die vaak over het hoofd wordt gezien.