Retrieval-Augmented Generation (RAG): полный гид
opensourceaillmragretrievaldatabasesit
Введение: Retrieval-Augmented Generation
Проблема
LLMs have limitations:
- Knowledge cutoff (trained on old data)
- No access to private data
- Hallucinations on specific topics
- Cannot access real-time information
Solution: Retrieval-Augmented Generation (RAG)
- Retrieve relevant documents
- Augment the prompt with context
- Generate answer from retrieved info
Факт: RAG can improve answer accuracy by 40-60% compared to pure LLM generation.
What is RAG?
RAG Architecture
RAG Pipeline:
1. Indexing Phase
Documents → Chunking → Embedding → Storage
2. Retrieval Phase
Query → Embedding → Similarity Search → Documents
3. Generation Phase
Query + Documents → Prompt → LLM → Answer
Basic RAG Flow
# Simple RAG implementation
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.llms import OpenAI
class SimpleRAG:
def __init__(self, documents, api_key):
# Create embeddings
embeddings = OpenAIEmbeddings(openai_api_key=api_key)
# Split documents into chunks
chunks = self.chunk_documents(documents)
# Create vector store
self.vectorstore = Chroma.from_documents(
chunks,
embeddings
)
# Initialize LLM
self.llm = OpenAI(temperature=0)
def chunk_documents(self, documents):
"""Split documents into chunks"""
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
)
return splitter.split_documents(documents)
def retrieve_and_generate(self, query, k=4):
"""Retrieve documents and generate answer"""
# Retrieve similar documents
docs = self.vectorstore.similarity_search(query, k=k)
# Build context
context = "\n\n".join([doc.page_content for doc in docs])
# Build prompt
prompt = f"""
Answer the question based on the following context:
Context:
{context}
Question: {query}
Answer:
"""
# Generate answer
return self.llm(prompt)
Document Chunking Strategies
Fixed-Size Chunking
def fixed_size_chunking(text, chunk_size=500, overlap=50):
"""
Split text into fixed-size chunks with overlap
"""
chunks = []
start = 0
while start < len(text):
end = start + chunk_size
chunks.append(text[start:end])
start = end - overlap
return chunks
# Example:
text = "Long document with many words..."
chunks = fixed_size_chunking(text, chunk_size=300, overlap=30)
# Result: ['Long document with many...', 'with many words and more...', ...]
Semantic Chunking
def semantic_chunking(text, embeddings, max_chunk_size=500):
"""
Split text at semantic boundaries
Uses embedding similarity to find natural breaks
"""
sentences = split_into_sentences(text)
chunks = []
current_chunk = [sentences[0]]
current_embedding = embeddings.embed_query(sentences[0])
for sentence in sentences[1:]:
sentence_embedding = embeddings.embed_query(sentence)
# Calculate similarity
similarity = cosine_similarity(
current_embedding, sentence_embedding
)
if similarity > 0.85: # Same topic
current_chunk.append(sentence)
else: # New topic
chunks.append(" ".join(current_chunk))
current_chunk = [sentence]
current_embedding = sentence_embedding
if current_chunk:
chunks.append(" ".join(current_chunk))
return chunks
# Semantic chunking preserves topic boundaries
# Better for retrieval quality
Parent-Document Retrieval
def parent_document_retrieval(documents, chunk_size=300):
"""
Store large parent documents, retrieve small chunks
Return full parent document when chunk matches
"""
# Split into small chunks
chunks = []
parents = {}
for doc in documents:
chunk_ids = []
for chunk in split(doc.content, chunk_size):
chunk_id = generate_id(chunk)
chunks.append({
"id": chunk_id,
"content": chunk,
"parent_id": doc.id
})
parents[chunk_id] = doc.content
chunk_ids.append(chunk_id)
doc.chunk_ids = chunk_ids
return chunks, parents
# When chunk is retrieved:
# 1. Get chunk_id from similarity search
# 2. Look up parent_id
# 3. Return full parent document as context
Embedding Models
Popular Embedding Models
# Embedding model comparison:
models = {
# OpenAI
"text-embedding-3-small": {
"dimensions": 1536,
"max_tokens": 8191,
"cost": "$0.02/1M tokens",
"similarity": 0.78, # MTEB score
},
"text-embedding-3-large": {
"dimensions": 3072,
"max_tokens": 8191,
"cost": "$0.13/1M tokens",
"similarity": 0.85,
},
# Open source
"sentence-transformers/all-MiniLM-L6-v2": {
"dimensions": 384,
"max_tokens": 256,
"cost": "Free (self-hosted)",
"similarity": 0.72,
"speed": "Fast (CPU)",
},
"BAAI/bge-large-en": {
"dimensions": 1024,
"max_tokens": 512,
"cost": "Free (self-hosted)",
"similarity": 0.81,
},
"jinaai/jina-embeddings-v3": {
"dimensions": 1024,
"max_tokens": 8192,
"cost": "Free (self-hosted)",
"similarity": 0.84,
},
}
# MTEB (Massive Text Embedding Benchmark) scores
# Higher = better embedding quality
Embedding Dimension Trade-offs
# Dimension vs. Quality vs. Speed
# Small (384-512 dims):
# - Fast similarity search
# - Less memory
# - Lower quality
# - Good for CPU inference
# Medium (768-1024 dims):
# - Balanced
# - Good quality/speed trade-off
# - Can run on GPU or CPU
# Large (1536-3072 dims):
# - Best quality
# - More memory
# - Slower search
# - Requires GPU for speed
def choose_embedding(use_case):
"""Choose embedding model based on use case"""
if use_case == "real-time-chat":
return "small" # Speed matters
elif use_case == "enterprise-search":
return "large" # Quality matters
else:
return "medium" # Balanced
Vector Databases
Chroma
import chromadb
from chromadb.config import Settings
# Create persistent client
client = chromadb.PersistentClient(path="./chroma-db")
# Create collection
collection = client.create_collection(
name="documents",
metadata={
"hnsw:space": "cosine", # Distance metric
"hnsw:M": 16, # Graph connectivity
"hnsw:search_ef": 10, # Search complexity
}
)
# Add documents
collection.add(
documents=[
"Document 1 content...",
"Document 2 content...",
],
metadatas=[
{"source": "file1.pdf", "page": 1},
{"source": "file2.pdf", "page": 3},
],
ids=["doc1", "doc2"]
)
# Query
results = collection.query(
query_texts=["search query"],
n_results=5,
where={"source": {"$eq": "file1.pdf"}}, # Filter
)
Pinecone
import pinecone
# Initialize
pinecone.init(api_key="your-api-key")
# Create index
pinecone.create_index(
name="my-index",
dimension=1536,
metric="cosine"
)
# Connect to index
index = pinecone.Index("my-index")
# Upsert vectors
index.upsert(
vectors=[
("doc1", embedding1, {"source": "file1"}),
("doc2", embedding2, {"source": "file2"}),
]
)
# Query
results = index.query(
vector=query_embedding,
top_k=5,
include_metadata=True,
)
Qdrant
from qdrant_client import QdrantClient
from qdrant_client.models import (
Distance,
VectorParams,
PointStruct,
)
# Initialize
client = QdrantClient(url="http://localhost:6333")
# Create collection
client.create_collection(
collection_name="documents",
vectors_config=VectorParams(
size=1536,
distance=Distance.COSINE,
),
)
# Add points
client.upsert(
collection_name="documents",
points=[
PointStruct(
id=1,
vector=embedding1,
payload={"source": "file1", "page": 1},
),
]
)
# Search
results = client.search(
collection_name="documents",
query_vector=query_embedding,
limit=5,
with_payload=True,
)
FAISS (Facebook AI Similarity Search)
import faiss
import numpy as np
# Create index
dimension = 1536
index = faiss.IndexFlatCosine(dimension)
# Add vectors
vectors = np.array(embeddings, dtype='float32')
index.add(vectors)
# Search
distances, indices = index.search(
np.array(query_embedding, dtype='float32'),
k=5
)
# Normalize for cosine similarity
faiss.normalize_L2(vectors)
index = faiss.IndexFlatIP(dimension) # Inner product = cosine
Retrieval Strategies
Similarity Search
# Cosine similarity
def cosine_similarity(a, b):
"""Calculate cosine similarity between two vectors"""
import numpy as np
return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
# Dot product
def dot_product_similarity(a, b):
"""Calculate dot product similarity"""
return np.dot(a, b)
# Euclidean distance
def euclidean_distance(a, b):
"""Calculate Euclidean distance"""
return np.linalg.norm(a - b)
# Convert distance to similarity
def distance_to_similarity(distance, distance_type="euclidean"):
"""Convert distance to similarity score"""
if distance_type == "euclidean":
return 1 / (1 + distance)
elif distance_type == "cosine":
return (1 + distance) / 2
Hybrid Search
# Combine vector + keyword search
def hybrid_search(query, vectorstore, keyword_index):
"""
Combine semantic and keyword-based retrieval
"""
# Vector search
vector_results = vectorstore.similarity_search(query, k=10)
# Keyword search (BM25)
keyword_results = keyword_index.search(query, k=10)
# Reciprocal Rank Fusion
fused = reciprocal_rank_fusion(
vector_results,
keyword_results,
k=60 # RRF constant
)
return fused[:5]
def reciprocal_rank_fusion(results_a, results_b, k=60):
"""
Reciprocal Rank Fusion
Combines results from two ranking systems
"""
scores = {}
for i, doc in enumerate(results_a):
scores[doc.id] = scores.get(doc.id, 0) + 1 / (k + i)
for i, doc in enumerate(results_b):
scores[doc.id] = scores.get(doc.id, 0) + 1 / (k + i)
# Sort by fused score
return sorted(scores.items(), key=lambda x: x[1], reverse=True)
Multi-Query Retrieval
# Generate multiple queries from different angles
def multi_query_retrieval(llm, original_query, vectorstore, k=5):
"""
Generate multiple queries and retrieve for each
"""
# Generate variations
query_gen_prompt = f"""
Generate 3 different queries that could be answered
by the same documents as: "{original_query}"
Return each query on a new line.
"""
variations = llm.generate(query_gen_prompt).split('\n')
# Retrieve for each query
all_results = []
for variation in variations:
results = vectorstore.similarity_search(variation, k=k)
all_results.extend(results)
# Deduplicate and return top results
unique = deduplicate_by_similarity(all_results)
return unique[:k]
RAG Variants
Naive RAG
Simple RAG:
Query → Embed → Retrieve → Concatenate → Generate
def naive_rag(query, vectorstore, llm):
docs = vectorstore.similarity_search(query, k=4)
context = "\n\n".join([d.page_content for d in docs])
prompt = f"Context: {context}\n\nQuestion: {query}\nAnswer:"
return llm(prompt)
Multi-Chunk RAG
Multi-Chunk RAG:
Query → Embed → Retrieve → Re-rank → Select → Generate
def multi_chunk_rag(query, vectorstore, reranker, llm):
# Retrieve more candidates
candidates = vectorstore.similarity_search(query, k=20)
# Re-rank with cross-encoder
ranked = reranker.candidates(candidates)
top_chunks = ranked[:4]
# Generate with context
context = format_context(top_chunks)
return llm(f"{context}\n\n{query}")
Self-RAG
Self-RAG:
Query → Retrieve → Evaluate relevance → Generate → Evaluate quality
def self_rag(query, vectorstore, llm):
# Retrieve
docs = vectorstore.similarity_search(query, k=5)
# Evaluate relevance
relevant_docs = []
for doc in docs:
relevance_score = evaluate_relevance(query, doc.content)
if relevance_score > 0.7:
relevant_docs.append(doc)
if not relevant_docs:
return "I don't have enough information to answer."
# Generate
context = "\n\n".join([d.content for d in relevant_docs])
answer = llm(f"Context: {context}\n\n{query}")
# Evaluate answer quality
quality = evaluate_answer(answer, query)
if quality < 0.5:
return "I'm not confident in this answer."
return answer
Agentic RAG
Agentic RAG:
Query → Agent decides: retrieve? search? ask follow-up? → Act
class AgenticRAG:
def __init__(self, llm, vectorstore, tools):
self.llm = llm
self.vectorstore = vectorstore
self.tools = tools
def process(self, query):
"""Agent loop for RAG"""
thoughts = []
while True:
# Decide next action
decision = self.llm.generate(f"""
Given query: {query}
Current thoughts: {thoughts}
Available tools: {self.tools}
What should I do next?
- retrieve: Search knowledge base
- answer: I have enough info
- ask: Need more info from user
- tool: Use a specific tool
""")
if decision == "answer":
return self.generate_final_answer(query, thoughts)
elif decision == "retrieve":
docs = self.vectorstore.similarity_search(query, k=5)
thoughts.append(f"Retrieved {len(docs)} documents")
elif decision == "tool":
result = self.execute_tool(query, thoughts)
thoughts.append(f"Tool result: {result}")
Evaluation
RAG Evaluation Metrics
# RAG evaluation dimensions
metrics = {
"retrieval_quality": {
"recall": "Did we retrieve all relevant docs?",
"precision": "Are retrieved docs relevant?",
"mrr": "Is the first relevant doc high ranked?",
},
"generation_quality": {
"faithfulness": "Does answer follow context?",
"answer_relevance": "Does answer address question?",
"context_utilization": "Was context used effectively?",
},
}
def evaluate_rag_system(questions, rag_system):
"""Evaluate RAG system end-to-end"""
results = {}
for q in questions:
answer, context = rag_system.generate(q)
# Evaluate retrieval
results[q] = {
"retrieval_recall": evaluate_recall(context, q.gold_docs),
"faithfulness": evaluate_faithfulness(answer, context),
"answer_relevance": evaluate_relevance(answer, q),
}
return aggregate(results)
RAGAS Framework
# RAGAS: Retrieval Augmented Generation Assessment
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevance
# Define evaluation dataset
dataset = Dataset.from_dict({
"questions": ["What is RAG?", ...],
"answers": [rag_system(q) for q in ...],
"contexts": [[retrieve(q) for _ in range(3)] for q in ...],
"ground_truths": ["RAG is...", ...],
})
# Evaluate
score = evaluate(
dataset=dataset,
metrics=[faithfulness, answer_relevance],
)
print(score)
# {
# "faithfulness": 0.82,
# "answer_relevance": 0.78,
# "context_precision": 0.85,
# }
Заключение
RAG is essential for:
- Up-to-date information: Retrieve latest data
- Private data: Access internal knowledge
- Citation: Show sources for answers
- Reduced hallucination: Ground answers in facts
Key components:
- Chunking: Split documents effectively
- Embeddings: Choose right model
- Vector DB: Scale and performance
- Retrieval: Hybrid search for best results
- Evaluation: Measure and improve
Best practices:
- Use hybrid search (vector + keyword)
- Re-rank retrieved documents
- Evaluate retrieval quality separately
- Monitor context utilization
- Iterate on chunking strategy