RAG 用 LangChain: 高度なフレームワークとパターン
ラングチェーン アプリケーションを構築するための参照フレームワークとなっています。 LLMに基づいています。 GitHub には 80,000 を超えるスターがあり、急速に成長しているコミュニティにより、 RAG システムのあらゆるコンポーネント (ドキュメント ローダー、テキスト) に強力な抽象化を提供します。 スプリッター、埋め込みモデル、ベクター ストア、レトリーバー、チェーン。しかし、その真の力は これらの構成要素を高度なパターンに組み合わせたときに現れます。
この記事では、LangChain を使用して完全な RAG システムを構築します。 基本的なパイプラインから高度なパターンまで 会話型 RAG (連続した質問の間の文脈記憶)、 マルチホップ検索 (複数の推論ステップを必要とするクエリ)、 ツール呼び出し (どの情報源に相談するかを決定するエージェント) 自己クエリの取得 (メタデータの自動セマンティック フィルタリング)。すべてに実行可能なコード例が含まれています。
何を学ぶか
- LangChain アーキテクチャ: チェーン、ランナブル、および LCEL (LangChain Expression Language)
- LangChain を使用した基本的な RAG パイプライン: ドキュメントから応答まで
- 会話型 RAG: コンテキスト記憶と履歴管理
- 複数ステップの推論を必要とする質問のマルチホップ検索
- セルフクエリ取得: クエリからのメタデータの自動フィルタリング
- LangChain のアンサンブル レトリーバーとハイブリッド検索
- 実稼働環境での UX を向上させるためのストリーミング応答
- LangSmith を使用した LangChain パイプラインのデバッグとテスト
1. ラングチェーン式言語 (LCEL)
バージョン 0.1.0 以降、LangChain は ラングチェーン式
言語 (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 の可能性の 1 つは並列合成です。それらは復元可能です。 異なるソースからのコンテキストを並行して取得し、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 は会話を 2 つのステップで管理します。
- クエリの再定式化: チャットの履歴を考慮して、現在の質問を、取得に必要なすべてのコンテキストを含むスタンドアロン クエリに再定式化します。
- 歴史のある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 で最も強力なパターンの 1 つです。 LLM がユーザーの自然なクエリを解釈して抽出できるようにします。 セマンティック クエリとメタデータ フィルターの両方が自動的に実行されます。ユーザーが書きます 「専門家によって書かれた RAG に関する 2024 年の記事」とシステムが自動的に抽出します 年は 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 の可観測性プラットフォームです。許可します to view each step of the chain, the prompts sent to the LLM, the documents retrieved, レイテンシとコスト。 It is essential for development debugging and monitoring 生産中です。
トレース用の 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 を使用する 従来のチェーン (LLMChain、RetrievalQA) の代わりに。 LCEL はよりパフォーマンスが高く、タイプセーフであり、ネイティブ ストリーミングをサポートしています。
- 開発時に LangSmith を有効にする: 自動トレースにより、何時間ものデバッグ時間が節約されます。実稼働環境ではコストを節約するために無効にすることができます。
- 多様性のためのMMR: 純粋な類似性の代わりに最大周辺関連性 (search_type="mmr") を使用して、取得者がほぼ同一のチャンクを取得しないようにします。
- 非同期/スループットを待つ: I/O 操作には ainvoke と astream を使用します (LLM、ベクトル DB)。これにより、スレッドのオーバーヘッドなしで同時リクエストを処理できます。
- 取得ロジックを生成から分離する: コードをテスト可能にし、テストでレトリーバーをモックできるようにします。
避けるべきアンチパターン
- チェーンのネストが深すぎます: LangChain を使用すると、非常に複雑なチェーンを構成できます。ネストが 3 ～ 4 レベルを超えると、デバッグが困難になります。チェーンを関数に分割することを検討してください。
- トークンコストを無視する: コンテキスト内の各ドキュメントによりコストが増加します。 LLM に送信されるトークンの数を測定して最適化します。
- バージョン管理を行わないプロンプトテンプレート: プロンプトはコードです。他のコンポーネントと同様に、バージョンを付け、テストし、変更を追跡します。
- RAG の LLM 高温: RAG の場合、温度 0.0 ～ 0.2 を使用します。高温では品質ではなく変動が増加し、幻覚が増加する傾向があります。
結論
LangChain は、RAG システムの複雑さを一連の構成要素に変換します。 モジュール式。私たちは、最も単純なもの (LCEL を使用した基本的な RAG) から、 より高度な (会話型 RAG、マルチホップ、自己クエリ)、あらゆる側面に対応 制作に関連するもの: ストリーミング、LangSmith によるトレース、ハイブリッド検索、 品質のためのベストプラクティス。
重要なポイント:
- LCEL はチェーンを構成する最新の方法です: 読み取り可能、タイプセーフ、ストリーミングネイティブ
- 会話型 RAG では、取得前にクエリを再構築する必要があります
- セルフクエリ取得により、自然なクエリからのメタデータのフィルタリングが自動化されます。
- マルチホップ取得により、複雑なクエリが連続したサブクエリに分解されます。
- EnsembleRetriever は BM25 + デンスを 1 つのコマンドで組み合わせます
- LangSmith は実稼働環境でのデバッグと評価に不可欠です
次の記事では、 コンテキストウィンドウの管理: コンテキストが利用可能な場合に LLM トークンの予算を管理および最適化する方法 モデルの能力を超えています。
シリーズは続く
- 記事 1: RAG の説明
- 第 2 条: 埋め込みとセマンティック検索
- 第3条：ベクターデータベース
- 第 4 条: ハイブリッド検索
- 第 5 条: 本番環境における RAG
- 第6条：RAG用LangChain（現行）
- 第 7 条: コンテキスト ウィンドウの管理