Что такое RAG (Retrieval-Augmented Generation) и как реализовать его с помощью LangChain?

Фаза 1: Indexing

from langchain_community.document_loaders import PyPDFLoader, DirectoryLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 1. Загрузка документов
loader = DirectoryLoader("./docs", glob="**/*.pdf", loader_cls=PyPDFLoader)
documents = loader.load()

# 2. Разбивка на чанки
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""],
)
chunks = splitter.split_documents(documents)
print(f"Created {len(chunks)} chunks from {len(documents)} documents")

# 3. Создание embeddings и векторного хранилища
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",
)
vectorstore.persist()

RAG (Retrieval-Augmented Generation) в LangChain

RAG решает ключевое ограничение LLM — незнание актуальных или приватных данных. Перед генерацией система извлекает релевантные документы и добавляет их в промпт как контекст. Это устраняет галлюцинации на фактических вопросах и убирает необходимость дообучения модели.

Архитектура RAG-пайплайна

Пайплайн состоит из двух фаз: Indexing (однократно при загрузке документов) и Retrieval + Generation (при каждом запросе).

Фаза 2: Retrieval + Generation

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# Загрузка существующего хранилища
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings(model="text-embedding-3-small"),
)

# Retriever — возвращает top-4 релевантных чанка
retriever = vectorstore.as_retriever(
    search_type="mmr",  # Maximum Marginal Relevance — разнообразие + релевантность
    search_kwargs={"k": 4, "fetch_k": 20},
)

llm = ChatOpenAI(model="gpt-4o", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", """
You are a helpful assistant. Answer the question using ONLY the provided context.
If the answer is not in the context, say "I don't have information about this."

Context:
{context}
"""),
    ("human", "{question}"),
])

def format_docs(docs):
    return "\n\n---\n\n".join(
        f"Source: {d.metadata.get('source', 'unknown')}\n{d.page_content}"
        for d in docs
    )

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# Запрос
answer = rag_chain.invoke("What are the main features of the product?")
print(answer)

RAG с историей диалога

from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.prompts import MessagesPlaceholder

condense_prompt = ChatPromptTemplate.from_messages([
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{question}"),
    ("human", "Given the conversation above, rephrase the follow-up question into a standalone question."),
])

condense_chain = condense_prompt | llm | StrOutputParser()

def condense_question(inputs):
    if inputs.get("chat_history"):
        return condense_chain.invoke(inputs)
    return inputs["question"]

conversational_rag = (
    RunnablePassthrough.assign(standalone_question=condense_question)
    | {"context": lambda x: retriever | format_docs)(x["standalone_question"]), "question": lambda x: x["standalone_question"]}
    | prompt
    | llm
    | StrOutputParser()
)

Streaming ответов

for chunk in rag_chain.stream("Explain the pricing model"):
    print(chunk, end="", flush=True)

Подводные камни

• chunk_overlap критичен — без перекрытия предложения на границах чанков теряют контекст; рекомендуется 10–20% от chunk_size.
• MMR медленнее cosine similarity — fetch_k документов сначала извлекается по similarity, потом применяется MMR; при большом fetch_k это дорого.
• Embedding модель должна быть одна — если индексировали text-embedding-3-small, а запросы делают через text-embedding-3-large, качество поиска резко падает.
• Chroma не масштабируется горизонтально — для production используйте Pinecone, Qdrant или pgvector с connection pooling.
• Большой контекст не всегда лучше — при k=10 модель получает много шума; оптимальное k обычно 3–5, определяется экспериментально.
• Метаданные теряются при split — TextSplitter копирует metadata родительского документа в каждый чанк, но custom поля нужно добавлять вручную через doc.metadata["chapter"] = ....
• Hypothetical Document Embeddings (HyDE) — для сложных вопросов стандартный retrieval плохо работает; рассмотрите генерацию гипотетического ответа перед поиском.

• Объяснять rag implementation только синтаксисом без shape, dtype, состояния или режима выполнения.
• Игнорировать leakage, воспроизводимость, пустые входы и скрытые копии данных.
• Не проверять production-симптомы: latency, память, ретраи, дрейф качества и несовпадение версий.

• Может ли связать rag implementation с реальным контрактом входов и выходов.
• Упоминает ли тесты, метрики, reproducibility и диагностику ошибок.
• Видит ли различие между demo-кодом в ноутбуке и production-пайплайном.
• Предлагает ли observability, rollback, ограничения стоимости и стратегию incident replay.