Advanced RAG: техники улучшения retrieval
opensourceaillmragretrievalit
Введение: простой RAG не всегда работает
Проблема Naive RAG
Naive RAG:
Query → Embed → Retrieve → Concat → Generate
✗ Chunking loses context
✗ Single embedding misses semantics
✗ No reranking
✗ Stale knowledge
Advanced RAG:
Query → Transform → Multi-retrieve → Rerank → Augment → Generate
✓ Better retrieval
✓ Better context
✓ Better answers
Факт: Advanced RAG может улучшить качество ответов на 30-50% по сравнению с naive RAG.
Что такое Advanced RAG?
RAG Spectrum
Pre-retrieval (до поиска):
- Query transformation
- Query rewriting
- Query expansion
Retrieval (во время поиска):
- Multi-vector search
- Hybrid search
- Multi-modal retrieval
Post-retrieval (после поиска):
- Reranking
- Context compression
- Selective context
When to Use Advanced RAG
Use naive RAG when:
✓ Small corpus (< 10K docs)
✓ Simple questions
✓ Fast prototyping
Use advanced RAG when:
✗ Large corpus (> 1M docs)
✗ Complex questions
✗ High accuracy required
✗ Multi-document reasoning
Уровень 1: Query Transformation
Query Rewriting
Original query:
"Как настроить ollama?"
Rewritten:
"Ollama setup configuration guide"
"How to install and configure Ollama local LLM"
"Ollama installation steps Linux Windows Mac"
Почему это работает:
- Original может быть слишком кратким
- Rewritten покрывает больше синонимов
- Multiple queries = better recall
Query Expansion
class QueryExpander:
def __init__(self, llm):
self.llm = llm
def expand(self, query: str, n: int = 3) -> list[str]:
"""Expand query using LLM"""
prompt = f"""
Given the following query, generate {n} related queries:
Query: {query}
Generate variations that might retrieve different relevant docs.
"""
response = self.llm.generate(prompt)
return self._parse_queries(response)
def search(self, original_query: str, top_k: int = 10):
expanded = self.expand(original_query, n=3)
all_results = []
for q in [original_query] + expanded:
results = self.vector_db.search(q, top_k=20)
all_results.extend(results)
# Deduplicate and return top_k
return self._deduplicate(all_results)[:top_k]
Hypothetical Document Embeddings (HyDE)
HyDE: гипотетический ответ для лучшего поиска
Original query:
"Что такое transformer architecture?"
HyDE approach:
1. Generate hypothetical answer:
"Transformer architecture is a neural network
design that uses self-attention mechanisms.
It consists of an encoder-decoder structure
with multi-head attention and positional encoding."
2. Embed the hypothetical answer
3. Search using this embedding
Why it works:
- Answer embedding closer to relevant docs
- Bridges query-document gap
HyDE Example
class HyDERetriever:
def __init__(self, llm, vector_db):
self.llm = llm
self.db = vector_db
def search(self, query: str, top_k: int = 5):
# Generate hypothetical answer
hypo_prompt = f"""
Write a short paragraph answering this question.
Do NOT say you're hypothetical. Write it as fact.
Question: {query}
"""
hypothetical = self.llm.generate(hypo_prompt)
# Embed and search
results = self.db.search(
query_text=hypothetical,
top_k=top_k
)
return results
Уровень 2: Multi-Vector Retrieval
Multiple Embeddings per Document
Single embedding:
Document: "LLM training requires GPU clusters"
Embedding: [0.23, -0.45, ...]
Query "GPU" → partial match
Multi-embedding:
Document: "LLM training requires GPU clusters"
Chunk 1: "LLM training" → [0.23, -0.45, ...]
Chunk 2: "GPU clusters" → [0.67, 0.12, ...]
Chunk 3: "training infrastructure" → [-0.11, 0.89, ...]
Query "GPU" → Chunk 2 match ✓
Query "training" → Chunk 1 match ✓
Section-Level Embeddings
class MultiVectorRetriever:
def __init__(self, embedding_model):
self.model = embedding_model
self.sections = {} # doc_id → [section_embeddings]
def ingest(self, document: dict):
"""Split doc into sections, embed each"""
sections = self._split_into_sections(
document['content']
)
embeddings = self.model.encode(sections)
self.sections[document['id']] = [
{"text": s, "embedding": e}
for s, e in zip(sections, embeddings)
]
def search(self, query: str, top_k: int = 5):
query_emb = self.model.encode(query)
all_scores = []
for doc_id, sec_list in self.sections.items():
for i, section in enumerate(sec_list):
score = cosine(query_emb, section['embedding'])
all_scores.append({
"doc_id": doc_id,
"section_idx": i,
"text": section['text'],
"score": score
})
return sorted(all_scores, key=lambda x: -x['score'])[:top_k]
Parent-Child Retrieval
Parent-Child strategy:
Child chunks (small, precise embedding):
"Self-attention computes scores for all positions"
"Key, Query, Value matrices project inputs"
"Multi-head attention uses parallel heads"
Parent docs (large, full context):
"Attention Mechanism: Self-attention computes
scores for all positions in the sequence.
Key, Query, Value matrices project inputs.
Multi-head attention uses parallel heads..."
Search: find child chunks
Return: parent documents
Уровень 3: Reranking
Why Reranking?
Vector search (recall phase):
Query → Top 100 from 1M docs
Fast but approximate
Reranking (precision phase):
Top 100 → Top 10
More accurate, slower
Reranker model:
Input: (query, document) pair
Output: relevance score [0, 1]
Cross-encoder:
[CLS] query [SEP] document [SEP]
→ BERT → sigmoid → 0.87
More accurate than cosine similarity!
Cross-Encoder Reranker
from sentence_transformers import CrossEncoder
# Cross-encoder reranker
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
# Query + candidate documents
query = "Что такое LLM?"
documents = [
"LLM — large language model",
"Machine learning для текста",
"Трансформер для NLP"
]
# Score all at once
scores = reranker.predict([(query, doc) for doc in documents])
# [0.87, 0.23, 0.45]
# Sort by score
results = sorted(zip(documents, scores), key=lambda x: -x[1])
Two-Stage Retrieval
class TwoStageRetriever:
def __init__(self, vector_db, reranker,
recall_k: int = 100, rank_k: int = 10):
self.db = vector_db
self.reranker = reranker
self.recall_k = recall_k
self.rank_k = rank_k
def search(self, query: str):
# Stage 1: Fast recall
candidates = self.db.search(
query, top_k=self.recall_k
)
# Stage 2: Accurate reranking
scores = self.reranker.predict(
[(query, doc['text']) for doc in candidates]
)
# Combine and sort
results = [
{**doc, "rerank_score": score}
for doc, score in zip(candidates, scores)
]
return sorted(results, key=lambda x: -x['rerank_score'])[:self.rank_k]
Popular Rerankers
Cross-encoder models:
- cross-encoder/ms-marco-MiniLM-L-6-v2 (33M params)
- BAAI/bge-reranker-base (137M params)
- BAAI/bge-reranker-large (335M params)
Performance:
MiniLM: 1ms per pair (CPU)
Base: 3ms per pair (CPU)
Large: 8ms per pair (CPU)
Accuracy:
MiniLM: 85% of large model accuracy
Base: 93% of large model accuracy
Large: 100% (baseline)
Уровень 4: Context Compression
Problem: Too Much Context
Retrieved 10 documents, 50K tokens:
✗ Exceeds context window
✗ Expensive (tokens = money)
✗ Noisy (irrelevant content)
Context compression:
50K tokens → 5K tokens
✓ Keeps relevant info
✓ Saves cost
✓ Improves signal-to-noise
LLM-based Compression
class ContextCompressor:
def __init__(self, llm):
self.llm = llm
def compress(self, query: str, documents: list[dict],
max_tokens: int = 4000) -> str:
"""Keep only relevant parts"""
prompt = f"""
Given the following documents and a query,
extract ONLY the information relevant to the query.
Query: {query}
Documents:
{''.join([doc['text'] for doc in documents])}
Return only relevant excerpts. No commentary.
"""
return self.llm.generate(prompt)
Embedding-based Compression
class SemanticCompression:
def compress(self, query_embedding, documents):
"""Keep documents with highest similarity"""
scores = []
for doc in documents:
score = cosine(query_embedding, doc['embedding'])
scores.append((doc, score))
# Sort and truncate
scores.sort(key=lambda x: -x[1])
total_tokens = 0
selected = []
for doc, score in scores:
if total_tokens + len(doc['tokens']) > max_tokens:
break
selected.append(doc)
total_tokens += len(doc['tokens'])
return selected
Уровень 5: Adaptive RAG
Self-Reflective RAG
Adaptive RAG: choose strategy based on query
Flow:
1. Analyze query type
2. Choose retrieval strategy
3. Retrieve
4. Check if answer is confident
5. If not, try alternative strategy
Query types:
- Factual: direct vector search
- Comparative: multi-query search
- Temporal: filter by date
- Complex: chain-of-verification
Self-RAG
class SelfRAG:
def __init__(self, llm, retriever):
self.llm = llm
self.retriever = retriever
def answer(self, query: str):
# Step 1: Should we retrieve?
should_retrieve = self._check_need_retrieval(query)
if should_retrieve:
# Step 2: Retrieve
docs = self.retriever.search(query)
# Step 3: Generate with context
answer = self._generate_with_context(query, docs)
# Step 4: Evaluate answer
support_score = self._evaluate_support(
query, answer, docs
)
if support_score < 0.5:
# Try with different strategy
docs = self.retriever.search_expanded(query)
answer = self._generate_with_context(
query, docs
)
return answer
Уровень 6: Graph RAG
Knowledge Graph + RAG
Knowledge Graph:
(LLM) --[requires]--> (GPU)
(GPU) --[used_for]--> (Training)
(Training) --[needs]--> (Data)
Graph RAG:
Query: "What does LLM need?"
1. Graph traversal:
LLM → requires → GPU → used_for → Training
2. Text retrieval:
Search for "GPU training requirements"
3. Combine:
Graph context + text context → LLM
GraphRAG Implementation
class GraphRAG:
def __init__(self, graph_db, vector_db):
self.graph = graph_db
self.vector_db = vector_db
def search(self, query: str):
# 1. Extract entities from query
entities = self._extract_entities(query)
# 2. Traverse graph
graph_context = self._traverse_graph(
entities, max_hops=2
)
# 3. Text search
text_context = self.vector_db.search(query)
# 4. Combine
combined = {
"graph": graph_context,
"text": text_context,
"query": query
}
return self.llm.generate(
self._format_combined(combined)
)
Уровень 7: Multi-Modal RAG
Beyond Text
Multi-modal RAG:
Query: "What does the chart show?"
Sources:
- Text documents
- Images (charts, diagrams)
- Tables
- Code snippets
Embeddings:
- Text: BERT → 768-dim
- Image: CLIP → 512-dim
- Table: structure + text → 768-dim
CLIP for Images
import clip
from PIL import Image
# CLIP model
model, preprocess = clip.load("ViT-B/32")
# Image embeddings
image = preprocess(Image.open("chart.png")).unsqueeze(0)
image_embedding = model.encode_image(image)
# shape: [1, 512]
# Text embeddings
text = clip.tokenize(["chart data", "graph trends"]).to(model.device)
text_embedding = model.encode_text(text)
# shape: [2, 512]
# Similarity
similarity = clip.similarity(text_embedding, image_embedding)
# [[0.85, 0.62]]
Уровень 8: Agentic RAG
Tool-Using RAG
Agentic RAG:
Agent decides HOW to retrieve
1. User asks question
2. Agent plans retrieval strategy
3. Agent executes tools:
- Vector search
- Web search
- Database query
- API call
4. Agent evaluates results
5. Agent generates answer
ReAct Pattern
class AgenticRAG:
def __init__(self, tools, llm):
self.tools = tools # search, calculate, etc.
self.llm = llm
def reason(self, query: str):
"""ReAct: Reason + Act"""
thoughts = []
observations = []
for step in range(5): # max 5 steps
# Thought + Action
action_prompt = f"""
Question: {query}
Thoughts: {''.join(thoughts)}
Observations: {''.join(observations)}
What tool to use?
"""
action = self.llm.generate(action_prompt)
# Execute tool
observation = self._execute_tool(action)
observations.append(f"Observation: {observation}")
# Check if done
if self._is_sufficient(observations):
return self.llm.generate(
f"Answer based on: {''.join(observations)}"
)
thoughts.append(f"Step {step} complete")
Уровень 9: Evaluation
Metrics
Retrieval metrics:
- Recall@K: % of relevant docs in top-K
- MRR: mean reciprocal rank
- NDCG: normalized DCG
Generation metrics:
- Faithfulness: answer supported by context?
- Answer relevance: answer matches query?
- Context relevance: context needed for answer?
Combined:
RAGAS score = weighted average
RAGAS Framework
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevance
# Ground truth
test_data = {
"questions": ["What is LLM?"],
"contexts": [["LLM stands for..."]],
"answers": ["LLM is a large language model."],
"ground_truths": ["Large Language Model"]
}
# Evaluate
result = evaluate(test_data, metrics=[
faithfulness,
answer_relevance
])
print(result)
# faithfulness: 0.87
# answer_relevance: 0.92
Уровень 10: Best Practices
Strategy Selection
Choose your RAG strategy:
Simple (fast setup):
- Naive RAG
- Single embedding
- No reranking
Medium (better quality):
- Hybrid search
- Reranking
- Context compression
Advanced (best quality):
- Multi-query retrieval
- Cross-encoder reranker
- Graph RAG
- Self-reflective
Production Checklist
Production RAG:
[ ] Hybrid search (text + vector)
[ ] Cross-encoder reranker (top 100 → top 10)
[ ] Context compression
[ ] Evaluation pipeline (RAGAS)
[ ] Feedback loop (thumbs up/down)
[ ]定期 re-ingest with new data
[ ] Monitor latency and cost
Заключение
Advanced RAG techniques significantly improve retrieval quality.
Ключевые выводы:
- Query transformation improves recall
- Cross-encoder reranking is essential for quality
- Context compression saves cost and improves signal
- Graph RAG adds structured knowledge
- Evaluation with RAGAS is critical
Ресурсы: