LangChain dla RAG: zaawansowane ramy i wzorce
LangChain stał się platformą odniesienia do tworzenia aplikacji w oparciu o LLM. Z ponad 80 000 gwiazdami na GitHubie i szybko rosnącą społecznością, oferuje potężne abstrakcje dla każdego komponentu systemu RAG: moduły ładujące dokumenty, tekst splittery, modele osadzające, magazyny wektorów, retrievery i łańcuchy. Ale to prawdziwa moc pojawia się, gdy połączysz te elementy składowe w zaawansowane wzorce.
W tym artykule zbudujemy kompletne systemy RAG z LangChain: zaczniemy od podstawowy potok aż do zaawansowanych wzorców, takich jak konwersacyjny RAG (pamięć kontekstowa pomiędzy kolejnymi pytaniami), pobieranie wieloskokowe (zapytania wymagające wielu etapów rozumowania), the wywołanie narzędzia (agenci, którzy decydują, z którego źródła skorzystać) i samodzielne pobieranie zapytań (automatyczne semantyczne filtrowanie metadanych). Wszystko z przykładami kodu wykonywalnego.
Czego się nauczysz
- Architektura LangChain: łańcuchy, elementy uruchamialne i LCEL (język wyrażeń LangChain)
- Podstawowy potok RAG z LangChain: od dokumentacji do odpowiedzi
- Konwersacyjne RAG: zarządzanie pamięcią kontekstową i historią
- Wyszukiwanie metodą wielu przeskoków w przypadku pytań wymagających rozumowania wieloetapowego
- Samodzielne pobieranie zapytań: automatyczne filtrowanie metadanych z zapytania
- Retriever zespołowy i wyszukiwanie hybrydowe w LangChain
- Odpowiedzi strumieniowe dla lepszego UX w produkcji
- Debugowanie i testowanie potoków LangChain za pomocą LangSmith
1. Język wyrażeń LangChain (LCEL)
Począwszy od wersji 0.1.0, LangChain wprowadził Wyrażenie LangChaina
Język (LCEL): deklaratywna składnia oparta na wzorcu potoku (|)
do tworzenia łańcuchów w sposób czytelny i bezpieczny dla typu. LCEL jest zoptymalizowany do przesyłania strumieniowego,
równoległość i śledzenie oraz jest nowoczesnym sposobem budowania potoków LangChain.
Podstawy LCEL: operatorzy rur i elementy eksploatacyjne
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 dla wielu kontekstów
Jednym z potencjałów LCEL jest skład równoległy: można je odzyskać konteksty z różnych źródeł równolegle i łączyć je przed przekazaniem ich do LLM.
RunnableParallel: pobieranie z wielu źródeł
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. Kompletny podstawowy rurociąg RAG
Zanim zajmiemy się zaawansowanymi wzorcami, zbudujmy kompletny i solidny rurociąg RAG z LangChain: od przyjęcia dokumentu, poprzez jego wyszukanie, aż po wygenerowanie odpowiedzi.
Rurociąg RAG w komplecie z 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. Konwersacyjne RAG: pamięć kontekstowa
Problem z podstawowym RAG polega na tym, że każde zapytanie jest traktowane niezależnie. W jednym prawdziwej rozmowy, użytkownik oczekuje, że system zapamięta jej kontekst poprzednie pytania. – A druga opcja? Bez wiedzy o co chodzi, nie ma to sensu on mówił. The Rozmowa RAG rozwiązuje ten problem.
LangChain zarządza rozmową w dwóch krokach:
- Przeformułowanie zapytania: biorąc pod uwagę historię czatu, przeformułuj bieżące pytanie na samodzielne zapytanie zawierające cały kontekst niezbędny do pobrania
- RAG z historią: użyj przeformułowanego zapytania do pobrania, a następnie wygeneruj odpowiedź podając zarówno pobrany kontekst, jak i historię czatów
Konwersacyjny RAG z 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. Samodzielne pobieranie zapytań: automatyczne filtrowanie metadanych
Il Samodzielne pobieranie zapytań to jeden z najpotężniejszych wzorców w LangChain: pozwala LLM interpretować naturalne zapytanie użytkownika i wyodrębniać je automatycznie zarówno zapytanie semantyczne, jak i filtry metadanych. Użytkownik pisze „Artykuły na temat RAG z 2024 r. napisane przez ekspertów”, a system automatycznie wyodrębnia rok filter=2024 i typ filter="expert".
Samodzielne pobieranie zapytań z metadanymi
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. Pobieranie wieloprzeskokowe dla złożonych zapytań
Niektóre pytania wymagają wielu etapów rozumowania: „Kto opracował model domyślnie używany przez LangChain i kiedy został założony?” wymaga przed znalezieniem że LangChain domyślnie używa OpenAI, a następnie znajdź datę założenia OpenAI. To się nazywa pobieranie wieloskokowe.
Multi-Hop RAG z rozkładem zapytań
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. Retriever zespołowy i wyszukiwanie hybrydowe
LangChain oferuje Zespół Retrievera który łączy wiele retrieverów z konfigurowalnymi wagami, stosując Reciprocal Rank Fusion do ostatecznego rankingu. Jest to najprostszy sposób na wdrożenie wyszukiwania hybrydowego (BM25 + wektor) w LangChain.
Retriever zespołowy: BM25 + gęsty
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: Śledzenie i debugowanie
LangSmitha to platforma obserwowalności dla LangChain. Pozwala aby zobaczyć każdy etap łańcucha, monity wysłane do LLM, odzyskane dokumenty, opóźnienia i koszty. Jest to niezbędne do debugowania i monitorowania rozwoju w produkcji.
Konfiguracja LangSmitha do śledzenia
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. Przesyłanie strumieniowe odpowiedzi dla lepszego UX
W produkcji odpowiedzi LLM mogą zająć 5-15 sekund. Pokaż słowa w miarę ich generowania (streamingu) drastycznie poprawia się percepcja prędkość przez użytkownika. LCEL obsługuje natywnie przesyłanie strumieniowe.
Przesyłanie strumieniowe za pomocą interfejsu FastAPI i zdarzeń wysyłanych przez serwer
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. Najlepsze praktyki i antywzorce LangChain
Najlepsze praktyki
- Zawsze używaj LCEL zamiast starszych łańcuchów (LLMCain, RetrievalQA). LCEL jest bardziej wydajny, bezpieczny dla typów i obsługuje natywne przesyłanie strumieniowe.
- Włącz LangSmith w fazie rozwoju: Automatyczne śledzenie oszczędza godziny debugowania. Możesz wyłączyć tę opcję w środowisku produkcyjnym, aby zaoszczędzić koszty.
- MMR za różnorodność: Użyj maksymalnej trafności marginalnej (search_type="mmr") zamiast czystego podobieństwa, aby uniemożliwić aporterowi odzyskanie prawie identycznych fragmentów.
- async/czekaj na przepustowość: Użyj ainvoke i astream dla operacji I/O (LLM, wektorowy DB). Umożliwia obsługę współbieżnych żądań bez narzutu wątku.
- Oddziel logikę wyszukiwania od generowania: sprawia, że kod jest testowalny, pozwala na kpinę z retrievera w testach.
Anty-wzorce, których należy unikać
- Łańcuchy zagnieżdżone zbyt głęboko: LangChain pozwala na komponowanie bardzo złożonych łańcuchów. Powyżej 3-4 poziomów zagnieżdżenia debugowanie staje się trudne. Rozważ podzielenie łańcucha na funkcje.
- Ignoruj koszty tokenów: Każdy dokument w kontekście zwiększa koszt. Mierz i optymalizuj liczbę tokenów wysyłanych do LLM.
- Szablon podpowiedzi bez wersjonowania: Podpowiedzi są kodem. Wersjonuj je, testuj i śledź zmiany, tak jak każdy inny komponent.
- LLM wysokie temperatury dla RAG: dla RAG należy stosować temperatury 0,0-0,2. Wysokie temperatury zwiększają zmienność, a nie jakość i mają tendencję do zwiększania halucynacji.
Wnioski
LangChain przekształca złożoność systemu RAG w szereg elementów składowych modułowe. Budowaliśmy rurociągi od najprostszych (podstawowy RAG z LCEL) do bardziej zaawansowane (konwersacyjny RAG, multi-hop, self-query), dotykające każdego aspektu istotne dla produkcji: przesyłanie strumieniowe, śledzenie za pomocą LangSmith, wyszukiwanie hybrydowe i najlepsze praktyki w zakresie jakości.
Kluczowe punkty:
- LCEL to nowoczesny sposób tworzenia łańcuchów: czytelny, bezpieczny dla typów i natywny dla przesyłania strumieniowego
- Konwersacyjny RAG wymaga przeformułowania zapytania przed pobraniem
- Samodzielne pobieranie zapytań automatyzuje filtrowanie metadanych z naturalnego zapytania
- Pobieranie wieloprzeskokowe rozkłada złożone zapytania na sekwencyjne zapytania podrzędne
- EnsembleRetriever łączy BM25 + gęsty za pomocą jednego polecenia
- LangSmith jest niezbędny do debugowania i oceny w środowisku produkcyjnym
W następnym artykule omówimy Zarządzanie oknem kontekstowym: jak zarządzać budżetem tokena LLM i optymalizować go, gdy kontekst jest dostępny przekracza możliwości modelu.