Production RAG: Production-ready RAG системы
ragproductionretrievalvector-dbpipeline
Введение
Production RAG системы отличаются от демо-версий сложной инфраструктурой, надёжностью и производительностью. Рассмотрим ключевые компоненты production RAG.
Production RAG Architecture
┌─────────────┐ ┌──────────────┐ ┌─────────────┐
│ Client │────▶│ API Layer │────▶│ Query │
│ (Web/Mobile)│ │ (FastAPI) │ │ Processing │
└─────────────┘ └──────────────┘ └─────────────┘
│
┌─────────────────────────┼─────────────────────────┐
│ │ │
┌─────▼─────┐ ┌──────▼──────┐ ┌──────▼──────┐
│ Query │ │ Embedding │ │ Query │
│ Rewrite │ │ Generation │ │ Routing │
│ & Expand │ │ │ │ │
└───────────┘ └─────────────┘ └─────────────┘
│
┌─────────────────────────┼─────────────────────────┐
│ │ │
┌─────▼─────┐ ┌──────▼──────┐ ┌──────▼──────┐
│ Hybrid │ │ Vector │ │ Keyword │
│ Search │ │ Search │ │ Search │
│ (sparse) │ │ (dense) │ │ (BM25) │
└───────────┘ └─────────────┘ └─────────────┘
│ │ │
┌─────▼─────────────────────────▼─────────────────────────▼─────┐
│ Reranker │
│ (Cross-Encoder Re-ranking) │
└─────────────────────────┬───────────────────────────────────┘
│
┌─────────▼─────────┐
│ LLM Generator │
│ (with context) │
└─────────┬─────────┘
│
┌─────────▼─────────┐
│ Response │
│ Post-process │
└─────────┬─────────┘
│
┌─────────▼─────────┐
│ Cache & Log │
└───────────────────┘
Document Ingestion Pipeline
Chunking strategies
from langchain.text_splitter import (
RecursiveCharacterTextSplitter,
TokenTextSplitter,
MarkdownTextSplitter,
CodeTextSplitter
)
# Recursive chunking — default
recursive_splitter = RecursiveCharacterTextSplitter(
chunk_size=512,
chunk_overlap=64,
separators=["\n\n", "\n", ". ", " ", ""]
)
# Token-based chunking
token_splitter = TokenTextSplitter(
chunk_size=256,
chunk_overlap=32,
encoding_name="cl100k_base"
)
# Markdown-aware chunking
markdown_splitter = MarkdownTextSplitter(
chunk_size=512,
chunk_overlap=50
)
# Code-aware chunking
code_splitter = CodeTextSplitter(
chunk_size=768,
chunk_overlap=64
)
def chunk_document(text: str, strategy: str = "recursive") -> list[str]:
if strategy == "recursive":
splitter = recursive_splitter
elif strategy == "token":
splitter = token_splitter
elif strategy == "markdown":
splitter = markdown_splitter
elif strategy == "code":
splitter = code_splitter
else:
raise ValueError(f"Unknown strategy: {strategy}")
return splitter.split_text(text)
Metadata enrichment
from langchain.docstore.document import Document
from typing import Dict, Any
def enrich_metadata(text: str, source: str) -> Dict[str, Any]:
"""Add metadata to document chunk"""
return {
"source": source,
"word_count": len(text.split()),
"char_count": len(text),
"language": detect_language(text),
"has_code": "```" in text,
"is_heading": text.startswith("#"),
"estimated_reading_time": len(text.split()) / 200, # minutes
}
def process_document(file_path: str, chunk_size: int = 512) -> list[Document]:
"""Process a single document into chunks with metadata"""
# Read document
content = read_file(file_path)
# Split into chunks
chunks = chunk_document(content, strategy="recursive")
# Create documents with metadata
documents = []
for i, chunk in enumerate(chunks):
metadata = enrich_metadata(chunk, file_path)
metadata["chunk_index"] = i
metadata["total_chunks"] = len(chunks)
doc = Document(
page_content=chunk,
metadata=metadata
)
documents.append(doc)
return documents
Batch ingestion with progress tracking
import time
from dataclasses import dataclass
from typing import Optional
@dataclass
class IngestionStats:
total_docs: int
total_chunks: int
successful: int
failed: int
duration_seconds: float
chunks_per_second: float
class DocumentIngestor:
def __init__(self, vector_store, embedding_model, batch_size=32):
self.vector_store = vector_store
self.embedding_model = embedding_model
self.batch_size = batch_size
def ingest_batch(self, documents: list[Document]) -> IngestionStats:
start_time = time.time()
total_docs = len(documents)
successful = 0
failed = 0
# Process in batches
for i in range(0, len(documents), self.batch_size):
batch = documents[i:i + self.batch_size]
try:
# Generate embeddings
texts = [doc.page_content for doc in batch]
embeddings = self.embedding_model.embed_documents(texts)
# Upsert to vector store
self.vector_store.add_embeddings(
embeddings=embeddings,
metadatas=[doc.metadata for doc in batch],
documents=texts
)
successful += len(batch)
except Exception as e:
print(f"Error ingesting batch {i}: {e}")
failed += len(batch)
# Progress logging
if (i + self.batch_size) % 100 == 0:
progress = (i + self.batch_size) / total_docs * 100
print(f"Progress: {progress:.1f}%")
duration = time.time() - start_time
return IngestionStats(
total_docs=total_docs,
total_chunks=successful + failed,
successful=successful,
failed=failed,
duration_seconds=duration,
chunks_per_second=(successful + failed) / duration
)
Embedding Models for Production
Model selection
EMBEDDING_MODELS = {
# Production-ready models
"text-embedding-3-small": {
"dimensions": 1536,
"max_tokens": 8191,
"cost_per_1k_tokens": 0.00002,
"speed": "fast",
"quality_score": 0.65, # MTEB score
},
"text-embedding-3-large": {
"dimensions": 3072,
"max_tokens": 8191,
"cost_per_1k_tokens": 0.00013,
"speed": "medium",
"quality_score": 0.76,
},
"nomic-embed-text": {
"dimensions": 768,
"max_tokens": 8192,
"cost_per_1k_tokens": 0.0, # self-hosted
"speed": "fast",
"quality_score": 0.62,
},
"bge-large-en": {
"dimensions": 1024,
"max_tokens": 512,
"cost_per_1k_tokens": 0.0,
"speed": "medium",
"quality_score": 0.70,
},
"e5-large-v2": {
"dimensions": 1024,
"max_tokens": 512,
"cost_per_1k_tokens": 0.0,
"speed": "medium",
"quality_score": 0.64,
},
}
Embedding service
import torch
from sentence_transformers import SentenceTransformer
from typing import List
class EmbeddingService:
def __init__(self, model_name: str = "nomic-embed-text", device: str = "cuda"):
self.model_name = model_name
self.device = device
self.model = SentenceTransformer(model_name, device=device)
# Batch processing
self.batch_size = 64 if device == "cuda" else 16
def embed_documents(self, texts: List[str]) -> List[List[float]]:
"""Embed multiple documents"""
all_embeddings = []
for i in range(0, len(texts), self.batch_size):
batch = texts[i:i + self.batch_size]
embeddings = self.model.encode(
batch,
batch_size=self.batch_size,
show_progress_bar=False,
normalize_embeddings=True
)
all_embeddings.extend(embeddings.tolist())
return all_embeddings
def embed_query(self, text: str) -> List[float]:
"""Embed a single query"""
embedding = self.model.encode(
text,
normalize_embeddings=True
)
return embedding.tolist()
def get_dimension(self) -> int:
return self.model.get_sentence_embedding_dimension()
Hybrid Search
Dense + Sparse combination
from typing import List, Tuple
class HybridSearcher:
def __init__(self, vector_store, bm25_index):
self.vector_store = vector_store
self.bm25_index = bm25_index
self.dense_weight = 0.5
self.sparse_weight = 0.5
self.reranker = None
def search(
self,
query: str,
top_k: int = 10,
filters: dict = None
) -> List[Tuple[Document, float]]:
# Dense search
query_embedding = self.embedding_service.embed_query(query)
dense_results = self.vector_store.similarity_search_with_score(
query_embedding,
k=top_k * 2,
filters=filters
)
# Sparse search (BM25)
sparse_results = self.bm25_index.search(
query,
top_k=top_k * 2,
filters=filters
)
# Normalize and combine scores
combined = self._reciprocal_fusion(dense_results, sparse_results, top_k)
# Rerank if reranker is available
if self.reranker:
combined = self._rerank(query, combined[:top_k * 2])
return combined[:top_k]
def _reciprocal_fusion(
self,
dense_results: list,
sparse_results: list,
top_k: int
) -> list:
"""Reciprocal rank fusion for combining results"""
score_dict = {}
for rank, (doc, score) in enumerate(dense_results, 1):
doc_id = doc.metadata.get("id")
if doc_id not in score_dict:
score_dict[doc_id] = {
"doc": doc,
"dense_score": 0,
"sparse_score": 0,
"fusion_score": 0
}
score_dict[doc_id]["dense_score"] = score
score_dict[doc_id]["fusion_score"] += 1 / rank
for rank, (doc, score) in enumerate(sparse_results, 1):
doc_id = doc.metadata.get("id")
if doc_id not in score_dict:
score_dict[doc_id] = {
"doc": doc,
"dense_score": 0,
"sparse_score": 0,
"fusion_score": 0
}
score_dict[doc_id]["sparse_score"] = score
score_dict[doc_id]["fusion_score"] += 1 / rank
# Normalize fusion scores
for doc_id in score_dict:
score_dict[doc_id]["fusion_score"] /= (
self.dense_weight + self.sparse_weight
)
# Sort by fusion score
sorted_results = sorted(
score_dict.values(),
key=lambda x: x["fusion_score"],
reverse=True
)
return [(item["doc"], item["fusion_score"]) for item in sorted_results[:top_k]]
Query Processing
Query rewriting and expansion
from langchain.prompts import PromptTemplate
from langchain.chat_models import ChatOpenAI
class QueryProcessor:
def __init__(self, llm=None):
self.llm = llm or ChatOpenAI(model="gpt-4o-mini")
self.rewrite_prompt = PromptTemplate(
input_variables=["query"],
template="""
Given a user query, rewrite it to improve retrieval.
Make it more specific and include relevant keywords.
Original query: {query}
Rewritten query:
"""
)
self.expand_prompt = PromptTemplate(
input_variables=["query"],
template="""
Given a user query, generate 3 alternative formulations
that could retrieve relevant documents.
Original query: {query}
Alternatives:
1.
2.
3.
"""
)
def rewrite_query(self, query: str) -> str:
"""Rewrite query for better retrieval"""
chain = self.rewrite_prompt | self.llm
response = chain.invoke({"query": query})
return response.content.strip()
def expand_query(self, query: str) -> List[str]:
"""Expand query into multiple formulations"""
chain = self.expand_prompt | self.llm
response = chain.invoke({"query": query})
# Parse alternatives
lines = response.content.strip().split("\n")
alternatives = [
line.strip().lstrip("123.")
for line in lines
if line.strip() and any(line.strip().startswith(f"{i}.") for i in range(1, 4))
]
return [query] + alternatives
def decompose_query(self, query: str) -> List[str]:
"""Decompose complex query into sub-queries"""
decompose_prompt = PromptTemplate(
input_variables=["query"],
template="""
Break down this query into sub-questions:
Query: {query}
Sub-questions:
"""
)
chain = decompose_prompt | self.llm
response = chain.invoke({"query": query})
return [q.strip() for q in response.content.strip().split("\n") if q.strip()]
Reranking
Cross-encoder reranker
from sentence_transformers import CrossEncoder
from typing import List, Tuple
class Reranker:
def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"):
self.model = CrossEncoder(model_name)
def rerank(
self,
query: str,
documents: List[Tuple[str, float]],
top_k: int = 5
) -> List[Tuple[str, float, float]]:
"""Rerank documents using cross-encoder"""
if not documents:
return []
# Prepare pairs
pairs = [[query, doc[0]] for doc in documents]
# Get scores
scores = self.model.predict(pairs)
# Combine with original scores
reranked = []
for (doc, original_score), new_score in zip(documents, scores):
# Weighted combination
combined_score = 0.3 * original_score + 0.7 * new_score
reranked.append((doc, original_score, new_score, combined_score))
# Sort and return top-k
reranked.sort(key=lambda x: x[3], reverse=True)
return reranked[:top_k]
Caching layer
Response caching
import hashlib
import json
import time
from typing import Optional
class ResponseCache:
def __init__(self, ttl: int = 3600, max_size: int = 10000):
self.ttl = ttl
self.max_size = max_size
self._cache = {}
self._access_order = []
def _make_key(self, query: str, top_k: int, filters: dict) -> str:
"""Generate cache key"""
key_data = json.dumps({
"query": query,
"top_k": top_k,
"filters": filters
}, sort_keys=True)
return hashlib.sha256(key_data.encode()).hexdigest()
def get(self, query: str, top_k: int = 5, filters: dict = None) -> Optional[list]:
key = self._make_key(query, top_k, filters)
if key not in self._cache:
return None
entry = self._cache[key]
if time.time() - entry["timestamp"] > self.ttl:
del self._cache[key]
self._access_order.remove(key)
return None
return entry["results"]
def set(self, query: str, results: list, top_k: int = 5, filters: dict = None):
key = self._make_key(query, top_k, filters)
# Evict oldest if at capacity
if len(self._cache) >= self.max_size and key not in self._cache:
oldest = self._access_order.pop(0)
del self._cache[oldest]
self._cache[key] = {
"results": results,
"timestamp": time.time()
}
self._access_order.append(key)
API Layer
FastAPI RAG service
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, Field
from typing import List, Optional, Dict
app = FastAPI(title="RAG Service")
class RAGRequest(BaseModel):
query: str = Field(..., min_length=1, max_length=500)
top_k: int = Field(default=5, ge=1, le=20)
use_cache: bool = True
filters: Optional[Dict[str, str]] = None
temperature: float = Field(default=0.7, ge=0, le=2)
class RAGResponse(BaseModel):
answer: str
sources: List[Dict[str, any]]
latency_ms: int
cached: bool
query_processed: str
@app.post("/rag", response_model=RAGResponse)
async def rag_endpoint(request: RAGRequest):
start_time = time.time()
# Check cache
if request.use_cache:
cached_results = response_cache.get(
request.query,
request.top_k,
request.filters
)
if cached_results:
latency = int((time.time() - start_time) * 1000)
return RAGResponse(
answer=cached_results["answer"],
sources=cached_results["sources"],
latency_ms=latency,
cached=True,
query_processed=request.query
)
# Process query
processed_query = query_processor.expand_query(request.query)
# Search
all_results = []
for pq in processed_query:
results = hybrid_searcher.search(pq, top_k=request.top_k * 2, filters=request.filters)
all_results.extend(results)
# Deduplicate and rerank
unique_docs = {}
for doc, score in all_results:
doc_id = doc.metadata.get("id")
if doc_id not in unique_docs or score > unique_docs[doc_id][1]:
unique_docs[doc_id] = (doc, score)
reranked = reranker.rerank(
request.query,
[(doc.page_content, score) for doc, score in unique_docs.values()],
top_k=request.top_k
)
# Generate answer
context = "\n\n".join([doc[0] for doc in reranked])
answer = generate_answer(request.query, context, request.temperature)
# Cache result
if request.use_cache:
response_cache.set(
request.query,
{"answer": answer, "sources": [doc[0] for doc in reranked]},
request.top_k,
request.filters
)
latency = int((time.time() - start_time) * 1000)
return RAGResponse(
answer=answer,
sources=[{"content": doc[0], "score": doc[1]} for doc in reranked],
latency_ms=latency,
cached=False,
query_processed=processed_query[0]
)
@app.get("/health")
async def health():
return {"status": "healthy"}
Error handling and fallbacks
Graceful degradation
class RAGService:
def __init__(self):
self.vector_store = get_vector_store()
self.embedding_service = EmbeddingService()
self.reranker = Reranker()
self.cache = ResponseCache()
async def query(self, query: str, **kwargs) -> dict:
try:
return await self._full_rag_pipeline(query, **kwargs)
except Exception as e:
logger.error(f"RAG failed: {e}")
# Fallback 1: Try with fewer sources
try:
return await self._full_rag_pipeline(query, top_k=3, **kwargs)
except Exception as e2:
logger.error(f"Fallback 1 failed: {e2}")
# Fallback 2: Direct LLM call
try:
llm = get_llm()
response = await llm.ainvoke(f"Answer based on general knowledge: {query}")
return {
"answer": response.content,
"sources": [],
"fallback": True
}
except Exception as e3:
logger.error(f"Fallback 2 failed: {e3}")
# Fallback 3: Error message
return {
"answer": "Sorry, I'm unable to process your request right now.",
"sources": [],
"fallback": True,
"error": "All RAG pipelines failed"
}
Итоги
Production RAG требует: надёжного ingestion pipeline, hybrid search, reranking, caching и graceful degradation. Тестируйте с real-world queries и monitor latency/accuracy metrics.