RAG용 LangChain: 고급 프레임워크 및 패턴
랭체인 애플리케이션 구축을 위한 참조 프레임워크가 되었습니다. LLM을 기반으로 합니다. GitHub에는 80,000개가 넘는 스타가 있고 빠르게 성장하는 커뮤니티를 통해 문서 로더, 텍스트 등 RAG 시스템의 모든 구성 요소에 대한 강력한 추상화를 제공합니다. 스플리터, 임베딩 모델, 벡터 저장소, 검색기 및 체인. 하지만 그 진정한 힘은 이러한 빌딩 블록을 고급 패턴으로 결합하면 나타납니다.
이 기사에서는 LangChain을 사용하여 완전한 RAG 시스템을 구축할 것입니다. 기본 파이프라인부터 다음과 같은 고급 패턴까지 대화형 RAG (연속 질문 사이의 맥락 기억), 다중 홉 검색 (여러 추론 단계가 필요한 쿼리) 도구 호출 (컨설팅할 소스를 결정하는 에이전트) 및 자체 쿼리 검색 (메타데이터의 자동 의미 필터링). 모두 실행 가능한 코드 예제가 포함되어 있습니다.
무엇을 배울 것인가
- LangChain 아키텍처: 체인, 실행 가능 항목 및 LCEL(LangChain Expression Language)
- LangChain을 사용한 기본 RAG 파이프라인: 문서화부터 응답까지
- 대화형 RAG: 상황별 메모리 및 히스토리 관리
- 다단계 추론이 필요한 질문에 대한 다중 홉 검색
- 자체 쿼리 검색: 쿼리에서 메타데이터를 자동으로 필터링합니다.
- LangChain의 앙상블 검색기와 하이브리드 검색
- 프로덕션에서 더 나은 UX를 위한 스트리밍 응답
- LangSmith를 사용하여 LangChain 파이프라인 디버깅 및 테스트
1. 랭체인 표현 언어(LCEL)
LangChain은 버전 0.1.0부터 랭체인 표현
언어(LCEL): 파이프 패턴을 기반으로 한 선언적 구문(|)
읽기 쉽고 유형이 안전한 방식으로 체인을 구성합니다. LCEL은 스트리밍에 최적화되어 있으며,
병렬성과 추적 기능을 갖추고 있으며 LangChain 파이프라인을 구축하는 현대적인 방법입니다.
LCEL 기본 사항: 파이프 연산자 및 실행 가능 항목
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
LCEL의 잠재력 중 하나는 병렬 구성입니다. 즉, 복구가 가능합니다. 서로 다른 소스의 컨텍스트를 병렬로 결합하여 LLM에 전달합니다.
RunnableParallel: 여러 소스에서 검색
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. 기본 RAG 파이프라인 완성
고급 패턴을 다루기 전에 완전하고 강력한 RAG 파이프라인을 구축해 보겠습니다. LangChain과 함께: 문서 수집부터 검색, 응답 생성까지.
LangChain으로 완성된 RAG 파이프라인
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: 맥락 기억
기본 RAG의 문제점은 각 쿼리가 독립적으로 처리된다는 것입니다. 하나로 실제 대화에서 사용자는 시스템이 대화 내용을 기억할 것으로 기대합니다. 이전 질문. "그리고 두 번째 옵션은요?" 무슨 내용인지 모르면 의미가 없지 그는 말하고 있었습니다. 그만큼 대화형 RAG 이 문제를 해결합니다.
LangChain은 두 단계로 대화를 관리합니다.
- 쿼리 재구성: 채팅 기록을 바탕으로 현재 질문을 검색에 필요한 모든 컨텍스트를 포함하는 독립형 쿼리로 재구성합니다.
- 역사가 있는 RAG: 검색을 위해 재구성된 쿼리를 사용한 다음 검색된 컨텍스트와 채팅 기록을 모두 제공하는 응답을 생성합니다.
LangChain과의 대화형 RAG
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. 자체 쿼리 검색: 메타데이터 자동 필터링
Il 셀프 쿼리 검색 LangChain의 가장 강력한 패턴 중 하나입니다. LLM이 사용자의 자연스러운 쿼리를 해석하고 추출할 수 있도록 합니다. 의미 체계 쿼리와 메타데이터 필터가 모두 자동으로 생성됩니다. 사용자가 쓴다 "전문가가 작성한 2024개의 RAG 기사" 시스템이 자동으로 추출 연도 filter=2024 및 유형 filter="expert".
메타데이터를 사용한 자체 쿼리 검색
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. 복잡한 쿼리를 위한 다중 홉 검색
일부 질문에는 여러 추론 단계가 필요합니다. "모델을 개발한 사람은 누구입니까?" LangChain이 기본적으로 사용하며 언제 설립되었나요?" 찾기 전에 필요 LangChain이 기본적으로 OpenAI를 사용한다는 것을 알고 나면 OpenAI의 창립일을 찾아보세요. 이것은 호출됩니다 다중 홉 검색.
쿼리 분해를 사용한 다중 홉 RAG
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. 앙상블 리트리버와 하이브리드 검색
LangChain이 제공하는 앙상블리트리버 여러 리트리버를 결합한 것입니다. 구성 가능한 가중치를 사용하여 최종 순위에 상호 순위 융합을 적용합니다. LangChain에서 하이브리드 검색(BM25 + 벡터)을 구현하는 가장 쉬운 방법입니다.
앙상블 리트리버: BM25 + 덴스
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: 추적 및 디버깅
랭스미스 LangChain의 관찰 플랫폼입니다. 허용 체인의 각 단계, LLM으로 전송된 프롬프트, 검색된 문서, 지연 시간과 비용. 개발 디버깅 및 모니터링에 필수적입니다. 생산 중.
추적을 위한 LangSmith 설정
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. 더 나은 UX를 위한 스트리밍 응답
프로덕션 환경에서는 LLM 응답에 5~15초 정도 걸릴 수 있습니다. 단어를 보여주세요 생성(스트리밍)되면서 인식이 크게 향상됩니다. 사용자의 속도. LCEL은 기본적으로 스트리밍을 지원합니다.
FastAPI 및 서버 전송 이벤트를 사용한 스트리밍
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 모범 사례 및 안티 패턴
모범 사례
- 항상 LCEL을 사용하세요 레거시 체인(LLLMChain, RetrievalQA) 대신. LCEL은 성능이 더 뛰어나고 유형이 안전하며 기본 스트리밍을 지원합니다.
- 개발 중 LangSmith 활성화: 자동 추적으로 디버깅 시간이 절약됩니다. 비용을 절약하기 위해 프로덕션에서 비활성화할 수 있습니다.
- 다양성을 위한 MMR: 검색자가 거의 동일한 청크를 검색하지 못하도록 순수 유사성 대신 최대 주변 관련성(search_type="mmr")을 사용합니다.
- 비동기/처리량 대기: I/O 작업(LLM, 벡터 DB)에는 ainvoke 및 astream을 사용합니다. 스레드 오버헤드 없이 동시 요청을 처리할 수 있습니다.
- 생성과 분리된 검색 논리: 코드를 테스트 가능하게 만들고 테스트에서 검색기를 모의할 수 있습니다.
피해야 할 안티패턴
- 체인이 너무 깊게 중첩됨: LangChain을 사용하면 매우 복잡한 체인을 구성할 수 있습니다. 3~4개 수준의 중첩을 넘어서면 디버깅이 어려워집니다. 체인을 기능으로 나누는 것을 고려해보세요.
- 토큰 비용 무시: 상황에 맞는 각 문서는 비용을 증가시킵니다. LLM으로 전송된 토큰 수를 측정하고 최적화합니다.
- 버전 관리가 없는 프롬프트 템플릿: 프롬프트는 코드입니다. 다른 구성 요소와 마찬가지로 버전을 지정하고, 테스트하고, 변경 사항을 추적하세요.
- RAG의 LLM 고온: RAG 사용 온도는 0.0-0.2입니다. 높은 온도는 품질이 아닌 가변성을 증가시키며 환각을 증가시키는 경향이 있습니다.
결론
LangChain은 RAG 시스템의 복잡성을 일련의 빌딩 블록으로 변환합니다. 모듈식. 우리는 가장 간단한(LCEL이 포함된 기본 RAG)부터 모든 측면을 다루는 고급(대화형 RAG, 멀티 홉, 자체 쿼리) 프로덕션 관련: 스트리밍, LangSmith를 사용한 추적, 하이브리드 검색 및 품질을 위한 모범 사례.
핵심 포인트:
- LCEL은 읽기 가능하고 유형이 안전하며 스트리밍 네이티브인 체인을 구성하는 현대적인 방법입니다.
- 대화형 RAG는 검색 전에 쿼리 재구성이 필요합니다.
- 자체 쿼리 검색은 자연 쿼리에서 메타데이터 필터링을 자동화합니다.
- 다중 홉 검색은 복잡한 쿼리를 순차적 하위 쿼리로 분해합니다.
- EnsembleRetriever는 BM25 + Dense를 단일 명령으로 결합합니다.
- LangSmith는 프로덕션 환경에서 디버깅 및 평가에 필수적입니다.
다음 기사에서는 컨텍스트 창 관리: 컨텍스트가 가능할 때 LLM 토큰 예산을 관리하고 최적화하는 방법 모델의 능력을 초과합니다.
시리즈는 계속됩니다
- 기사 1: RAG 설명
- 기사 2: 임베딩 및 의미 검색
- 기사 3: 벡터 데이터베이스
- 기사 4: 하이브리드 검색
- 기사 5: 생산 중인 RAG
- 조항 6: RAG용 LangChain(현재)
- 기사 7: 컨텍스트 창 관리