Architektura RAG: naiwne, zaawansowane i modułowe wzorce RAG
Termin „RAG” w rzeczywistości obejmuje bardzo szerokie spektrum architektur, od prostego wzoru po trzy kroki od 2023 r. do systemów modułowych od 2026 r., które integrują routing zapytań, zmianę rankingu i samo-RAG i kontrole spójności. Zrozumienie tej ewolucji ma fundamentalne znaczenie: Naiwny RAG jest szybki we wdrożeniu, ale powoduje niską jakość wyszukiwania w przypadku złożonych dokumentów; theZaawansowany RAG rozwiązuje określone problemy związane z wyszukiwaniem; The Modułowy SZARA zapewnia maksymalną elastyczność systemów w produkcji.
W tym przewodniku omówiono trzy architektury z prawdziwym kodem Pythona i porównawczymi wskaźnikami jakości oraz kryteria wyboru odpowiedniego poziomu złożoności dla Twojego przypadku użycia.
Czego się nauczysz
- Naiwny RAG: podstawowa architektura, ograniczenia i kiedy wystarczy
- Zaawansowane RAG: przed pobraniem (przepisanie zapytania, HyDE), po pobraniu (ponowne ustawienie rankingu)
- Modułowe RAG: Rurociągi trasowe, samo-RAG, CRAG i rurociągi komponowalne
- Metryki RAGAS umożliwiające obiektywne porównywanie architektur
- Kompletny kod Pythona dla każdej architektury
- Przewodnik po podejmowaniu decyzji: kiedy przejść na następny poziom
Naiwny RAG: podstawowy wzór
Naiwny RAG podąża za przepływem pobierania i generowania indeksu bez optymalizacji:
- Indeksuj dokumenty ze stałymi fragmentami (zwykle 512–1024 tokenów)
- Konwertuje zapytanie na osadzanie i wyszukuje k najbardziej podobnych fragmentów
- Połącz fragmenty w monit i wygeneruj odpowiedź
# 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"])
Granice naiwnego RAG: Słaba wydajność w przypadku niejednoznacznych zapytań, pobieranie fragmentów częściowo istotne, nie prowadzi się spraw, w których odzyskane dokumenty są ze sobą sprzeczne, zmienna jakość z dokumentami strukturalnymi (tabele, kod, listy).
Zaawansowane RAG: optymalizacje przed i po pobieraniu
Zaawansowany RAG dodaje optymalizacje w fazach przed i po pobieraniu. Najwięcej technik wpływ:
Wstępne pobieranie: przepisywanie zapytań i HyDE
Zapytania użytkowników są często niejednoznaczne lub źle sformułowane. Przepisywanie zapytań wykorzystuje LLM do przeformułuj zapytanie w formie bardziej odpowiedniej dla wyszukiwania semantycznego.
# 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
Po pobraniu: zmiana rankingu za pomocą kodera krzyżowego
Osadzanie wektorów wykorzystuje reprezentację „bi-enkodera” (oddzielne zapytanie i dokument): i szybko ale mniej precyzyjne. Zmiana rankingu między koderami (razem zapytanie + dokument) poprawia precyzję o 15–25% kosztem dodatkowego opóźnienia (zwykle 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}
Modułowa RAG: Architektura modułowa
Modułowy RAG 2026 traktuje każdy etap rurociągu jako wymienny moduł. Wzory najważniejsze:
CRAG: Korygujące RAG
CRAG dodaje klasyfikator trafności: jeśli odnalezione dokumenty mają niski wynik, system wykonuje zapasowe wyszukiwanie w Internecie, zamiast generować je z nieistotnym kontekstem.
# 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"])
Porównanie jakości: naiwne, zaawansowane i modułowe
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)
Kiedy przejść na wyższy poziom
- Naiwny -> Zaawansowany: jeśli wierność < 0,80 lub użytkownicy zgłaszają odpowiedzi nieistotne częste; dodatkowy koszt ~2x
- Zaawansowane -> Modułowe: Jeśli Twoja baza wiedzy obejmuje tylko podzbiór żądanych tematów lub jeśli zapytania dotyczą tematów heterogenicznych; dodatkowy koszt ~1,3x
- Zachowaj naiwność: jeśli Twoja baza wiedzy jest dobrze zorganizowana, zapytania takie są jednorodność i wierność > 0,85 już przy wzorze podstawowym
Wnioski
Właściwa architektura RAG zależy od złożoności Twojego przypadku użycia. Zawsze zaczynaj od Naiwny RAG, mierz za pomocą RAGAS i wyprzedzaj tylko wtedy, gdy dane to uzasadniają. Dodaj złożoność bez pomiarów prowadzi do przeprojektowanych systemów, które kosztują więcej bez ulepszeń mierzalne.
Następny artykuł omawia strategie fragmentowania — komponent potoku pobierania co ma największy wpływ na jakość Naive RAG, a o czym często się zapomina.