RAG Deep Dive: от простого к production-ready
opensourceaillmragit
Введение: что такое RAG?
RAG (Retrieval-Augmented Generation) — подход, при котором LLM дополняется внешним источником знаний:
Без RAG:
User: "Какая зарплата в Google?"
LLM: "Я не знаю, моя база знаний до 2024 года..."
С RAG:
User: "Какая зарплата в Google?"
System: ищет в базе знаний → находит данные
LLM: "По данным на 2026 год, средняя зарплата..."
RAG решает три проблемы:
- Устаревшие знания — модель всегда может получить актуальные данные
- Приватность — данные не попадают в веса модели
- Верифицируемость — можно показать источник ответа
Базовый RAG pipeline
4 шага простого RAG
1. Indexing (индексация):
Документы → чанки → эмбеддинги → векторная база
2. Retrieval (поиск):
Вопрос → эмбеддинг → поиск похожих чанков
3. Augmentation (обогащение):
Вопрос + найденные чанки → prompt
4. Generation (генерация):
LLM отвечает на основе вопроса и контекста
Код простого RAG
from langchain.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.llms import OpenAI
from langchain.chains import RetrievalQA
# 1. Загрузка документов
loader = DirectoryLoader('docs/')
documents = loader.load()
# 2. Разбиение на чанки
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50
)
chunks = splitter.split_documents(documents)
# 3. Эмбеддинги и векторная база
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)
# 4. Поиск и генерация
qa_chain = RetrievalQA.from_chain_type(
llm=OpenAI(),
chain_type="stuff",
retriever=vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 4}
)
)
response = qa_chain.run("Как настроить Docker?")
Chunking: как разбивать документы
Стратегии chunking
1. Fixed-size splitting:
Разбивка по количеству символов/слов
+ Просто
- Может разрезать предложения
2. Recursive character splitting (LangChain):
Сначала по двойным переносам строк
Потом по переносам строк
Потом по предложениям
+ Сохраняет структуру
- Стандартный подход
3. Semantic chunking:
Разбивка по смысловым границам
+ Сохраняет смысл
- Медленнее, сложнее
4. Document-aware splitting:
Учёт структуры (заголовки, абзацы, списки)
+ Лучший контекст
- Требует анализа структуры
Оптимальные размеры чанков
Эмбеддинг модель → оптимальный размер:
text-embedding-3-small → 256-512 tokens
text-embedding-3-large → 512-1024 tokens
bge-large → 512-1024 tokens
nomic-embed-text → 256-512 tokens
Overlap: 10-20% от chunk_size
Без overlap: теряются границы контекста
С большим overlap: больше шума
Пример chunking с metadata
from langchain.text_splitter import MarkdownHeaderTextSplitter
headers_to_split_on = [
("#", "Header1"),
("##", "Header2"),
("###", "Header3"),
]
splitter = MarkdownHeaderTextSplitter(
headers_to_split_on=headers_to_split_on
)
chunks = splitter.split_text(document)
# Каждый чанк содержит:
# - page_content: текст чанка
# - metadata: {"Header1": "...", "Header2": "..."}
# → можно фильтровать по заголовкам при поиске
Embedding модели
Сравнение embedding моделей
2024-2025 benchmarks (MTEB):
Модель | Размер | MTEB Score
text-embedding-3-small | 137M | 63.1%
text-embedding-3-large | 300M | 66.7%
bge-large-en-v1.5 | 335M | 62.8%
nomic-embed-text | 137M | 61.5%
jina-embeddings-v2-base | 137M | 61.2%
voyage-multilingual-2 | 137M | 64.3%
Для русского:
sdvity/muse-v2-ru-large | 550M | 72.1% (RuBenchmark)
intfloat/multilingual-e5-large | 674M | 65.0%
Использование embedding
# OpenAI embeddings
from openai import OpenAI
client = OpenAI()
response = client.embeddings.create(
model="text-embedding-3-small",
input=["Как настроить Docker?"]
)
embedding = response.data[0].embedding
# Длина: 1536 для small, 3072 для large
# Локальные embeddings (Ollama)
import requests
response = requests.post(
"http://localhost:11434/api/embeddings",
json={
"model": "nomic-embed-text",
"prompt": "Как настроить Docker?"
}
)
embedding = response.json()["embedding"]
# Длина: 768
Векторные базы данных
Сравнение векторных БД
Chroma:
+ Простая установка (pip install)
+ Встроенная в Python
- Не для production
- Нет replication
FAISS (Facebook):
+ Очень быстрый
+ Легкий
- Нет метаданных (нужен wrapper)
- Только in-memory / local
Pinecone:
+ Managed service
+ Очень быстрый
- Дорогой
- Только cloud
Qdrant:
+ Open source + cloud
+ Filter query
+ Rust (быстрый)
- Больше ресурсов
Weaviate:
+ Hybrid search (vector + BM25)
+ GraphQL
+ Open source
- Сложнее в настройке
Milvus:
+ Distributed
+ Enterprise features
- Тяжёлый
pgvector (PostgreSQL):
+ Простая интеграция с PostgreSQL
+ SQL queries
+ Filter
- Медленнее при больших данных
Пример с Qdrant
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, Point
client = QdrantClient(url="http://localhost:6333")
# Создание коллекции
client.create_collection(
collection_name="docs",
vectors_config=VectorParams(
size=1536,
distance=Distance.COSINE
)
)
# Добавление документов
points = [
Point(
id=i,
vector=embedding,
payload={
"text": chunk.content,
"source": chunk.source,
"page": chunk.page
}
)
for i, embedding in enumerate(embeddings)
]
client.upsert(collection_name="docs", points=points)
# Поиск
results = client.search(
collection_name="docs",
query_vector=question_embedding,
limit=5,
query_filter={
"must": [
{
"key": "source",
"match": {"value": "docker.md"}
}
]
}
)
Hybrid search (BM25 + Vector)
Vector search:
- Семантический поиск
- Хорош для смысла
- Плох для точных названий/кодов
BM25 (keyword search):
- Точное совпадение слов
- Хорош для названий, кодов
- Плох для синонимов
Hybrid = 0.3 × BM25 + 0.7 × Vector
Лучший из обоих миров!
Weaviate и Qdrant поддерживают hybrid out-of-the-box
Продвинные техники RAG
Re-ranking
Проблема: первые k результатов не всегда лучшие
Решение: re-ranking
1. Быстрый поиск: k=50 (vector search)
2. Точный rerank: выбрать top 5
Cross-encoder reranker:
BGE-reranker-large: 330M параметров
Jina-reranker-v2: 137M параметров
Качество: +15-20% к recall
Скорость: -200ms на запрос
from FlagEmbedding import FlagReranker
reranker = FlagReranker(
'BAAI/bge-reranker-large',
use_gpu=True
)
# 50 кандидатов → 5 лучших
pairs = [[question, doc] for doc in candidates]
scores = reranker.compute_score(pairs)
# Сортируем и берём top 5
top_docs = sorted(zip(candidates, scores),
key=lambda x: x[1], reverse=True)[:5]
Multi-query retrieval
Проблема: один вопрос → один embedding
Вопрос может иметь разные смыслы
Решение: multi-query
LLM генерирует 3-5 вариантов вопроса
Ищем по каждому
Объединяем результаты
Пример:
Вопрос: "Как настроить Docker?"
Варианты:
1. "Docker setup guide"
2. "How to install and configure Docker container"
3. "Docker installation steps"
# Генерация вариантов вопроса
query_gen_prompt = """
Generate 3 different versions of this question:
Original: {question}
Return only the questions, one per line.
"""
variations = llm.generate(query_gen_prompt.format(question=q))
# variations = ["var1", "var2", "var3"]
# Поиск по каждому варианту
all_results = []
for variant in variations:
results = vectorstore.similarity_search(variant, k=10)
all_results.extend(results)
# Убираем дубликаты
unique_results = list({doc.page_content: doc for doc in all_results}.values())
Query transformation
1. Hypothetical Document Embeddings (HyDE):
LLM генерирует "гипотетический ответ"
Эмбеддинг ответа ищем вместо эмбеддинга вопроса
Работает когда:
- Ответ содержит конкретные факты
- Вопрос требует точного совпадения
2. Step-back prompting:
Сначала генерируем общий вопрос
Ищем по общему → получаем контекст
Затем ищем по конкретному → получаем детали
3. Decomposition:
Разбиваем сложный вопрос на под-вопросы
Ищем по каждому
Объединяем ответы
Context window management
Проблема: контекст ограничен
Context window:
Llama 3: 8K / 128K
GPT-4: 8K / 128K
Claude: 100K
Но: чем больше контекст, тем больше "шум"
Стратегии управления контекстом
1. Stuffing (простой RAG):
Все чанки в один prompt
+ Просто
- Не влезает в контекст
2. Map-Reduce:
Map: для каждого чанка → ответ
Reduce: объединить ответы
+ Работает с большим количеством
- Теряется связь между чанками
3. Refine:
Последовательная обработка
Ответ на предыдущем → контекст для следующего
+ Лучшее качество
- Медленнее
4. RAG-Fusion:
Переформулируем query
Ищем по всем вариантам
Rerank результаты
Генерируем ответ
Пример Map-Reduce
def map_reduce_rag(question, chunks):
# Map: каждый чанк → частичный ответ
partial_answers = []
for chunk in chunks:
prompt = f"""
Context: {chunk}
Question: {question}
Answer based on context only.
"""
answer = llm.generate(prompt)
partial_answers.append(answer)
# Reduce: объединить ответы
combined = "\n\n".join(partial_answers)
final_prompt = f"""
Partial answers:
{combined}
Synthesize into one final answer.
"""
return llm.generate(final_prompt)
Evaluation RAG
Метрики оценки
1. Retrieval metrics:
- Recall@k: сколько релевантных документов в top-k
- MRR (Mean Reciprocal Rank): средний ранг первого релевантного
- NDCG (Normalized Discounted Cum Gain): взвешенный рейтинг
2. Generation metrics:
- Faithfulness: ответ основан на контексте?
- Answer Relevance: ответ отвечает на вопрос?
- Context Precision: релевантен ли контекст?
3. End-to-end:
- Exact Match: точное совпадение с эталоном
- BLEU / ROUGE: n-gram overlap
- LLM-as-a-judge: LLM оценивает ответ
LLM-as-a-judge
evaluation_prompt = """
Evaluate the following answer:
Question: {question}
Context: {context}
Answer: {answer}
Rate on scale 1-5:
1. Faithfulness (based on context?)
2. Relevance (answers question?)
3. Completeness (covers all aspects?)
Return JSON: {"faithfulness": 4, "relevance": 5, "completeness": 3}
"""
def evaluate_rag(question, context, answer):
result = llm.generate(evaluation_prompt.format(
question=question,
context=context,
answer=answer
))
return json.loads(result)
LangSmith и другие инструменты
LangSmith:
- Tracing (отслеживание каждого шага)
- Dataset evaluation
- Performance monitoring
- Cost tracking
Ragas:
- Open source evaluation framework
- Metrics: faithfulness, answer relevancy, context precision
pip install ragas
from ragas import evaluate
result = evaluate(
dataset,
metrics=[faithfulness, answer_relevancy]
)
Production RAG
Архитектура production RAG
Production RAG System:
API Layer (FastAPI/Node):
/query → обработка запроса
/index → индексация документов
/health → мониторинг
Queue (Redis/Celery):
Асинхронная индексация
Batch processing
Vector DB (Qdrant/Weaviate):
Replication
Backup
Monitoring
Cache (Redis):
Кэширование частых запросов
TTL: 1-24 часа
Monitoring:
LangSmith / Phoenix
Latency, error rate, cost
Feedback loop
Кэширование
import hashlib
import redis
r = redis.Redis(host='localhost', port=6379)
def cached_rag_query(question):
# Хеш вопроса
q_hash = hashlib.md5(question.encode()).hexdigest()
# Проверяем кэш
cached = r.get(f"rag:{q_hash}")
if cached:
return json.loads(cached)
# Выполняем RAG
result = rag_pipeline.run(question)
# Кэшируем на 1 час
r.setex(f"rag:{q_hash}", 3600, json.dumps(result))
return result
Обновление индекса
Стратегии обновления:
1. Full reindex:
Полная перестройка индекса
+ Точно
- Дowntime, дорого
2. Incremental update:
Добавляем только новые документы
+ Быстро
- Нужно отслеживать изменения
3. Delta indexing:
Отслеживаем изменения по файлам
Пересчитываем только изменённые чанки
+ Оптимально
- Сложнее в реализации
Итоги
RAG — мощный инструмент для:
- Базы знаний компании
- Поддержки клиентов
- Поиска по документам
- Интеграции с LLM
Рекомендации:
1. Начните с простого RAG
2. Добавьте re-ranking для качества
3. Оценивайте с помощью Ragas/LangSmith
4. Кэшируйте частые запросы
5. Используйте hybrid search