LangChain pentru RAG: Cadru și modele avansate
LangChain a devenit cadrul de referință pentru aplicații de construcție bazat pe LLM. Cu peste 80.000 de stele pe GitHub și o comunitate în creștere rapidă, oferă abstracții puternice pentru fiecare componentă a unui sistem RAG: încărcătoare de documente, text splittere, modele de încorporare, magazine de vectori, retrievers și lanțuri. Dar adevărata ei putere apare atunci când combinați aceste blocuri de construcție în modele avansate.
În acest articol vom construi sisteme RAG complete cu LangChain: vom începe de la conductă de bază până la modele avansate, cum ar fi RAG conversațional (memoria contextuală între întrebări consecutive), the recuperare multi-hop (interogări care necesită mai mulți pași de raționament), the chemarea instrumentului (agenți care decid ce sursă să consulte) și regăsire auto-interogare (filtrarea semantică automată a metadatelor). Toate cu exemple de cod executabil.
Ce vei învăța
- Arhitectura LangChain: lanțuri, rulabile și LCEL (LangChain Expression Language)
- Conducta RAG de bază cu LangChain: de la documentare la răspuns
- RAG conversațional: memorie contextuală și management al istoriei
- Recuperare multi-hop pentru întrebări care necesită raționament în mai mulți pași
- Preluare automată a interogării: filtrarea automată a metadatelor din interogare
- Ansamblu retriever și căutare hibridă în LangChain
- Streaming răspunsuri pentru o mai bună UX în producție
- Depanarea și testarea conductelor LangChain cu LangSmith
1. LangChain Expression Language (LCEL)
Începând cu versiunea 0.1.0, LangChain a introdus Expresia LangChain
Limbă (LCEL): o sintaxă declarativă bazată pe modelul pipe (|)
pentru a compune lanțuri într-un mod lizibil și sigur de tipare. LCEL este optimizat pentru streaming,
paralelism și urmărire și este modalitatea modernă de a construi conducte LangChain.
Noțiuni de bază LCEL: Operatori de conducte și elemente de rulare
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 pentru mai multe contexte
Unul dintre potențialele LCEL este compoziția paralelă: pot fi recuperate contexte din surse diferite în paralel și combinați-le înainte de a le trece la LLM.
RunnableParallel: Preluare din mai multe surse
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. Conducta RAG de bază completă
Înainte de a aborda modelele avansate, să construim o conductă RAG completă și robustă cu LangChain: de la ingerarea documentelor la regăsire până la generarea răspunsului.
RAG Pipeline complet cu 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. RAG conversațional: Memoria contextuală
Problema cu RAG de bază este că fiecare interogare este tratată independent. Într-una conversație reală, utilizatorul se așteaptă ca sistemul să-și amintească contextul întrebări anterioare. — Și a doua variantă? Nu are sens fără să știi despre ce este vorba el vorbea. The RAG conversațional rezolvă această problemă.
LangChain gestionează conversația în doi pași:
- Reformularea interogării: având în vedere istoricul chat-ului, reformulați întrebarea curentă într-o interogare independentă care conține tot contextul necesar pentru regăsire
- RAG cu istorie: utilizați interogarea reformulată pentru extragere, apoi generați răspunsul furnizând atât contextul preluat, cât și istoricul chat-ului
RAG conversațional cu 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. Preluare automată a interogării: filtrarea automată a metadatelor
Il Recuperare auto-interogare este unul dintre cele mai puternice modele din LangChain: permite LLM să interpreteze interogarea și extragerea naturală a utilizatorului automat atât interogarea semantică, cât și filtrele de metadate. Utilizatorul scrie „2024 articole despre RAG scrise de experți” și sistemul extrage automat filtrul de an=2024 și tipul filter="expert".
Recuperare auto-interogare cu metadate
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. Recuperare multi-hop pentru interogări complexe
Unele întrebări necesită mai mulți pași de raționament: „Cine a dezvoltat modelul folosit de LangChain în mod implicit și când a fost fondat?" necesită înainte de a găsi că LangChain folosește OpenAI în mod implicit, apoi găsiți data înființării OpenAI. Aceasta se numește recuperare multi-hop.
Multi-Hop RAG cu descompunere de interogări
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 și Hybrid Search
LangChain oferă o EnsembleRetriever care combină mai mulți retrievers cu ponderi configurabile, aplicând Reciprocal Rank Fusion pentru clasamentul final. Este cel mai simplu mod de a implementa căutarea hibridă (BM25 + vector) în 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: Urmărire și depanare
LangSmith este platforma de observabilitate pentru LangChain. Permite pentru a vizualiza fiecare pas al lanțului, solicitările trimise la LLM, documentele preluate, latențe și costuri. Este esențial pentru depanarea și monitorizarea dezvoltării in productie.
Configurare LangSmith pentru urmărire
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. Transmiteți răspunsuri pentru o mai bună UX
În producție, răspunsurile LLM pot dura 5-15 secunde. Arată cuvintele pe măsură ce sunt generate (streaming) îmbunătățește drastic percepția asupra viteza de către utilizator. LCEL acceptă streaming în mod nativ.
Streaming cu FastAPI și evenimente trimise de server
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. Cele mai bune practici și anti-modele LangChain
Cele mai bune practici
- Utilizați întotdeauna LCEL în loc de lanțuri moștenite (LLMCain, RetrievalQA). LCEL este mai performant, mai sigur de tip și acceptă streaming nativ.
- Activați LangSmith în dezvoltare: Urmărirea automată economisește ore de depanare. Puteți să-l dezactivați în producție pentru a economisi costuri.
- MMR pentru diversitate: Folosiți Relevanța marginală maximă (search_type="mmr") în loc de similitudinea pură pentru a preveni retriever-ul să recupereze bucăți aproape identice.
- asincron/așteptați debitul: Utilizați ainvoke și astream pentru operațiuni I/O (LLM, vector DB). Vă permite să gestionați cereri concurente fără supraîncărcare a firului.
- Separați logica de recuperare de generație: face codul testabil, vă permite să batjocoriți retrieverul în teste.
Anti-modele de evitat
- Lanțurile se cuibăreau prea adânc: LangChain vă permite să compuneți lanțuri foarte complexe. Dincolo de 3-4 niveluri de cuib devine dificil de depanat. Luați în considerare împărțirea lanțului în funcții.
- Ignorați costurile cu simboluri: Fiecare document în context crește costul. Măsurați și optimizați numărul de jetoane trimise la LLM.
- Șablon prompt fără versiune: prompturile sunt cod. Versiune-le, testează-le și urmărește modificările ca orice altă componentă.
- LLM temperaturi ridicate pentru RAG: pentru RAG folosiți temperaturi 0,0-0,2. Temperaturile ridicate cresc variabilitatea, nu calitatea și tind să crească halucinațiile.
Concluzii
LangChain transformă complexitatea unui sistem RAG într-o serie de blocuri de construcție modulare. Am construit conducte de la cele mai simple (RAG de bază cu LCEL) până la mai avansat (RAG conversațional, multi-hop, auto-interogare), atingând fiecare aspect relevante pentru producție: streaming, urmărire cu LangSmith, căutare hibridă și cele mai bune practici pentru calitate.
Punctele cheie:
- LCEL este modalitatea modernă de a compune lanțuri: lizibil, sigur de tip, nativ de streaming
- RAG conversațional necesită reformularea interogării înainte de recuperare
- Recuperarea automată a interogării automatizează filtrarea metadatelor din interogarea naturală
- Recuperarea multi-hop descompune interogările complexe în subinterogări secvenţiale
- EnsembleRetriever combină BM25 + dense cu o singură comandă
- LangSmith este esențial pentru depanare și evaluare în producție
În următorul articol vom explora Gestionarea ferestrelor de context: cum să gestionați și să optimizați bugetul de simbol LLM atunci când contextul este disponibil depășește capacitățile modelului.