LangChain pro RAG: Advanced Framework and Patterns
LangChain stal se referenčním rámcem pro vytváření aplikací založené na LLM. S více než 80 000 hvězdičkami na GitHubu a rychle rostoucí komunitou, nabízí výkonné abstrakce pro každou komponentu systému RAG: nakladače dokumentů, text štípačky, vkládací modely, vektorové obchody, retrívry a řetězy. Ale jeho skutečná síla objeví se, když zkombinujete tyto stavební kameny do pokročilých vzorů.
V tomto článku vytvoříme kompletní RAG systémy s LangChain: začneme od základní potrubí až po pokročilé vzory jako např konverzační RAG (kontextová paměť mezi po sobě jdoucími otázkami), víceskokové vyhledávání (dotazy, které vyžadují více kroků uvažování), volání nástroje (agenti, kteří rozhodují o tom, který zdroj konzultovat) a vyhledávání pomocí vlastního dotazu (automatické sémantické filtrování metadat). Vše s příklady spustitelného kódu.
Co se naučíte
- Architektura LangChain: řetězce, runnables a LCEL (LangChain Expression Language)
- Základní RAG potrubí s LangChain: od dokumentace k reakci
- Konverzační RAG: kontextová paměť a správa historie
- Multi-hop vyhledávání pro otázky vyžadující vícekrokové uvažování
- Načítání pomocí vlastního dotazu: Automatické filtrování metadat z dotazu
- Ensemble retriever a hybrid search v LangChain
- Streamování odpovědí pro lepší uživatelské prostředí ve výrobě
- Ladění a testování potrubí LangChain pomocí LangSmith
1. LangChain Expression Language (LCEL)
Počínaje verzí 0.1.0 představil LangChain LangChain výraz
jazyk (LCEL): deklarativní syntaxe založená na svislém vzoru (|)
skládat řetězce čitelným a typově bezpečným způsobem. LCEL je optimalizován pro streamování,
paralelismus a trasování a je moderním způsobem budování plynovodů LangChain.
LCEL Basics: Pipe Operators and 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 pro více kontextů
Jedním z potenciálů LCEL je paralelní složení: lze je obnovit kontexty z různých zdrojů paralelně a zkombinujte je, než je předáte LLM.
RunnableParallel: Načítání z více zdrojů
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. Dokončete základní potrubí RAG Pipeline
Než se pustíme do pokročilých vzorů, postavme kompletní a robustní potrubí RAG s LangChain: od příjmu dokumentu přes vyhledání až po generování odpovědi.
RAG Pipeline Kompletní s 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. Konverzační RAG: Kontextová paměť
Problém základního RAG je, že každý dotaz je zpracován nezávisle. V jednom skutečné konverzace, uživatel očekává, že si systém zapamatuje kontext předchozí otázky. "A druhá možnost?" Bez znalosti o co jde, to nedává smysl mluvil. The Konverzační RAG řeší tento problém.
LangChain řídí konverzaci ve dvou krocích:
- Reformulace dotazu: vzhledem k historii chatu přeformulujte aktuální otázku na samostatný dotaz, který obsahuje veškerý kontext nezbytný pro vyhledání
- RAG s historií: použijte k vyhledání přeformulovaný dotaz a poté vygenerujte odpověď poskytující načtený kontext i historii chatu
Konverzační RAG s LangChainem
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: Automatické filtrování metadat
Il Self-Query Retrieval je jedním z nejsilnějších vzorů v LangChain: umožňuje LLM interpretovat přirozený dotaz uživatele a extrahovat jej automaticky jak sémantický dotaz, tak filtry metadat. Uživatel píše „2024 článků o RAG napsaných odborníky“ a systém automaticky extrahuje filtr roku=2024 a typ filter="expert".
Self-Query načítání s metadaty
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 vyhledávání pro komplexní dotazy
Některé otázky vyžadují několik kroků uvažování: „Kdo vytvořil model používá LangChain standardně a kdy byl založen?" vyžaduje před nalezením že LangChain standardně používá OpenAI, pak najděte datum založení OpenAI. Tomu se říká víceskokové vyhledávání.
Multi-Hop RAG s Query Decomposition
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 a Hybrid Search
LangChain nabízí a EnsembleRetriever která kombinuje více retrívrů s konfigurovatelnými váhami, při použití Reciprocal Rank Fusion pro konečné pořadí. Je to nejjednodušší způsob, jak implementovat hybridní vyhledávání (BM25 + vektor) v 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: Sledování a ladění
LangSmith je platforma pozorovatelnosti pro LangChain. Umožňuje pro zobrazení každého kroku řetězce, výzev odeslaných do LLM, načtených dokumentů, latence a náklady. Je nezbytný pro ladění a sledování vývoje ve výrobě.
Nastavení LangSmith pro sledování
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. Streamování odpovědí pro lepší UX
Ve výrobě mohou odpovědi LLM trvat 5–15 sekund. Ukažte slova jak jsou generovány (streaming), výrazně zlepšuje vnímání rychlost uživatelem. LCEL podporuje streamování nativně.
Streamování pomocí FastAPI a 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. LangChain Best Practices a Anti-Patterns
Nejlepší postupy
- Vždy používejte LCEL namísto starších řetězců (LLMCain, RetrievalQA). LCEL je výkonnější, typově bezpečnější a podporuje nativní streamování.
- Povolit LangSmith ve vývoji: Automatické trasování šetří hodiny ladění. Můžete jej deaktivovat ve výrobě, abyste ušetřili náklady.
- MMR pro rozmanitost: Použijte maximální okrajovou relevanci (search_type="mmr") místo čisté podobnosti, abyste zabránili načítání téměř identických kusů.
- asynchronní/čekání na propustnost: Použijte ainvoke a astream pro I/O operace (LLM, vektorová DB). Umožňuje vám zpracovávat souběžné požadavky bez režie vlákna.
- Oddělte logiku vyhledávání od generování: umožňuje testovat kód, umožňuje vám zesměšňovat retrívra v testech.
Anti-vzory, kterým je třeba se vyhnout
- Řetězy zapuštěné příliš hluboko: LangChain umožňuje skládat velmi složité řetězce. Za 3-4 úrovněmi vnoření je obtížné ladit. Zvažte rozdělení řetězce na funkce.
- Ignorujte náklady na tokeny: Každý dokument v kontextu zvyšuje náklady. Změřte a optimalizujte počet tokenů odeslaných do LLM.
- Šablona výzvy bez verzování: Výzvy jsou kód. Verujte je, testujte je a sledujte změny jako u jakékoli jiné komponenty.
- LLM vysoké teploty pro RAG: pro použití RAG teploty 0,0-0,2. Vysoké teploty zvyšují variabilitu, nikoli kvalitu, a mají tendenci zvyšovat halucinace.
Závěry
LangChain transformuje složitost systému RAG do řady stavebních bloků modulární. Postavili jsme potrubí od nejjednodušších (základní RAG s LCEL) až po pokročilejší (konverzační RAG, multi-hop, self-query), dotýkající se každého aspektu relevantní pro produkci: streamování, sledování pomocí LangSmith, hybridní vyhledávání a osvědčené postupy pro kvalitu.
Klíčové body:
- LCEL je moderní způsob vytváření řetězců: čitelný, typově bezpečný, streaming-native
- Konverzační RAG vyžaduje před načtením přeformulování dotazu
- Načítání pomocí vlastního dotazu automatizuje filtrování metadat z přirozeného dotazu
- Víceskokové vyhledávání rozkládá složité dotazy na sekvenční dílčí dotazy
- EnsembleRetriever kombinuje BM25 + dense s jediným příkazem
- LangSmith je nezbytný pro ladění a hodnocení ve výrobě
V příštím článku prozkoumáme Správa kontextových oken: jak spravovat a optimalizovat rozpočet tokenů LLM, když je k dispozici kontext přesahuje možnosti modelu.