LangChain per RAG: Framework e Pattern Avanzati
LangChain è diventato il framework di riferimento per costruire applicazioni basate su LLM. Con oltre 80.000 stelle su GitHub e una community in rapida crescita, offre astrazioni potenti per ogni componente di un sistema RAG: document loaders, text splitters, embedding models, vector stores, retrievers e chains. Ma la sua vera potenza emerge quando si combinano questi building block in pattern avanzati.
In questo articolo costruiremo sistemi RAG completi con LangChain: partiremo dalla pipeline base fino ad arrivare a pattern avanzati come il conversational RAG (memoria contestuale tra domande consecutive), il multi-hop retrieval (query che richiedono più passaggi di reasoning), il tool calling (agenti che decidono quale fonte consultare) e il self-query retrieval (filtraggio semantico automatico dei metadati). Tutto con esempi di codice eseguibili.
Cosa Imparerai
- Architettura di LangChain: chains, runnables e LCEL (LangChain Expression Language)
- Pipeline RAG base con LangChain: dalla documentazione alla risposta
- Conversational RAG: memoria contestuale e history management
- Multi-hop retrieval per domande che richiedono ragionamento a più passi
- Self-query retrieval: filtraggio automatico dei metadati dalla query
- Ensemble retriever e hybrid search in LangChain
- Streaming responses per migliore UX in produzione
- Debug e testing di pipeline LangChain con LangSmith
1. LangChain Expression Language (LCEL)
A partire dalla versione 0.1.0, LangChain ha introdotto il LangChain Expression
Language (LCEL): una sintassi dichiarativa basata sul pattern pipe (|)
per comporre chain in modo leggibile e type-safe. LCEL è ottimizzato per streaming,
parallelismo e tracing, ed è il modo moderno di costruire pipeline LangChain.
LCEL Basics: Pipe Operator e Runnables
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_qdrant import QdrantVectorStore
# Setup componenti base
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# Prompt template per RAG
rag_prompt = ChatPromptTemplate.from_template("""
Sei un assistente tecnico esperto. Rispondi alla domanda basandoti SOLO sul contesto
fornito. Se il contesto non contiene informazioni sufficienti, dillo esplicitamente.
Contesto:
{context}
Domanda: {question}
Risposta:""")
# Vector store (assumendo Qdrant in locale)
vectorstore = QdrantVectorStore.from_existing_collection(
embedding=embeddings,
url="http://localhost:6333",
collection_name="rag_docs"
)
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 5}
)
# LCEL Pipeline - sintassi pipe
def format_docs(docs):
"""Formatta i documenti recuperati come stringa di contesto"""
return "\n\n---\n\n".join(
f"[Fonte: {doc.metadata.get('source', 'N/A')}]\n{doc.page_content}"
for doc in docs
)
# Pipeline con LCEL
rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| rag_prompt
| llm
| StrOutputParser()
)
# Invocazione
answer = rag_chain.invoke("Cos'è il RAG e quali problemi risolve?")
print(answer)
# Streaming (importante per UX in produzione!)
for chunk in rag_chain.stream("Quali sono i principali vector database?"):
print(chunk, end="", flush=True)
1.1 RunnableParallel per Context Multipli
Una delle potenzialita di LCEL è la composizione parallela: si possono recuperare contesti da fonti diverse in parallelo e combinarli prima di passarli all'LLM.
RunnableParallel: Retrieval da Fonti Multiple
from langchain_core.runnables import RunnableParallel
# Due retriever diversi: documentazione tecnica e FAQ
tech_retriever = tech_vectorstore.as_retriever(search_kwargs={"k": 3})
faq_retriever = faq_vectorstore.as_retriever(search_kwargs={"k": 2})
# Pipeline con retrieval parallelo
multi_source_chain = (
RunnableParallel(
tech_context=tech_retriever | format_docs,
faq_context=faq_retriever | format_docs,
question=RunnablePassthrough()
)
| ChatPromptTemplate.from_template("""
Domanda: {question}
Documentazione Tecnica:
{tech_context}
FAQ:
{faq_context}
Risposta basata su entrambe le fonti:""")
| llm
| StrOutputParser()
)
answer = multi_source_chain.invoke("Come si configura l'autenticazione?")
2. Pipeline RAG Base Completa
Prima di affrontare i pattern avanzati, costruiamo una pipeline RAG completa e robusta con LangChain: dall'ingestion dei documenti al retrieval alla generazione della risposta.
Pipeline RAG Completa con LangChain
from langchain_community.document_loaders import (
PyPDFLoader, TextLoader, WebBaseLoader,
DirectoryLoader, UnstructuredMarkdownLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_qdrant import QdrantVectorStore
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from pathlib import Path
from typing import List
import logging
logger = logging.getLogger(__name__)
class LangChainRAGSystem:
"""Sistema RAG completo con LangChain"""
def __init__(
self,
collection_name: str = "rag_docs",
embedding_model: str = "text-embedding-3-small",
llm_model: str = "gpt-4o-mini"
):
self.embeddings = OpenAIEmbeddings(model=embedding_model)
self.llm = ChatOpenAI(model=llm_model, temperature=0.1)
self.collection_name = collection_name
# Text splitter ottimizzato per RAG
self.text_splitter = RecursiveCharacterTextSplitter(
chunk_size=512,
chunk_overlap=64,
separators=["\n\n", "\n", ". ", "! ", "? ", " "],
add_start_index=True # salva posizione nel documento originale
)
# Inizializza o connetti al vector store
self.vectorstore = self._init_vectorstore()
self.retriever = self.vectorstore.as_retriever(
search_type="mmr", # Maximum Marginal Relevance per diversità
search_kwargs={
"k": 5,
"fetch_k": 20, # recupera 20, poi MMR seleziona 5 diversi
"lambda_mult": 0.7 # 0=massima diversità, 1=massima similarità
}
)
# Prompt RAG
self.prompt = ChatPromptTemplate.from_template("""
Sei un assistente tecnico preciso. Rispondi alla domanda basandoti ESCLUSIVAMENTE
sul contesto fornito. Non inventare informazioni non presenti nel contesto.
Se il contesto non è sufficiente per rispondere completamente, dillo esplicitamente
e rispondi solo sulla parte coperta dal contesto.
Contesto:
{context}
Domanda: {question}
Risposta:""")
# Chain LCEL
self.chain = self._build_chain()
def _init_vectorstore(self):
"""Inizializza il vector store"""
try:
return QdrantVectorStore.from_existing_collection(
embedding=self.embeddings,
url="http://localhost:6333",
collection_name=self.collection_name
)
except Exception:
# Crea collection se non esiste
return QdrantVectorStore.from_documents(
documents=[],
embedding=self.embeddings,
url="http://localhost:6333",
collection_name=self.collection_name
)
def _build_chain(self):
"""Costruisce la chain LCEL"""
def format_docs(docs):
formatted = []
for i, doc in enumerate(docs, 1):
source = doc.metadata.get("source", "N/A")
page = doc.metadata.get("page", "")
header = f"[Fonte {i}: {source}{f', p.{page}' if page else ''}]"
formatted.append(f"{header}\n{doc.page_content}")
return "\n\n---\n\n".join(formatted)
return (
{"context": self.retriever | format_docs, "question": RunnablePassthrough()}
| self.prompt
| self.llm
| StrOutputParser()
)
def ingest_pdf(self, pdf_path: str) -> int:
"""Ingesta un PDF nel sistema RAG"""
loader = PyPDFLoader(pdf_path)
documents = loader.load()
chunks = self.text_splitter.split_documents(documents)
# Aggiungi metadati
for chunk in chunks:
chunk.metadata["ingested_at"] = str(Path(pdf_path).stat().st_mtime)
chunk.metadata["doc_type"] = "pdf"
self.vectorstore.add_documents(chunks)
logger.info(f"Ingested {len(chunks)} chunks from {pdf_path}")
return len(chunks)
def ingest_directory(self, directory: str, glob: str = "**/*.txt") -> int:
"""Ingesta tutti i file in una directory"""
loader = DirectoryLoader(
directory,
glob=glob,
loader_cls=TextLoader,
loader_kwargs={"encoding": "utf-8"},
show_progress=True
)
documents = loader.load()
chunks = self.text_splitter.split_documents(documents)
self.vectorstore.add_documents(chunks)
return len(chunks)
def query(self, question: str) -> str:
"""Risponde a una domanda"""
return self.chain.invoke(question)
def query_with_sources(self, question: str) -> dict:
"""Risponde e restituisce anche le fonti"""
from langchain.chains import RetrievalQAWithSourcesChain
docs = self.retriever.invoke(question)
answer = self.chain.invoke(question)
sources = list(set(
doc.metadata.get("source", "N/A") for doc in docs
))
return {
"answer": answer,
"sources": sources,
"num_docs": len(docs)
}
3. Conversational RAG: Memoria Contestuale
Il problema del RAG base è che ogni query è trattata in modo indipendente. In una conversazione reale, l'utente si aspetta che il sistema ricordi il contesto delle domande precedenti. "E la seconda opzione?" non ha senso senza sapere di cosa si stava parlando. Il Conversational RAG risolve questo problema.
LangChain gestisce la conversazione in due step:
- Query reformulation: data la storia della chat, riformula la domanda corrente in una query standalone che contenga tutto il contesto necessario per il retrieval
- RAG con storia: usa la query riformulata per il retrieval, poi genera la risposta fornendo sia il contesto recuperato sia la storia della chat
Conversational RAG con LangChain
from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from typing import Dict
class ConversationalRAG:
"""RAG conversazionale con memoria della chat history"""
def __init__(self, retriever, llm):
self.retriever = retriever
self.llm = llm
self.store: Dict[str, ChatMessageHistory] = {}
# Step 1: Prompt per riformulare la query usando la storia
contextualize_q_prompt = ChatPromptTemplate.from_messages([
("system", """Dato una storia della chat e l'ultima domanda dell'utente,
che potrebbe fare riferimento al contesto della chat, formula una domanda standalone
che sia comprensibile senza la storia della chat. NON rispondere alla domanda,
riformulala solo se necessario, altrimenti restituiscila com'e."""),
MessagesPlaceholder("chat_history"),
("human", "{input}")
])
# Retriever history-aware: riformula la query prima del retrieval
self.history_aware_retriever = create_history_aware_retriever(
llm, retriever, contextualize_q_prompt
)
# Step 2: Prompt per la risposta con contesto e storia
qa_prompt = ChatPromptTemplate.from_messages([
("system", """Sei un assistente tecnico preciso. Rispondi alla domanda
basandoti sul contesto fornito e sulla storia della conversazione.
Se il contesto non contiene la risposta, dillo chiaramente.
Contesto:
{context}"""),
MessagesPlaceholder("chat_history"),
("human", "{input}")
])
# Chain per combinare documenti e generare risposta
question_answer_chain = create_stuff_documents_chain(llm, qa_prompt)
# Chain RAG completa con history
self.rag_chain = create_retrieval_chain(
self.history_aware_retriever,
question_answer_chain
)
# Wrapper con gestione automatica della history
self.conversational_rag = RunnableWithMessageHistory(
self.rag_chain,
self._get_session_history,
input_messages_key="input",
history_messages_key="chat_history",
output_messages_key="answer"
)
def _get_session_history(self, session_id: str) -> BaseChatMessageHistory:
"""Ottieni o crea la history per una sessione"""
if session_id not in self.store:
self.store[session_id] = ChatMessageHistory()
return self.store[session_id]
def chat(self, message: str, session_id: str = "default") -> str:
"""Invia un messaggio nella conversazione"""
result = self.conversational_rag.invoke(
{"input": message},
config={"configurable": {"session_id": session_id}}
)
return result["answer"]
def get_history(self, session_id: str = "default") -> list:
"""Ottieni la storia della conversazione"""
if session_id not in self.store:
return []
return [
{"role": "human" if isinstance(m, HumanMessage) else "ai",
"content": m.content}
for m in self.store[session_id].messages
]
# Esempio di utilizzo
conv_rag = ConversationalRAG(retriever=retriever, llm=llm)
# Conversazione multi-turno
responses = []
questions = [
"Cos'è LangChain?",
"Quali sono i suoi componenti principali?", # "suoi" si riferisce a LangChain
"Quale di questi è il più importante per il RAG?" # "questi" = componenti citati prima
]
for q in questions:
answer = conv_rag.chat(q, session_id="user123")
print(f"Q: {q}")
print(f"A: {answer}\n")
4. Self-Query Retrieval: Filtraggio Automatico dei Metadati
Il Self-Query Retrieval è uno dei pattern più potenti di LangChain: permette all'LLM di interpretare la query naturale dell'utente e di estrarre automaticamente sia la query semantica che i filtri sui metadati. L'utente scrive "articoli del 2024 su RAG scritti da esperti" e il sistema estrae automaticamente il filtro per anno=2024 e il filtro per tipo="esperto".
Self-Query Retrieval con Metadati
from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo
from langchain_openai import ChatOpenAI
from langchain_qdrant import QdrantVectorStore
# Descrivi i metadati disponibili nel vector store
metadata_field_info = [
AttributeInfo(
name="source",
description="Il file o URL sorgente del documento",
type="string",
),
AttributeInfo(
name="author",
description="L'autore del documento o articolo",
type="string",
),
AttributeInfo(
name="year",
description="L'anno di pubblicazione del documento (e.g. 2023, 2024)",
type="integer",
),
AttributeInfo(
name="category",
description="La categoria del contenuto (e.g. 'tutorial', 'paper', 'documentation')",
type="string",
),
AttributeInfo(
name="difficulty",
description="Il livello di difficolta (beginner, intermediate, advanced)",
type="string",
),
]
# Descrizione del documento per guidare il query constructor
document_content_description = """
Articoli tecnici e documentazione su AI engineering, RAG, LLM, embedding,
vector databases e machine learning.
"""
# Self-Query Retriever
self_query_retriever = SelfQueryRetriever.from_llm(
llm=ChatOpenAI(model="gpt-4o-mini", temperature=0),
vectorstore=vectorstore,
document_contents=document_content_description,
metadata_field_info=metadata_field_info,
verbose=True, # mostra la query strutturata generata
search_kwargs={"k": 5}
)
# Query naturali con filtri impliciti
examples = [
"Tutorial su RAG del 2024 per principianti",
"Paper avanzati su embedding scritti da Reimers",
"Documentazione su Qdrant o Pinecone"
]
for query in examples:
print(f"\nQuery: {query}")
docs = self_query_retriever.invoke(query)
print(f"Trovati: {len(docs)} documenti")
for doc in docs:
print(f" - {doc.metadata.get('source', 'N/A')} ({doc.metadata.get('year', 'N/A')})")
5. Multi-Hop Retrieval per Domande Complesse
Alcune domande richiedono più passaggi di reasoning: "Chi ha sviluppato il modello usato da LangChain per default e quando è stato fondato?" richiede prima di trovare che LangChain usa OpenAI di default, poi di trovare la data di fondazione di OpenAI. Questo si chiama multi-hop retrieval.
Multi-Hop RAG con Decomposizione della Query
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from typing import List
class MultiHopRAG:
"""RAG con decomposizione della query in sub-query"""
def __init__(self, retriever, llm):
self.retriever = retriever
self.llm = llm
# Chain per decomporre la query in sub-query
self.decompose_chain = (
ChatPromptTemplate.from_template("""
Decomponi questa domanda complessa in 2-4 sotto-domande più semplici che,
rispondendo in sequenza, permettono di rispondere alla domanda originale.
Domanda originale: {question}
Fornisci le sotto-domande come lista numerata, una per riga.
Solo la lista, niente altro.""")
| llm
| StrOutputParser()
)
# Chain per la risposta finale con tutti i contesti
self.answer_chain = (
ChatPromptTemplate.from_template("""
Hai ricevuto informazioni da più passaggi di ricerca per rispondere alla domanda.
Sintetizza queste informazioni in una risposta coerente e completa.
Domanda originale: {original_question}
Informazioni raccolte:
{gathered_info}
Risposta sintetica:""")
| llm
| StrOutputParser()
)
def _parse_subquestions(self, text: str) -> List[str]:
"""Estrae le sotto-domande dalla risposta del LLM"""
lines = text.strip().split('\n')
subquestions = []
for line in lines:
line = line.strip()
if line and (line[0].isdigit() or line.startswith('-')):
# Rimuovi numerazione o bullet
clean = line.lstrip('0123456789.-) ').strip()
if clean:
subquestions.append(clean)
return subquestions
def multi_hop_query(self, question: str) -> dict:
"""Esegui multi-hop retrieval con decomposizione della query"""
print(f"Domanda originale: {question}\n")
# Step 1: Decomposizione
subquestions_text = self.decompose_chain.invoke({"question": question})
subquestions = self._parse_subquestions(subquestions_text)
print(f"Sub-queries generate: {len(subquestions)}")
# Step 2: Retrieval e risposta per ogni sub-query
gathered_info = []
all_sources = []
for i, subq in enumerate(subquestions, 1):
print(f" Hop {i}: {subq}")
docs = self.retriever.invoke(subq)
context = "\n".join(doc.page_content for doc in docs[:3])
# Risposta parziale per questa sub-query
partial_answer = self.llm.invoke(
f"Contesto: {context}\nDomanda: {subq}\nRisposta breve:"
).content
gathered_info.append(f"Sotto-domanda {i}: {subq}\nRisposta: {partial_answer}")
all_sources.extend(doc.metadata.get("source", "") for doc in docs)
# Step 3: Sintesi finale
final_answer = self.answer_chain.invoke({
"original_question": question,
"gathered_info": "\n\n".join(gathered_info)
})
return {
"answer": final_answer,
"subquestions": subquestions,
"num_hops": len(subquestions),
"sources": list(set(s for s in all_sources if s))
}
6. Ensemble Retriever e Hybrid Search
LangChain offre un EnsembleRetriever che combina più retriever con pesi configurabili, applicando Reciprocal Rank Fusion per il ranking finale. È il modo più semplice per implementare hybrid search (BM25 + vector) in LangChain.
Ensemble Retriever: BM25 + Dense
from langchain.retrievers import EnsembleRetriever, BM25Retriever
from langchain_community.vectorstores import Qdrant
# BM25 retriever per ricerca keyword
bm25_retriever = BM25Retriever.from_documents(
documents, # lista di Document objects
k=5
)
# Dense retriever per ricerca semantica
dense_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
# Ensemble con pesi: 40% BM25, 60% dense
ensemble_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, dense_retriever],
weights=[0.4, 0.6]
# weights controlla l'importanza relativa dei due retriever
# nel Reciprocal Rank Fusion
)
# Uso normale - interfaccia identica a qualsiasi retriever
docs = ensemble_retriever.invoke("Come si implementa il reranking?")
# Integrazione nella chain LCEL
hybrid_rag_chain = (
{"context": ensemble_retriever | format_docs, "question": RunnablePassthrough()}
| rag_prompt
| llm
| StrOutputParser()
)
answer = hybrid_rag_chain.invoke("Tutorial BM25 + vector search")
7. LangSmith: Tracing e Debugging
LangSmith è la piattaforma di observability per LangChain. Permette di visualizzare ogni step della chain, i prompt inviati all'LLM, i documenti recuperati, le latenze e i costi. È fondamentale per il debugging in sviluppo e per il monitoring in produzione.
Setup LangSmith per Tracing
import os
from langsmith import Client
# Configura LangSmith (opzionale ma fortemente consigliato in produzione)
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "rag-production"
# Ora tutte le invocazioni delle chain vengono automaticamente tracciate!
# Visita app.langchain.com per vedere i trace
# Valutazione con LangSmith Evaluators
from langsmith.evaluation import evaluate as ls_evaluate
from langsmith.schemas import Run, Example
def faithfulness_evaluator(run: Run, example: Example) -> dict:
"""Valutatore personalizzato per faithfulness"""
answer = run.outputs.get("answer", "")
context = run.outputs.get("context", "")
ground_truth = example.outputs.get("answer", "")
# Usa un LLM come giudice
judge = ChatOpenAI(model="gpt-4o-mini", temperature=0)
score = judge.invoke(
f"""Su scala 0-1, quanto la seguente risposta è supportata dal contesto?
Risposta: {answer}
Contesto: {context[:500]}
Rispondi SOLO con un numero tra 0 e 1."""
).content
try:
return {"score": float(score.strip()), "key": "faithfulness"}
except:
return {"score": 0.5, "key": "faithfulness"}
# Dataset di test su LangSmith
client = Client()
# Crea dataset (solo la prima volta)
dataset = client.create_dataset(
"rag-evaluation",
description="Dataset per valutazione sistema RAG"
)
# Aggiungi esempi
examples = [
{
"inputs": {"question": "Cos'è LangChain?", "query": "Cos'è LangChain?"},
"outputs": {"answer": "LangChain è un framework per costruire applicazioni LLM"}
},
# ... altri esempi
]
# Valuta la chain sul dataset
results = ls_evaluate(
lambda inputs: rag_chain.invoke(inputs["question"]),
data="rag-evaluation",
evaluators=[faithfulness_evaluator],
experiment_prefix="v1-baseline"
)
8. Streaming Responses per una Migliore UX
In produzione, le risposte degli LLM possono richiedere 5-15 secondi. Mostrare le parole man mano che vengono generate (streaming) migliora drasticamente la percezione di velocità da parte dell'utente. LCEL supporta lo streaming nativamente.
Streaming con FastAPI e Server-Sent Events
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
import asyncio
app = FastAPI()
# Versione async della chain per streaming
async_rag_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| rag_prompt
| llm # llm supporta streaming nativo
| StrOutputParser()
)
@app.get("/rag/stream")
async def stream_rag(question: str):
"""Endpoint con streaming via Server-Sent Events"""
async def generate():
# Recupera i documenti prima (non streamable)
docs = await retriever.ainvoke(question)
context = format_docs(docs)
# Stream della generazione LLM
async for chunk in llm.astream(
rag_prompt.format_messages(
context=context,
question=question
)
):
if chunk.content:
# Formato SSE
yield f"data: {chunk.content}\n\n"
yield "data: [DONE]\n\n"
return StreamingResponse(
generate(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no" # Disabilita buffer nginx
}
)
@app.post("/rag/query")
async def query_rag(question: str):
"""Endpoint normale (non streaming)"""
answer = await async_rag_chain.ainvoke(question)
return {"answer": answer}
9. Best Practices e Anti-Pattern LangChain
Best Practices
- Usa sempre LCEL invece delle chain legacy (LLMChain, RetrievalQA). LCEL è più performante, type-safe e supporta streaming nativo.
- Abilita LangSmith in sviluppo: il tracing automatico risparmia ore di debugging. Puoi disabilitarlo in produzione per risparmiare sui costi.
- MMR per diversità: usa Maximum Marginal Relevance (search_type="mmr") invece di pure similarity per evitare che il retriever recuperi chunk quasi identici.
- async/await per throughput: usa ainvoke e astream per le operazioni I/O (LLM, vector DB). Permette di gestire richieste concorrenti senza thread overhead.
- Separa la logica di retrieval dalla generazione: rende il codice testabile, permette di mockare il retriever nei test.
Anti-Pattern da Evitare
- Chain annidate troppo profondamente: LangChain permette di comporre catene molto complesse. Oltre 3-4 livelli di nesting diventa difficile da debuggare. Considera di spezzare la chain in funzioni.
- Ignorare i costi dei token: ogni documento nel contesto aumenta il costo. Misura e ottimizza il numero di tokens inviati all'LLM.
- Prompt template senza versioning: i prompt sono codice. Versionali, testali e tracciane i cambiamenti come faresti con qualsiasi altro componente.
- LLM temperature alta per RAG: per RAG usa temperature 0.0-0.2. Temperature alta aumenta la variabilità, non la qualità, e tende ad aumentare le allucinazioni.
Conclusioni
LangChain trasforma la complessità di un sistema RAG in una serie di building block componibili. Abbiamo costruito pipeline dalla più semplice (RAG base con LCEL) alla più avanzata (conversational RAG, multi-hop, self-query), toccando ogni aspetto rilevante per la produzione: streaming, tracing con LangSmith, hybrid search e best practices per la qualità.
I punti chiave:
- LCEL è il modo moderno di comporre chain: leggibile, type-safe, streaming-native
- Conversational RAG richiede query reformulation prima del retrieval
- Self-query retrieval automatizza il filtraggio dei metadati dalla query naturale
- Multi-hop retrieval decompone domande complesse in sub-query sequenziali
- EnsembleRetriever combina BM25 + dense con un solo comando
- LangSmith è fondamentale per debugging e evaluation in produzione
Nel prossimo articolo esploreremo il Context Window Management: come gestire e ottimizzare il budget di token dell'LLM quando il contesto disponibile supera le capacità del modello.