🐍 Python RAG Песочница

Базовый пример RAG

import math
import re
from collections import Counter

class SimpleRAG:
    """RAG система на чистом Python без внешних зависимостей"""
    
    def __init__(self):
        self.documents = []
        self.doc_vectors = []
        self.vocabulary = set()
        
    def tokenize(self, text):
        """Простая токенизация текста"""
        text = text.lower()
        tokens = re.findall(r'\b[а-яa-z]+\b', text)
        return tokens
    
    def create_vector(self, text, vocabulary):
        """Создание вектора из текста (bag-of-words)"""
        tokens = self.tokenize(text)
        token_counts = Counter(tokens)
        
        vector = []
        for word in sorted(vocabulary):
            vector.append(token_counts.get(word, 0))
        
        # Нормализация вектора
        magnitude = math.sqrt(sum(x**2 for x in vector))
        if magnitude > 0:
            vector = [x/magnitude for x in vector]
        
        return vector
    
    def cosine_similarity(self, vec1, vec2):
        """Вычисление косинусного сходства"""
        return sum(a * b for a, b in zip(vec1, vec2))
    
    def add_documents(self, docs):
        """📍 Шаги 1-2: Encode + Index документов в Vector Database"""
        print("📍 Шаг 1: Encode - Преобразование документов в векторы")
        self.documents = docs
        
        # Построение словаря
        for doc in docs:
            tokens = self.tokenize(doc)
            self.vocabulary.update(tokens)
        
        # Создание векторов для каждого документа
        for i, doc in enumerate(docs, 1):
            vector = self.create_vector(doc, self.vocabulary)
            self.doc_vectors.append(vector)
            print(f"   Документ {i}: вектор размерности {len(vector)}")
        
        print(f"📍 Шаг 2: Index - Сохранено {len(docs)} векторов в базе")
        return len(docs)
    
    def search(self, query, k=2):
        """📍 Шаги 3-5: Query Encoding + Similarity Search + Retrieve"""
        print(f"\n📍 Шаг 3: Encode Query - '{query}'")
        
        # Векторизация запроса
        query_vector = self.create_vector(query, self.vocabulary)
        print(f"   Вектор запроса создан (размерность: {len(query_vector)})")
        
        print("📍 Шаг 4: Similarity Search - Поиск в Vector Database")
        
        # Вычисление сходства со всеми документами
        similarities = []
        for i, doc_vec in enumerate(self.doc_vectors):
            sim = self.cosine_similarity(query_vector, doc_vec)
            similarities.append((i, sim))
            print(f"   Документ {i+1}: сходство = {sim:.3f}")
        
        # Сортировка по убыванию сходства
        similarities.sort(key=lambda x: x[1], reverse=True)
        
        print(f"📍 Шаг 5: Retrieve - Извлечение топ-{k} документов")
        
        # Возврат топ-k результатов
        results = []
        for idx, score in similarities[:k]:
            if score > 0:
                results.append({
                    'document': self.documents[idx],
                    'score': score,
                    'index': idx + 1
                })
        
        return results
    
    def generate_answer(self, query, context_docs):
        """📍 Шаги 6-7: Create Prompt + Generate Response"""
        print("\n📍 Шаг 6: Prompt - Формирование промпта с контекстом")
        
        if not context_docs:
            return "Релевантные документы не найдены"
        
        # Формирование контекста
        context = "\\n".join([f"- {doc['document']}" for doc in context_docs])
        prompt = f"Вопрос: {query}\\nКонтекст:\\n{context}\\nОтвет:"
        
        print("   Промпт сформирован с контекстом из документов")
        print("📍 Шаг 7: Response - Генерация ответа LLM")
        
        # Упрощенная логика генерации ответа
        query_lower = query.lower()
        if "rag" in query_lower:
            answer = "RAG (Retrieval-Augmented Generation) объединяет поиск по документам и генерацию ответов, обеспечивая точные результаты на основе релевантного контекста"
        elif "быстр" in query_lower or "алгоритм" in query_lower:
            answer = "Согласно документам, HNSW и FAISS показывают лучшую производительность для быстрого векторного поиска"
        else:
            answer = f"На основе найденных документов: {context_docs[0]['document']}"
        
        return answer

# ========== ДЕМОНСТРАЦИЯ RAG PIPELINE ==========
print("=" * 70)
print("🚀 ДЕМОНСТРАЦИЯ RAG PIPELINE")
print("=" * 70)

# Инициализация RAG системы
rag = SimpleRAG()

# База документов
documents = [
    "RAG (Retrieval-Augmented Generation) объединяет поиск и генерацию для точных ответов",
    "FAISS - библиотека Facebook для быстрого поиска по векторам",
    "HNSW показывает лучшую производительность для приближенного поиска",
    "Эмбеддинги преобразуют текст в числовые векторы для семантического поиска",
    "Recall@k измеряет долю найденных релевантных документов"
]

print("\\n📚 ЭТАП ИНДЕКСАЦИИ")
print("-" * 70)
num_docs = rag.add_documents(documents)
print(f"\\n✅ Индексация завершена: {num_docs} документов в базе")

# Тестовые запросы
queries = [
    "Что такое RAG?",
    "Какой алгоритм поиска самый быстрый?"
]

for query in queries:
    print("\\n" + "=" * 70)
    print(f"🔍 ОБРАБОТКА ЗАПРОСА: '{query}'")
    print("=" * 70)
    
    # Поиск релевантных документов (шаги 3-5)
    results = rag.search(query, k=2)
    
    print(f"\\n📊 Найдено релевантных документов: {len(results)}")
    for result in results:
        print(f"   [{result['index']}] Score: {result['score']:.3f}")
        print(f"       {result['document'][:60]}...")
    
    # Генерация ответа (шаги 6-7)  
    answer = rag.generate_answer(query, results)
    
    print(f"\\n✨ ФИНАЛЬНЫЙ ОТВЕТ:")
    # Форматирование ответа с переносами
    words = answer.split()
    line = ""
    for word in words:
        if len(line) + len(word) + 1 <= 66:
            line = line + " " + word if line else word
        else:
            print(f"   {line}")
            line = word
    if line:
        print(f"   {line}")

print("\\n" + "=" * 70)
print("✅ RAG PIPELINE ВЫПОЛНЕН УСПЕШНО!")
print("=" * 70)
Результат выполнения

Нажмите "Запустить код" для выполнения...
Python RAG Песочница

RAG Pipeline Architecture

Базовый пример RAG

Результат выполнения

Особенности реализации

✅ Преимущества:

📝 Упрощения:
