RAG Systems: Retrieval-Augmented Generation с локальными моделями
ragretrievallocal-llmembeddingsdatabases
Что такое RAG?
RAG (Retrieval-Augmented Generation) — это подход, который комбинирует поиск информации с генерацией ответов. Вместо того чтобы полагаться только на параметры модели, RAG извлекает релевантные документы из базы знаний и использует их как контекст для генерации ответа.
┌─────────────┐ ┌──────────────┐ ┌─────────────┐
│ User │────▶│ Retriever │────▶│ Documents │
│ Query │ │ (Embedding │ │ (Chunks) │
│ │ │ + Search) │ │ │
└─────────────┘ └──────────────┘ └─────────────┘
│
▼
┌─────────────┐ ┌──────────────┐ ┌─────────────┐
│ Final │◀────│ Generator │◀────│ Context │
│ Answer │ │ (LLM) │ │ + Query │
└─────────────┘ └──────────────┘ └─────────────┘
Почему RAG с локальными моделями?
- Конфиденциальность — данные не покидают ваш сервер
- Кастомные данные — модель знает только ваши документы
- Контроль — полный контроль над пайплайном
- Стоимость — нет API-затрат на каждый запрос
Архитектура RAG системы
1. Индексация документов
import os
from pathlib import Path
from typing import List, Dict
import hashlib
class DocumentIndexer:
"""Индексация документов для RAG"""
def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
def load_documents(self, directory: str) -> List[Dict]:
"""Загрузка документов из директории"""
documents = []
path = Path(directory)
for file_path in path.rglob('*'):
if file_path.suffix in ['.md', '.txt', '.pdf', '.docx']:
content = file_path.read_text(encoding='utf-8')
doc_id = hashlib.md5(str(file_path).encode()).hexdigest()[:12]
documents.append({
'id': doc_id,
'source': str(file_path),
'content': content,
'metadata': {
'filename': file_path.name,
'extension': file_path.suffix,
'size': file_path.stat().st_size
}
})
return documents
def chunk_document(self, content: str, metadata: Dict = None) -> List[Dict]:
"""Разбиение документа на чанки"""
chunks = []
words = content.split()
if len(words) <= self.chunk_size // 5: # ~5 chars per word
return [{
'content': content,
'metadata': metadata or {},
'chunk_index': 0
}]
start = 0
chunk_index = 0
while start < len(words):
end = start + self.chunk_size // 5
chunk = ' '.join(words[start:end])
chunks.append({
'content': chunk,
'metadata': {
**(metadata or {}),
'chunk_index': chunk_index,
'start_word': start,
'end_word': end
},
'chunk_index': chunk_index
})
start = end - self.chunk_overlap
chunk_index += 1
return chunks
def index_directory(self, directory: str) -> List[Dict]:
"""Полный пайплайн индексации"""
documents = self.load_documents(directory)
all_chunks = []
for doc in documents:
chunks = self.chunk_document(doc['content'], doc['metadata'])
all_chunks.extend(chunks)
return all_chunks
2. Embedding модели
import torch
from transformers import AutoTokenizer, AutoModel
from sentence_transformers import SentenceTransformer
import numpy as np
class EmbeddingService:
"""Сервис генерации эмбеддингов"""
def __init__(self, model_name: str = 'sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2'):
self.model = SentenceTransformer(model_name)
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.device = 'cuda' if torch.cuda.is_available() else 'cpu'
self.model.to(self.device)
def embed_documents(self, texts: List[str]) -> np.ndarray:
"""Генерация эмбеддингов для списка документов"""
embeddings = self.model.encode(
texts,
batch_size=32,
show_progress_bar=True,
convert_to_numpy=True
)
# L2 normalize
norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
embeddings = embeddings / (norms + 1e-8)
return embeddings
def embed_query(self, text: str) -> np.ndarray:
"""Генерация эмбеддинга для запроса"""
embedding = self.model.encode(text, convert_to_numpy=True)
norm = np.linalg.norm(embedding)
return embedding / (norm + 1e-8)
def get_embedding_dim(self) -> int:
"""Размерность эмбеддингов"""
return self.model.get_sentence_embedding_dimension()
# Сравнение моделей
EMBEDDING_MODELS = {
'paraphrase-multilingual-MiniLM-L12-v2': {
'params': '115M',
'dim': 384,
'speed': 'fast',
'quality': 'good',
'languages': 50+
},
'paraphrase-multilingual-mpnet-base-v2': {
'params': '278M',
'dim': 768,
'speed': 'medium',
'quality': 'excellent',
'languages': 100+
},
'all-MiniLM-L6-v2': {
'params': '22M',
'dim': 384,
'speed': 'very fast',
'quality': 'good',
'languages': 1 (English)
},
'bge-large-multilingual-v1.5': {
'params': '568M',
'dim': 1024,
'speed': 'slow',
'quality': 'excellent',
'languages': 100+
}
}
3. Векторные базы данных
import sqlite3
import json
import numpy as np
from typing import List, Dict, Optional, Tuple
import heapq
class SimpleVectorStore:
"""Простой векторный store на SQLite + numpy"""
def __init__(self, db_path: str = 'vectors.db'):
self.db_path = db_path
self.conn = sqlite3.connect(db_path)
self._create_tables()
def _create_tables(self):
cursor = self.conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS chunks (
id TEXT PRIMARY KEY,
content TEXT NOT NULL,
metadata TEXT,
embedding_blob BLOB,
dim INTEGER
)
''')
self.conn.commit()
def add_chunks(self, chunks: List[Dict], embeddings: np.ndarray):
"""Добавление чанков с эмбеддингами"""
cursor = self.conn.cursor()
for chunk, embedding in zip(chunks, embeddings):
chunk_id = f"{chunk['metadata'].get('source', 'unknown')}_{chunk['chunk_index']}"
cursor.execute('''
INSERT OR REPLACE INTO chunks
(id, content, metadata, embedding_blob, dim)
VALUES (?, ?, ?, ?, ?)
''', (
chunk_id,
chunk['content'],
json.dumps(chunk.get('metadata', {})),
embedding.tobytes(),
len(embedding)
))
self.conn.commit()
def search(self, query_embedding: np.ndarray, top_k: int = 5) -> List[Dict]:
"""Поиск похожих чанков"""
cursor = self.conn.cursor()
cursor.execute('SELECT id, content, metadata, embedding_blob, dim FROM chunks')
rows = cursor.fetchall()
scores = []
for row in rows:
embedding = np.frombuffer(row[3], dtype=np.float32)
score = np.dot(query_embedding, embedding)
scores.append((score, row))
# Top-k
scores.sort(reverse=True)
results = []
for score, row in scores[:top_k]:
results.append({
'id': row[0],
'content': row[1],
'metadata': json.loads(row[2]),
'score': float(score)
})
return results
# ChromaDB пример
import chromadb
class ChromaVectorStore:
"""Векторный store на ChromaDB"""
def __init__(self, collection_name: str = 'documents'):
self.client = chromadb.Client()
self.collection = self.client.get_or_create_collection(
name=collection_name,
metadata={"hnsw:space": "cosine"}
)
def add_documents(self, documents: List[Dict], embeddings: np.ndarray):
"""Добавление документов в ChromaDB"""
ids = [
f"{doc['metadata'].get('source', 'unknown')}_{doc['chunk_index']}"
for doc in documents
]
self.collection.upsert(
ids=ids,
documents=[doc['content'] for doc in documents],
metadatas=[doc.get('metadata', {}) for doc in documents],
embeddings=embeddings.tolist()
)
def search(self, query_embedding: np.ndarray, top_k: int = 5) -> List[Dict]:
"""Поиск в ChromaDB"""
results = self.collection.query(
query_embeddings=[query_embedding.tolist()],
n_results=top_k,
include=['documents', 'metadatas', 'distances']
)
return [
{
'content': docs[i],
'metadata': results['metadatas'][0][i],
'score': 1.0 - results['distances'][0][i] # Convert distance to similarity
}
for i in range(len(results['ids'][0]))
]
4. Поиск с перекомбинацией (Reranking)
from transformers import AutoModelForSequenceClassification, AutoTokenizer
class Reranker:
"""Реранкер для улучшения качества поиска"""
def __init__(self, model_name: str = 'BAAI/bge-reranker-base'):
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
self.device = 'cuda' if torch.cuda.is_available() else 'cpu'
self.model.to(self.device)
self.model.eval()
@torch.no_grad()
def rerank(self, query: str, documents: List[str], top_k: int = 3) -> List[Dict]:
"""Реранкинг документов"""
pairs = [[query, doc] for doc in documents]
inputs = self.tokenizer(
pairs,
padding=True,
truncation=True,
return_tensors='pt',
max_length=512
).to(self.device)
scores = self.model(**inputs).logits.squeeze().cpu().numpy()
# Sigmoid to get probabilities
import scipy.special
probabilities = scipy.special.expit(scores)
# Get top-k
indexed_scores = list(enumerate(probabilities))
indexed_scores.sort(key=lambda x: x[1], reverse=True)
return [
{
'content': documents[idx],
'score': float(prob),
'original_index': idx
}
for idx, prob in indexed_scores[:top_k]
]
5. Генерация ответа
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
class RAGGenerator:
"""Генератор ответов для RAG"""
def __init__(self, model_name: str = 'Qwen/Qwen2.5-7B-Instruct'):
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.pipe = pipeline(
"text-generation",
model=model_name,
tokenizer=model_name,
device=0 if torch.cuda.is_available() else -1,
max_new_tokens=1024,
temperature=0.7,
top_p=0.9
)
def generate(self, query: str, context: List[str]) -> str:
"""Генерация ответа на основе контекста"""
context_text = '\n\n'.join(context)
prompt = f"""Ответь на вопрос, используя предоставленные документы.
Если в документах нет ответа, скажи "Я не нашел достаточно информации для ответа".
Документы:
{context_text}
Вопрос: {query}
Ответ:"""
result = self.pipe(prompt, max_new_tokens=1024, temperature=0.7, top_p=0.9)
return result[0]['generated_text']
Полный RAG пайплайн
class RAGPipeline:
"""Полный RAG пайплайн"""
def __init__(
self,
embedding_model: str = 'sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2',
rag_model: str = 'Qwen/Qwen2.5-7B-Instruct',
vector_store_path: str = 'chroma_db'
):
self.embedder = EmbeddingService(embedding_model)
self.generator = RAGGenerator(rag_model)
self.vector_store = ChromaVectorStore()
self.indexed = False
def index_documents(self, directory: str):
"""Индексация документов"""
indexer = DocumentIndexer()
chunks = indexer.index_directory(directory)
texts = [chunk['content'] for chunk in chunks]
embeddings = self.embedder.embed_documents(texts)
self.vector_store.add_documents(chunks, embeddings)
self.indexed = True
print(f"Индексировано {len(chunks)} чанков")
def query(self, question: str, top_k: int = 5) -> Dict:
"""Запрос к RAG системе"""
if not self.indexed:
raise ValueError("Сначала проиндексируйте документы")
# 1. Embedding запроса
query_embedding = self.embedder.embed_query(question)
# 2. Поиск
results = self.vector_store.search(query_embedding, top_k=top_k)
# 3. Реранкинг (опционально)
# reranker = Reranker()
# results = reranker.rerank(question, [r['content'] for r in results], top_k=3)
# 4. Генерация
contexts = [r['content'] for r in results]
answer = self.generator.generate(question, contexts)
return {
'query': question,
'answer': answer,
'sources': [
{
'content': r['content'][:200] + '...',
'score': r['score'],
'metadata': r.get('metadata', {})
}
for r in results
]
}
Продвинутые техники
1. Hybrid Search (BM25 + Vector)
from rank_bm25 import BM25Okapi
import re
class HybridSearch:
"""Гибридный поиск: BM25 + Vector"""
def __init__(self):
self.bm25 = None
self.vector_store = None
self.tokenizer = lambda text: re.findall(r'\w+', text.lower())
def index(self, documents: List[str], embeddings: np.ndarray):
"""Индексация для гибридного поиска"""
# BM25
tokenized_docs = [self.tokenizer(doc) for doc in documents]
self.bm25 = BM25Okapi(tokenized_docs)
# Vector
self.vector_store = ChromaVectorStore()
self.vector_store.add_documents(
[{'content': doc, 'metadata': {}} for doc in documents],
embeddings
)
def search(self, query: str, top_k: int = 5, alpha: float = 0.5) -> List[Dict]:
"""
Гибридный поиск
alpha: вес векторного поиска (1-alpha) - вес BM25
"""
# BM25 scores
bm25_scores = self.bm25.get_scores(self.tokenizer(query))
# Vector search
vector_results = self.vector_store.search(
self.embedder.embed_query(query),
top_k=top_k * 2
)
# Normalize BM25 scores
max_bm25 = max(bm25_scores) if max(bm25_scores) > 0 else 1
normalized_bm25 = {
i: score / max_bm25
for i, score in enumerate(bm25_scores)
}
# Combine scores
combined = {}
for result in vector_results:
doc_id = result['metadata'].get('source', '')
# Find document index
for i, doc in enumerate(self.documents):
if doc_id in doc:
vector_score = result['score']
bm25_score = normalized_bm25.get(i, 0)
combined[doc_id] = {
'score': alpha * vector_score + (1 - alpha) * bm25_score,
'content': doc,
'vector_score': vector_score,
'bm25_score': bm25_score
}
break
# Sort by combined score
sorted_results = sorted(combined.values(), key=lambda x: x['score'], reverse=True)
return sorted_results[:top_k]
2. Multi-query Retrieval
class MultiQueryRAG:
"""RAG с множественными запросами"""
def __init__(self, model_name: str = 'Qwen/Qwen2.5-7B-Instruct'):
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.model = AutoModelForCausalLM.from_pretrained(model_name)
def generate_queries(self, original_query: str, num_queries: int = 3) -> List[str]:
"""Генерация альтернативных запросов"""
prompt = f"""Generate {num_queries} different versions of the following question
to improve retrieval in a RAG system. Each version should phrase the question
differently but seek the same information.
Original question: {original_query}
Generated questions (one per line):"""
# Use LLM to generate queries
# ... (implementation depends on your LLM setup)
return [
f"What are the best practices for {original_query}?",
f"How to implement {original_query}?",
f"{original_query} tutorial and guide"
]
def query(self, original_query: str, vector_store, embedder: EmbeddingService):
"""Multi-query retrieval"""
# Generate alternative queries
queries = self.generate_queries(original_query)
# Search for each query
all_results = []
for query in queries:
embedding = embedder.embed_query(query)
results = vector_store.search(embedding, top_k=5)
all_results.extend(results)
# Deduplicate by content similarity
unique_results = self._deduplicate(all_results)
return unique_results[:5]
def _deduplicate(self, results: List[Dict], threshold: float = 0.9) -> List[Dict]:
"""Удаление дубликатов по similarity threshold"""
unique = []
seen_contents = []
for result in results:
is_duplicate = False
for seen in seen_contents:
sim = np.dot(
self._embed(result['content']),
self._embed(seen)
)
if sim > threshold:
is_duplicate = True
break
if not is_duplicate:
unique.append(result)
seen_contents.append(result['content'])
return unique
Производительность и оптимизация
Benchmarks
| Компонент | Модель | Время/запрос | Качество |
|---|---|---|---|
| Embedding (small) | MiniLM-L6 | 10ms | Good |
| Embedding (medium) | MiniLM-L12 | 25ms | Good |
| Embedding (large) | bge-large | 80ms | Excellent |
| Vector Search (1M) | ChromaDB | 5ms | N/A |
| Vector Search (10M) | ChromaDB | 20ms | N/A |
| Reranker | bge-reranker | 50ms | +15% quality |
| LLM (7B) | Qwen2.5-7B | 2s | N/A |
Оптимизации
- Кэширование эмбеддингов — не пересчитывайте для одних и тех же документов
- Batch processing — обрабатывайте запросы батчами
- Quantized embedding models — используйте INT8 модели
- HNSW индекс — для быстрого поиска в больших наборах
- Reranking только top-20 — не рераньте все результаты
Заключение
RAG с локальными моделями — это мощный подход для создания систем, которые знают ваши данные. Ключевые компоненты: хороший эмбеддинг-модель, векторная база данных, и качественный LLM для генерации. Hybrid search и reranking значительно улучшают качество.