Introduzione
La Retrieval-Augmented Generation è iniziata come un'idea semplice: invece di affidarsi esclusivamente alla memoria parametrica di un modello linguistico, recuperare documenti rilevanti da una base di conoscenza esterna e includerli nel contesto del prompt. Questo pattern RAG naïve, composto da embed-recupera-genera, ha funzionato sorprendentemente bene per la risposta a domande semplici su collezioni di documenti strutturati. Ma quando i team hanno portato il RAG in produzione per casi d'uso più complessi, le sue limitazioni sono diventate dolorosamente evidenti. Le query che richiedevano ragionamento attraverso documenti multipli, le domande ambigue che necessitavano chiarimenti e le basi di conoscenza con tipi di contenuto eterogenei hanno esposto la fragilità della pipeline recupera-poi-genera.
Il RAG agentico rappresenta un cambiamento fondamentale nel modo in cui progettiamo i sistemi di recupero. Invece di una pipeline fissa dove ogni passaggio alimenta linearmente il successivo, il RAG agentico dà al modello linguistico la capacità di pianificare la propria strategia di recupero, valutare la qualità dei risultati recuperati, riformulare le query quando il recupero iniziale fallisce e decidere quando ha informazioni sufficienti per generare una risposta finale. Il modello diventa un partecipante attivo nel processo di recupero piuttosto che un consumatore passivo del contesto recuperato.
Questa guida copre i pattern di architettura, i framework di implementazione, i metodi di valutazione e le considerazioni di produzione per costruire sistemi RAG agentici. Ci basiamo su deployment reali in basi di conoscenza aziendali, sistemi di assistenza clienti e piattaforme di documentazione tecnica dove questi pattern sono stati validati su larga scala.
Dal RAG Naïve al RAG Agentico: Cosa è Cambiato
Il RAG naïve segue una pipeline deterministica a tre passaggi: la query dell'utente viene embedded, una ricerca di similarità vettoriale recupera i top-k frammenti di documenti più simili, e questi frammenti vengono concatenati in un prompt che l'LLM utilizza per generare una risposta. Questa pipeline ha tre debolezze fondamentali che il RAG agentico affronta.
Primo, il RAG naïve presume che la query dell'utente sia già ben formulata per il recupero. In pratica, le query degli utenti sono spesso vaghe, multi-sfaccettate o utilizzano terminologia che non corrisponde al vocabolario dei documenti indicizzati. Un utente che chiede informazioni sui fallimenti di deployment potrebbe aver bisogno di documenti sulla gestione degli errori, la configurazione dell'infrastruttura e le pipeline CI/CD, ma una singola ricerca vettoriale potrebbe far emergere solo una di queste sfaccettature.
Secondo, il RAG naïve non ha un cancello di qualità. Se i documenti recuperati sono irrilevanti, obsoleti o insufficienti, la pipeline procede comunque e l'LLM genera una risposta da un contesto scarso. Non c'è alcun meccanismo affinché il sistema riconosca un fallimento del recupero e riprovi.
Terzo, il RAG naïve tratta tutte le query in modo identico. Una domanda di ricerca fattuale, una domanda analitica complessa e una domanda che richiede sintesi da fonti multiple passano tutte attraverso la stessa pipeline recupera-genera. Il RAG agentico introduce logica condizionale che seleziona diverse strategie di recupero e generazione basandosi sulle caratteristiche della query.
La transizione dal RAG naïve a quello agentico implica l'aggiunta di tre capacità: pianificazione delle query (scomporre query complesse in sotto-query), valutazione del recupero (valutare se il contesto recuperato è sufficiente) e raffinamento iterativo (riformulare le query e ri-recuperare quando i risultati sono insufficienti). Queste capacità trasformano una pipeline statica in un sistema dinamico e auto-correttivo.
Pattern di Architettura Fondamentali
I sistemi RAG agentici sono costruiti da un piccolo numero di pattern componibili. Comprendere questi pattern ti permette di progettare sistemi su misura per il tuo caso d'uso specifico.
Routing
Il pattern agentico più semplice instrada le query verso diversi backend di recupero basandosi sulla classificazione della query. Un agente router analizza la query in arrivo e la dirige verso la fonte di conoscenza più appropriata:
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
router_prompt = ChatPromptTemplate.from_messages([
("system", """Classify the user query into one of these categories:
- technical_docs: Questions about API usage, configuration, or code
- policy: Questions about company policies, procedures, or compliance
- support: Questions about troubleshooting or known issues
- general: General questions that don't fit other categories
Respond with ONLY the category name."""),
("human", "{query}")
])
router_chain = router_prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()
Il routing è prezioso quando la tua base di conoscenza abbraccia domini multipli con diverse strategie di indicizzazione. La documentazione tecnica potrebbe essere indicizzata con chunking consapevole del codice e embeddings, mentre i documenti di policy utilizzano chunking semantico con filtri di metadati.
Decomposizione delle Query
Le domande complesse spesso richiedono informazioni da frammenti di documenti multipli che non apparirebbero insieme in un singolo recupero. La decomposizione delle query spezza una query complessa in sotto-query indipendenti, recupera per ciascuna e sintetizza i risultati:
decomposition_prompt = ChatPromptTemplate.from_messages([
("system", """Break the following complex question into 2-4 simpler
sub-questions that, when answered together, provide a complete answer
to the original question. Return as a JSON array of strings."""),
("human", "{query}")
])
# Example: "How does our API rate limiting compare to competitors
# and what are customers saying about it?"
# Decomposes to:
# ["What are our current API rate limiting policies?",
# "What rate limiting do our main competitors use?",
# "What customer feedback have we received about rate limiting?"]
Auto-Correzione
L'auto-correzione è il pattern che più distingue il RAG agentico dal RAG naïve. Dopo il recupero, un passaggio di valutazione esamina se i documenti recuperati sono rilevanti e sufficienti. Se non lo sono, il sistema riformula la query e riprova:
grading_prompt = ChatPromptTemplate.from_messages([
("system", """You are a retrieval quality grader. Given a user question
and a set of retrieved documents, determine:
1. Are the documents relevant to the question? (yes/no)
2. Do the documents contain sufficient information to answer? (yes/no)
3. If insufficient, suggest a reformulated query.
Respond as JSON with keys: relevant, sufficient, reformulated_query"""),
("human", "Question: {query}\n\nDocuments: {documents}")
])
Questo pattern tipicamente permette 2-3 tentativi di recupero prima di ricorrere a un messaggio di fallimento elegante. Le query riformulate spesso passano dalla ricerca per similarità semantica alla ricerca basata su parole chiave, ampliano l'ambito o mirano a campi di metadati specifici.
LangGraph per l'Orchestrazione di Agenti con Stato
LangGraph è diventato il framework standard per costruire sistemi RAG agentici perché fornisce gestione dello stato esplicita, routing condizionale e supporto per cicli che le astrazioni più semplici basate su catene in LangChain non possono esprimere.
Un sistema RAG agentico basato su LangGraph è definito come un grafo di stati dove i nodi rappresentano passaggi di elaborazione e gli archi rappresentano transizioni tra passaggi. Gli archi condizionali permettono al grafo di ramificarsi basandosi su risultati intermedi:
from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Annotated
from operator import add
class AgentState(TypedDict):
query: str
sub_queries: List[str]
documents: List[dict]
generation: str
retry_count: int
retrieval_grade: str
def route_query(state: AgentState) -> AgentState:
"""Classify and route the incoming query."""
query = state["query"]
classification = router_chain.invoke({"query": query})
state["route"] = classification
return state
def retrieve(state: AgentState) -> AgentState:
"""Retrieve documents based on query and route."""
query = state["query"]
docs = retriever.invoke(query)
state["documents"] = docs
return state
def grade_documents(state: AgentState) -> AgentState:
"""Grade retrieved documents for relevance and sufficiency."""
grade = grading_chain.invoke({
"query": state["query"],
"documents": state["documents"]
})
state["retrieval_grade"] = grade["sufficient"]
state["retry_count"] = state.get("retry_count", 0) + 1
if not grade["sufficient"] and grade.get("reformulated_query"):
state["query"] = grade["reformulated_query"]
return state
def generate(state: AgentState) -> AgentState:
"""Generate final answer from retrieved context."""
answer = generation_chain.invoke({
"query": state["query"],
"documents": state["documents"]
})
state["generation"] = answer
return state
def should_retry(state: AgentState) -> str:
"""Decide whether to retry retrieval or proceed to generation."""
if state["retrieval_grade"] == "yes":
return "generate"
if state["retry_count"] >= 3:
return "generate" # Give up and generate with what we have
return "retrieve"
# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("route", route_query)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade", grade_documents)
workflow.add_node("generate", generate)
workflow.set_entry_point("route")
workflow.add_edge("route", "retrieve")
workflow.add_edge("retrieve", "grade")
workflow.add_conditional_edges("grade", should_retry, {
"retrieve": "retrieve",
"generate": "generate"
})
workflow.add_edge("generate", END)
app = workflow.compile()
La struttura del grafo rende il flusso di controllo esplicito e debuggabile. Puoi visualizzare il grafo, tracciare le esecuzioni attraverso ogni nodo e comprendere esattamente perché il sistema ha preso un percorso particolare per una data query. Questa trasparenza è critica per i sistemi di produzione dove devi diagnosticare fallimenti e spiegare il comportamento.
LangGraph supporta anche pattern più avanzati come il recupero parallelo attraverso indici multipli, checkpoint human-in-the-loop dove l'agente si mette in pausa per la conferma dell'utente, e stato persistente che sopravvive tra le conversazioni.
Strategie di Recupero
Il livello di recupero in un sistema RAG agentico è tipicamente più sofisticato di una singola ricerca in un vector store. I sistemi di produzione combinano strategie di recupero multiple.
Ricerca Ibrida
La ricerca ibrida combina il recupero vettoriale denso con il recupero sparso basato su parole chiave, dandoti la comprensione semantica degli embeddings con la precisione del matching di parole chiave BM25:
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
from langchain_chroma import Chroma
vectorstore = Chroma(
collection_name="documents",
embedding_function=embedding_model,
)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 10
hybrid_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.6, 0.4],
)
Re-ranking
Dopo il recupero iniziale, un re-ranker con cross-encoder valuta ogni documento rispetto alla query con una precisione molto superiore alla similarità con bi-encoder. Questo è computazionalmente costoso ma migliora drasticamente la precisione:
from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank
reranker = CohereRerank(model="rerank-v3.5", top_n=5)
compression_retriever = ContextualCompressionRetriever(
base_compressor=reranker,
base_retriever=hybrid_retriever,
)
Recupero Multi-Indice
Le basi di conoscenza di produzione spesso coprono indici multipli con schemi diversi, modelli di embeddings e tipi di documenti. Un sistema agentico può interrogare indici multipli in parallelo e unire i risultati:
async def multi_index_retrieve(query: str, indexes: list) -> list:
"""Retrieve from multiple indexes in parallel."""
import asyncio
async def retrieve_from_index(index, query):
return await index.aretrieve(query)
tasks = [retrieve_from_index(idx, query) for idx in indexes]
results = await asyncio.gather(*tasks)
# Flatten and deduplicate
all_docs = []
seen_ids = set()
for result_set in results:
for doc in result_set:
if doc.metadata["id"] not in seen_ids:
all_docs.append(doc)
seen_ids.add(doc.metadata["id"])
return all_docs
Valutazione con RAGAS e DeepEval
Valutare i sistemi RAG agentici richiede di misurare dimensioni multiple di qualità. RAGAS e DeepEval sono i due framework di valutazione più ampiamente adottati, ciascuno con punti di forza distinti.
RAGAS fornisce un insieme di metriche senza riferimento che valutano la qualità del RAG senza richiedere risposte di verità base per ogni query di test:
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
)
from datasets import Dataset
eval_dataset = Dataset.from_dict({
"question": questions,
"answer": generated_answers,
"contexts": retrieved_contexts,
"ground_truth": reference_answers,
})
results = evaluate(
dataset=eval_dataset,
metrics=[
faithfulness,
answer_relevancy,
context_precision,
context_recall,
],
)
print(results)
La fedeltà misura se la risposta generata è supportata dal contesto recuperato, rilevando le allucinazioni. La rilevanza della risposta misura se la risposta affronta la domanda. La precisione del contesto misura se i documenti recuperati sono rilevanti, e il richiamo del contesto misura se il recupero ha catturato tutte le informazioni necessarie.
DeepEval estende la valutazione con metriche aggiuntive particolarmente rilevanti per i sistemi agentici:
from deepeval import evaluate
from deepeval.metrics import (
FaithfulnessMetric,
AnswerRelevancyMetric,
ContextualRelevancyMetric,
HallucinationMetric,
)
from deepeval.test_case import LLMTestCase
test_case = LLMTestCase(
input="How do I configure rate limiting?",
actual_output=generated_answer,
expected_output=reference_answer,
retrieval_context=retrieved_chunks,
)
metrics = [
FaithfulnessMetric(threshold=0.8),
AnswerRelevancyMetric(threshold=0.7),
ContextualRelevancyMetric(threshold=0.7),
HallucinationMetric(threshold=0.5),
]
evaluate(test_cases=[test_case], metrics=metrics)
Per il RAG agentico specificamente, dovresti anche misurare l'efficienza del recupero: quanti round di recupero necessita il sistema in media, quale percentuale di query richiede riformulazione e come si degrada la qualità della risposta attraverso i tentativi di retry. Queste metriche sono specifiche del loop agentico e non sono coperte dai framework standard di valutazione RAG.
Pattern di Produzione
Portare il RAG agentico dal prototipo alla produzione richiede di affrontare preoccupazioni di affidabilità, sicurezza e operative che non emergono durante lo sviluppo.
Guardrail
Ogni sistema RAG di produzione necessita di guardrail che impediscano all'LLM di generare risposte dannose, fuori tema o senza supporto fattuale:
from guardrails import Guard
from guardrails.hub import ToxicLanguage, CompetitorCheck
guard = Guard().use_many(
ToxicLanguage(on_fail="exception"),
CompetitorCheck(
competitors=["competitor_a", "competitor_b"],
on_fail="fix"
),
)
raw_response = generation_chain.invoke({"query": query, "documents": docs})
validated_response = guard.validate(raw_response)
Catene di Fallback
Quando il loop agentico non riesce a trovare contesto sufficiente dopo il massimo di retry, il sistema necessita di degradazione elegante piuttosto che generare una risposta inaffidabile:
def generate_with_fallback(state: AgentState) -> AgentState:
if state["retrieval_grade"] != "yes" and state["retry_count"] >= 3:
state["generation"] = (
"I wasn't able to find sufficient information in our knowledge base "
"to fully answer your question. Here's what I found:\n\n"
f"{partial_answer_from_context(state['documents'])}\n\n"
"For a complete answer, I'd recommend contacting the support team."
)
state["confidence"] = "low"
else:
state["generation"] = generation_chain.invoke({
"query": state["query"],
"documents": state["documents"]
})
state["confidence"] = "high"
return state
Human-in-the-Loop
Per applicazioni ad alto rischio, LangGraph supporta punti di interruzione dove l'agente mette in pausa l'esecuzione e attende l'approvazione umana:
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
# Add interrupt before generation for sensitive queries
workflow.add_node("human_review", lambda state: state)
workflow.add_conditional_edges(
"grade",
lambda state: "human_review" if state.get("sensitive") else "generate",
{"human_review": "human_review", "generate": "generate"}
)
app = workflow.compile(
checkpointer=checkpointer,
interrupt_before=["human_review"]
)
Osservabilità con LangSmith e Phoenix
L'osservabilità è non negoziabile per i sistemi agentici in produzione. La natura non deterministica del flusso di controllo guidato da LLM significa che non puoi prevedere né testare ogni possibile percorso di esecuzione. Hai bisogno di tracciamento completo per capire cosa sta facendo il tuo sistema in produzione.
LangSmith fornisce tracciamento end-to-end per le applicazioni LangGraph:
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "agentic-rag-production"
# All LangGraph invocations are automatically traced
result = app.invoke({"query": "How do I configure rate limiting?"})
Ogni traccia mostra il percorso di esecuzione completo attraverso il grafo: quali nodi sono stati visitati, quali erano gli input e output dell'LLM ad ogni passaggio, risultati del recupero, decisioni di valutazione e conteggi di token. Questo è essenziale per il debug dei casi in cui l'agente prende un percorso inaspettato o genera una risposta scadente.
Arize Phoenix fornisce un'alternativa open source con focus sul monitoraggio della qualità del recupero:
import phoenix as px
from phoenix.trace.langchain import LangChainInstrumentor
px.launch_app()
LangChainInstrumentor().instrument()
# Traces are now collected in Phoenix
result = app.invoke({"query": user_query})
Phoenix è particolarmente prezioso per monitorare la qualità del recupero nel tempo. Traccia la deriva degli embeddings, le distribuzioni di rilevanza del recupero e può avvisare quando la qualità del recupero si degrada, cosa che spesso indica che la base di conoscenza si è allontanata dalla distribuzione di addestramento del modello di embeddings e necessita re-indicizzazione.
Metriche chiave da monitorare in produzione:
# Custom metrics to track
metrics = {
"avg_retrieval_rounds": [],
"reformulation_rate": [],
"fallback_rate": [],
"latency_p50_ms": [],
"latency_p99_ms": [],
"token_cost_per_query": [],
"faithfulness_score": [],
}
Ottimizzazione di Costi e Latenza
I sistemi RAG agentici sono intrinsecamente più costosi e lenti del RAG naïve perché effettuano chiamate multiple all'LLM per query. Ottimizzare costi e latenza senza sacrificare la qualità richiede decisioni architettoniche attente.
Usa modelli più piccoli per routing e valutazione. Il router e il valutatore di documenti non necessitano della piena capacità di ragionamento di un modello di frontiera. GPT-4o-mini o Claude 3.5 Haiku possono gestire questi compiti a una frazione del costo e della latenza:
# Use a small, fast model for routing and grading
routing_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
grading_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# Use a larger model only for final generation
generation_model = ChatOpenAI(model="gpt-4o", temperature=0.1)
Metti in cache i risultati del recupero in modo aggressivo. Molte query degli utenti sono variazioni della stessa domanda sottostante. Una cache semantica che abbina le query entro una soglia di similarità può eliminare recuperi e chiamate LLM ridondanti:
from langchain_community.cache import RedisSemanticCache
set_llm_cache(RedisSemanticCache(
redis_url="redis://localhost:6379",
embedding=embedding_model,
score_threshold=0.95,
))
Imposta timeout aggressivi su ogni passaggio per prevenire latenza incontrollata. Una singola chiamata LLM lenta non dovrebbe causare il timeout dell'intera richiesta:
from asyncio import wait_for, TimeoutError
async def retrieve_with_timeout(state, timeout_seconds=5):
try:
return await wait_for(retrieve(state), timeout=timeout_seconds)
except TimeoutError:
state["documents"] = []
state["retrieval_grade"] = "no"
return state
Casi di Studio Reali e Anti-Pattern
Il pattern di fallimento più comune nei deployment di RAG agentico è il loop di retry infinito. Senza un limite rigido sui tentativi di recupero e un comportamento di fallback chiaro, l'agente può ciclare attraverso riformulazioni indefinitamente, bruciando token e latenza. Imposta sempre limiti di retry espliciti e misura il tuo tasso di riformulazione. Se più del 20% delle query richiede riformulazione, il problema è probabilmente nel tuo livello di recupero (chunking scadente, modello di embeddings sbagliato, indice obsoleto) piuttosto che nella logica dell'agente.
Un altro anti-pattern frequente è l'over-routing. I team a volte costruiscono grafi di routing complessi con decine di indici specializzati quando un singolo retriever ibrido ben progettato avrebbe prestazioni migliori. Inizia con l'architettura più semplice che soddisfa i tuoi requisiti e aggiungi complessità solo quando hai metriche che mostrano che gli approcci più semplici sono insufficienti.
Il riempimento della finestra di contesto è un terzo anti-pattern. Recuperare 20 frammenti di documenti e stiparli tutti nel prompt spreca token e può effettivamente degradare la qualità della risposta perché l'LLM fatica a identificare le informazioni rilevanti in un mare di contesto marginalmente correlato. Il re-ranking a un piccolo numero di frammenti di alta qualità (tipicamente 3-5) supera costantemente le finestre di contesto più grandi nei benchmark di valutazione.
Infine, fai attenzione all'over-ottimizzazione guidata dalla valutazione. I team che ottimizzano esclusivamente per i punteggi RAGAS possono costruire sistemi che ottengono buoni punteggi nei benchmark ma producono risposte eccessivamente caute e evasive che gli utenti trovano inutili. Bilancia la valutazione quantitativa con la revisione qualitativa delle conversazioni reali degli utenti.
Il Futuro: Sistemi RAG Multi-Agente
La prossima frontiera nel RAG agentico sono i sistemi multi-agente dove agenti specializzati collaborano su compiti complessi di recupero e sintesi. Invece di un singolo agente che gestisce l'intera pipeline di recupero-generazione, si schiera un team di agenti con specializzazioni diverse.
Un agente di ricerca gestisce la decomposizione delle query e il recupero multi-hop attraverso grandi collezioni di documenti. Un agente di fact-checking verifica le affermazioni generate contro i documenti sorgente. Un agente di sintesi combina i risultati di sotto-query multiple in una risposta coerente e ben strutturata. Un agente editor revisiona l'output finale per chiarezza, accuratezza e tono.
from langgraph.graph import StateGraph
# Multi-agent RAG architecture
multi_agent = StateGraph(MultiAgentState)
multi_agent.add_node("planner", planner_agent)
multi_agent.add_node("researcher", researcher_agent)
multi_agent.add_node("fact_checker", fact_check_agent)
multi_agent.add_node("synthesizer", synthesis_agent)
multi_agent.set_entry_point("planner")
multi_agent.add_edge("planner", "researcher")
multi_agent.add_edge("researcher", "fact_checker")
multi_agent.add_conditional_edges(
"fact_checker",
lambda s: "researcher" if s["needs_more_research"] else "synthesizer",
)
multi_agent.add_edge("synthesizer", END)
Questa architettura è più costosa per query ma gestisce domande di ricerca complesse con cui i sistemi a singolo agente hanno difficoltà. La sfida di design chiave è definire interfacce chiare tra agenti e assicurare che l'overhead della comunicazione inter-agente non annulli i benefici della specializzazione.
L'ecosistema di strumenti sta convergendo per supportare nativamente i pattern multi-agente. La composizione di sotto-grafi di LangGraph, il livello di orchestrazione agenti di LlamaIndex e CrewAI forniscono tutti primitive per costruire sistemi RAG multi-agente. Man mano che questi framework maturano e i costi continuano a scendere, il RAG multi-agente diventerà pratico per una gamma più ampia di applicazioni di produzione. I principi trattati in questa guida — gestione dello stato esplicita, valutazione della qualità del recupero, fallback eleganti e osservabilità completa — rimangono essenziali indipendentemente dal fatto che si schieri un singolo agente o un team di agenti specializzati.