RAG на ноутбуке: от PDF до Qwen3:8B без LangChain и API

Зачем еще одна RAG-статья?

Потому что 90% гайдов начинаются с "установи LangChain" и заканчиваются на "купите API ключ". А я ненавижу зависимости, которые тащат за собой 200 мегабайт ненужного кода и требуют интернета для работы.

Моя цель - собрать систему, которая:

• Работает полностью оффлайн
• Не требует GPU (хотя с ним быстрее)
• Умеет парсить даже сканы PDF
• Отвечает на русском и английском
• Весит меньше 10ГБ
• Не использует LangChain (это принципиально)

Если вы хотите понять, как RAG работает изнутри, а не просто скопировать готовый код - этот гайд для вас.

Что у нас в стеке и почему именно это

Сначала разберемся, зачем нам каждый компонент. Не просто "берем потому что все берут", а с пониманием альтернатив.

Теперь, когда понимаете зачем каждый инструмент, переходим к установке. Сразу предупрежу - если у вас Windows, некоторые вещи будут работать медленнее. Лучше использовать WSL2 или Linux.

1Подготовка окружения: ставим только нужное

Первая ошибка новичков - установить все подряд. Мы поставим минимальный набор, проверим совместимость версий.

# Создаем виртуальное окружение
python -m venv rag_env
source rag_env/bin/activate  # для Windows: rag_env\Scripts\activate

# Ставим базовые зависимости
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu
pip install transformers>=4.40.0  # важна версия для Qwen3
pip install sentence-transformers>=2.7.0
pip install faiss-cpu>=1.8.0
pip install pymupdf>=1.24.0  # быстрее чем fitz
pip install pillow>=10.0.0
pip install pytesseract>=0.3.10
pip install langchain-community==0.0.10  # только для утилит, не для основного кода

Проверяем установку:

import torch
print(f"PyTorch: {torch.__version__}")
print(f"CUDA доступен: {torch.cuda.is_available()}")  # если есть GPU

2Парсинг PDF: сканы и текст в одном пайплайне

Вот где большинство статей врут. Они показывают парсинг текстовых PDF, но молчат про сканированные. А в реальности 70% документов - сканы.

Создаем файл pdf_parser.py:

import fitz  # PyMuPDF
from PIL import Image
import pytesseract
import io
import os
from typing import List, Tuple

def extract_text_from_pdf(pdf_path: str, use_ocr: bool = True) -> List[Tuple[int, str]]:
    """
    Извлекает текст из PDF.
    Возвращает список кортежей (номер_страницы, текст)
    """
    pages_text = []
    
    try:
        doc = fitz.open(pdf_path)
        
        for page_num in range(len(doc)):
            page = doc.load_page(page_num)
            
            # Пробуем извлечь текст обычным способом
            text = page.get_text("text")
            
            # Если текста мало или нет совсем - используем OCR
            if use_ocr and (not text or len(text.strip()) < 50):
                pix = page.get_pixmap(dpi=200)
                img_data = pix.tobytes("ppm")
                
                # Конвертируем в PIL Image
                img = Image.open(io.BytesIO(img_data))
                
                # OCR с русским и английским языками
                custom_config = r'--oem 3 --psm 6 -l rus+eng'
                text = pytesseract.image_to_string(img, config=custom_config)
            
            pages_text.append((page_num + 1, text.strip()))
        
        doc.close()
        
    except Exception as e:
        print(f"Ошибка при обработке {pdf_path}: {e}")
        return []
    
    return pages_text

# Пример использования
if __name__ == "__main__":
    # Тестируем на вашем PDF
    pages = extract_text_from_pdf("document.pdf")
    for page_num, text in pages:
        print(f"--- Страница {page_num} ---")
        print(text[:500])  # первые 500 символов
        print()

Почему именно такой подход? Потому что не все PDF одинаковы. Некоторые содержат и текст, и сканы на разных страницах. Наш код определяет, нужен ли OCR, автоматически.

3Членкинг: как не резать по живому

Самая частая ошибка - резать текст по фиксированному количеству символов. Так вы разрежете предложения пополам, потеряете контекст.

Вместо этого используем рекурсивное разделение:

from langchain.text_splitter import RecursiveCharacterTextSplitter

def chunk_documents(pages_text: List[Tuple[int, str]], 
                    chunk_size: int = 1000,
                    chunk_overlap: int = 200) -> List[dict]:
    """
    Разбивает текст на чанки с сохранением контекста.
    """
    # Объединяем все страницы
    full_text = "\n\n".join([f"Страница {num}: {text}" for num, text in pages_text])
    
    # Используем умный сплиттер
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", ". ", "! ", "? ", " ", ""]
    )
    
    chunks = splitter.split_text(full_text)
    
    # Создаем метаданные для каждого чанка
    chunk_docs = []
    for i, chunk in enumerate(chunks):
        # Определяем номера страниц из чанка
        page_numbers = []
        for page_num, _ in pages_text:
            if f"Страница {page_num}:" in chunk:
                page_numbers.append(page_num)
        
        chunk_docs.append({
            "id": i,
            "text": chunk,
            "page_numbers": page_numbers,
            "chunk_size": len(chunk)
        })
    
    return chunk_docs

# Альтернатива без langchain (если принципиально):
def simple_smart_chunking(text: str, max_chunk: int = 1000) -> List[str]:
    """
    Простой членкинг по предложениям.
    """
    sentences = text.replace('\n', ' ').split('. ')
    chunks = []
    current_chunk = ""
    
    for sentence in sentences:
        if len(current_chunk) + len(sentence) < max_chunk:
            current_chunk += sentence + ". "
        else:
            if current_chunk:
                chunks.append(current_chunk.strip())
            current_chunk = sentence + ". "
    
    if current_chunk:
        chunks.append(current_chunk.strip())
    
    return chunks

Размер чанка 1000 символов - не случайность. e5-multilingual работает лучше с текстами до 512 токенов (примерно 1000 символов). Перекрытие 200 символов нужно, чтобы контекст не терялся на границах.

4Эмбеддинги: e5-multilingual вместо OpenAI

Почему e5-multilingual-v2, а не что-то другое? Потому что он:

• Бесплатный и оффлайн
• Поддерживает русский идеально
• Размер эмбеддинга 1024 (меньше чем у некоторых, но достаточно)
• Точность сравнима с text-embedding-ada-002

Загружаем модель (в первый раз скачается 1.3ГБ):

from sentence_transformers import SentenceTransformer
import numpy as np
import pickle
import os

class EmbeddingManager:
    def __init__(self, model_name: str = "intfloat/e5-multilingual-v2"):
        """
        Менеджер для работы с эмбеддингами.
        Кэширует модель и эмбеддинги.
        """
        self.model_name = model_name
        self.model = None
        self.embedding_cache = {}
        
    def load_model(self):
        """Загружает модель (занимает 1-2 минуты в первый раз)"""
        if self.model is None:
            print(f"Загрузка модели {self.model_name}...")
            self.model = SentenceTransformer(self.model_name)
            print("Модель загружена")
        
    def get_embeddings(self, texts: List[str], batch_size: int = 32) -> np.ndarray:
        """
        Создает эмбеддинги для списка текстов.
        Важно: добавляем префикс "query: " или "passage: " в зависимости от задачи.
        """
        self.load_model()
        
        # Для документов используем префикс "passage: "
        prefixed_texts = ["passage: " + text for text in texts]
        
        embeddings = self.model.encode(
            prefixed_texts,
            batch_size=batch_size,
            show_progress_bar=True,
            normalize_embeddings=True  # важно для косинусного сходства
        )
        
        return embeddings
    
    def save_embeddings(self, embeddings: np.ndarray, filepath: str):
        """Сохраняет эмбеддинги в файл"""
        with open(filepath, 'wb') as f:
            pickle.dump(embeddings, f)
        print(f"Эмбеддинги сохранены в {filepath}")
    
    def load_embeddings(self, filepath: str) -> np.ndarray:
        """Загружает эмбеддинги из файла"""
        with open(filepath, 'rb') as f:
            embeddings = pickle.load(f)
        print(f"Загружено {len(embeddings)} эмбеддингов")
        return embeddings

5FAISS индекс: поиск за миллисекунды

FAISS от Facebook - это не просто библиотека, это произведение инженерного искусства. Она использует SIMD инструкции процессора для поиска в миллионах векторов за миллисекунды.

import faiss
import numpy as np

class FAISSIndex:
    def __init__(self, dimension: int = 1024):
        """
        Создает FAISS индекс для векторного поиска.
        dimension: размерность эмбеддингов (1024 для e5-multilingual-v2)
        """
        self.dimension = dimension
        self.index = None
        self.chunks = []  # храним оригинальные тексты
        self.metadata = []  # храним метаданные
        
    def build_index(self, embeddings: np.ndarray, chunks: List[dict]):
        """
        Строит индекс из эмбеддингов.
        """
        if len(embeddings) == 0:
            raise ValueError("Нет эмбеддингов для индексации")
            
        # Преобразуем в float32 (требование FAISS)
        embeddings = embeddings.astype('float32')
        
        # Создаем индекс с Inner Product (эквивалентно косинусному сходству при нормализованных векторах)
        self.index = faiss.IndexFlatIP(self.dimension)
        
        # Добавляем векторы в индекс
        self.index.add(embeddings)
        
        # Сохраняем оригинальные данные
        self.chunks = [chunk["text"] for chunk in chunks]
        self.metadata = chunks
        
        print(f"Индекс построен. Векторов: {self.index.ntotal}")
    
    def search(self, query: str, embedding_manager, k: int = 5) -> List[dict]:
        """
        Ищет k наиболее релевантных чанков.
        """
        if self.index is None:
            raise ValueError("Индекс не построен")
        
        # Создаем эмбеддинг для запроса (с префиксом "query: ")
        query_embedding = embedding_manager.model.encode(
            ["query: " + query],
            normalize_embeddings=True
        ).astype('float32')
        
        # Ищем в индексе
        distances, indices = self.index.search(query_embedding, k)
        
        # Формируем результаты
        results = []
        for i, (distance, idx) in enumerate(zip(distances[0], indices[0])):
            if idx != -1:  # -1 означает отсутствие результата
                results.append({
                    "text": self.chunks[idx],
                    "metadata": self.metadata[idx],
                    "score": float(distance),
                    "rank": i + 1
                })
        
        return results
    
    def save_index(self, filepath: str):
        """Сохраняет индекс на диск"""
        if self.index is None:
            raise ValueError("Индекс не построен")
            
        faiss.write_index(self.index, filepath)
        
        # Сохраняем метаданные отдельно
        metadata_file = filepath.replace(".faiss", "_meta.pkl")
        with open(metadata_file, 'wb') as f:
            pickle.dump({"chunks": self.chunks, "metadata": self.metadata}, f)
        
        print(f"Индекс сохранен в {filepath}")
    
    def load_index(self, filepath: str):
        """Загружает индекс с диска"""
        self.index = faiss.read_index(filepath)
        
        # Загружаем метаданные
        metadata_file = filepath.replace(".faiss", "_meta.pkl")
        with open(metadata_file, 'rb') as f:
            data = pickle.load(f)
            self.chunks = data["chunks"]
            self.metadata = data["metadata"]
        
        print(f"Индекс загружен. Векторов: {self.index.ntotal}")

Почему IndexFlatIP, а не IndexFlatL2? Потому что наши эмбеддинги нормализованы (normalize_embeddings=True), и скалярное произведение равно косинусному сходству. Это быстрее.

6Qwen3:8B - локальная LLM, которая понимает русский

Qwen3 от Alibaba - темная лошадка среди открытых моделей. При 8Б параметрах она показывает результаты как 13Б модели, а с русским справляется лучше большинства конкурентов.

from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
import torch

class QwenChat:
    def __init__(self, model_name: str = "Qwen/Qwen3-8B-Instruct"):
        """
        Загрузка и настройка Qwen3-8B.
        Модель займет ~5ГБ в 4-битной квантовке.
        """
        self.model_name = model_name
        self.tokenizer = None
        self.model = None
        self.pipeline = None
        
    def load_model(self, quantize: bool = True):
        """
        Загружает модель.
        quantize=True - использует 4-битную квантовку для экономии памяти.
        """
        print(f"Загрузка модели {self.model_name}...")
        
        # Загружаем токенизатор
        self.tokenizer = AutoTokenizer.from_pretrained(
            self.model_name,
            trust_remote_code=True
        )
        
        # Настройки загрузки модели
        load_kwargs = {
            "torch_dtype": torch.float16,
            "device_map": "auto",
            "trust_remote_code": True
        }
        
        if quantize:
            # 4-битная квантовка для экономии памяти
            load_kwargs.update({
                "load_in_4bit": True,
                "bnb_4bit_compute_dtype": torch.float16,
                "bnb_4bit_quant_type": "nf4",
                "bnb_4bit_use_double_quant": True
            })
        
        # Загружаем модель
        self.model = AutoModelForCausalLM.from_pretrained(
            self.model_name,
            **load_kwargs
        )
        
        # Создаем пайплайн для генерации
        self.pipeline = pipeline(
            "text-generation",
            model=self.model,
            tokenizer=self.tokenizer,
            device_map="auto"
        )
        
        print("Модель загружена")
    
    def generate_response(self, context: str, question: str, 
                         max_length: int = 1024,
                         temperature: float = 0.7) -> str:
        """
        Генерирует ответ на основе контекста и вопроса.
        """
        if self.pipeline is None:
            self.load_model()
        
        # Формируем промпт в формате Qwen
        prompt = f"""<|im_start|>system
Ты - помощник, который отвечает на вопросы на основе предоставленного контекста.
Отвечай только на основе контекста. Если в контексте нет информации для ответа, скажи "Я не нашел информацию об этом в предоставленных документах".

Контекст:
{context}
<|im_end|>
<|im_start|>user
{question}
<|im_end|>
<|im_start|>assistant
"""
        
        # Генерируем ответ
        result = self.pipeline(
            prompt,
            max_new_tokens=max_length,
            temperature=temperature,
            do_sample=True,
            top_p=0.9,
            repetition_penalty=1.1,
            eos_token_id=self.tokenizer.eos_token_id,
            pad_token_id=self.tokenizer.pad_token_id
        )
        
        # Извлекаем только ответ ассистента
        generated_text = result[0]['generated_text']
        response = generated_text.split("<|im_start|>assistant")[-1].strip()
        
        # Убираем возможные теги конца
        response = response.replace("<|im_end|>", "").strip()
        
        return response

7Собираем все вместе: полный пайплайн

Теперь соединим все компоненты в единую систему. Создадим класс, который управляет всем процессом.

import time
from pathlib import Path

class LocalRAGSystem:
    def __init__(self, pdf_folder: str = "documents"):
        """
        Полная RAG система для работы с PDF.
        """
        self.pdf_folder = Path(pdf_folder)
        self.embedding_manager = EmbeddingManager()
        self.faiss_index = FAISSIndex()
        self.llm = None  # Загрузим по требованию
        self.chunks = []
        
    def process_documents(self, rebuild: bool = False):
        """
        Обрабатывает все PDF в папке и строит индекс.
        """
        index_file = self.pdf_folder / "faiss_index.faiss"
        embeddings_file = self.pdf_folder / "embeddings.pkl"
        chunks_file = self.pdf_folder / "chunks.json"
        
        # Если индекс уже есть и rebuild=False - загружаем
        if not rebuild and index_file.exists():
            print("Загрузка существующего индекса...")
            self.faiss_index.load_index(str(index_file))
            return
        
        # Иначе обрабатываем документы заново
        print("Обработка документов...")
        
        all_pages = []
        pdf_files = list(self.pdf_folder.glob("*.pdf"))
        
        if not pdf_files:
            print(f"В папке {self.pdf_folder} не найдено PDF файлов")
            return
        
        for pdf_file in pdf_files:
            print(f"Обработка {pdf_file.name}...")
            pages = extract_text_from_pdf(str(pdf_file))
            all_pages.extend(pages)
            print(f"  Извлечено {len(pages)} страниц")
        
        # Разбиваем на чанки
        print("Разбиение на чанки...")
        self.chunks = chunk_documents(all_pages)
        print(f"Создано {len(self.chunks)} чанков")
        
        # Создаем эмбеддинги
        print("Создание эмбеддингов...")
        texts = [chunk["text"] for chunk in self.chunks]
        embeddings = self.embedding_manager.get_embeddings(texts)
        
        # Строим индекс
        print("Построение FAISS индекса...")
        self.faiss_index.build_index(embeddings, self.chunks)
        
        # Сохраняем все
        self.faiss_index.save_index(str(index_file))
        self.embedding_manager.save_embeddings(embeddings, str(embeddings_file))
        
        print("Обработка завершена!")
    
    def ask_question(self, question: str, k: int = 5) -> str:
        """
        Задает вопрос системе.
        """
        # Ищем релевантные чанки
        start_time = time.time()
        results = self.faiss_index.search(question, self.embedding_manager, k=k)
        search_time = time.time() - start_time
        
        if not results:
            return "Не найдено релевантной информации в документах."
        
        # Объединяем контекст
        context = "\n\n".join([f"[Документ {i+1}] {r['text']}" for i, r in enumerate(results)])
        
        # Загружаем LLM если еще не загружена
        if self.llm is None:
            print("Загрузка LLM... (это займет время в первый раз)")
            self.llm = QwenChat()
            self.llm.load_model()
        
        # Генерируем ответ
        print(f"Поиск занял {search_time:.2f} сек. Генерация ответа...")
        response = self.llm.generate_response(context, question)
        
        # Добавляем источники
        sources = "\n\nИсточники:\n" + "\n".join(
            [f"- Страница {r['metadata']['page_numbers']}: {r['text'][:100]}..." 
             for r in results]
        )
        
        return response + sources

Запускаем систему: от PDF до ответа

Теперь все готово. Создаем папку documents, кидаем туда PDF файлы и запускаем:

# Создаем систему
rag = LocalRAGSystem("documents")

# Обрабатываем документы (в первый раз займет время)
rag.process_documents()

# Задаем вопросы
questions = [
    "Какие основные требования к проекту?",
    "Сколько стоит реализация?",
    "Какие сроки выполнения работ?"
]

for question in questions:
    print(f"\nВопрос: {question}")
    answer = rag.ask_question(question)
    print(f"Ответ: {answer}")
    print("-" * 80)

Оптимизации и подводные камни

После того как система работает, нужно ее оптимизировать. Вот что я узнал на практике:

1. Память - главный враг

Qwen3-8B в 4-битной квантовке занимает ~5ГБ. e5-multilingual - еще 1.3ГБ. FAISS индекс для 1000 документов - ~100МБ. Итого минимум 6.5ГБ RAM.

Что делать если не хватает:

• Используйте Qwen3-4B вместо 8B
• Уменьшите batch_size в get_embeddings до 8 или 16
• Используйте память SSD как swap (но это медленно)

2. Скорость первого запуска

Первая загрузка моделей занимает 5-10 минут. После этого они кэшируются в ~/.cache/huggingface.

Ускорить можно:

# Предзагружаем модели
python -c "from sentence_transformers import SentenceTransformer; SentenceTransformer('intfloat/e5-multilingual-v2')"
python -c "from transformers import AutoTokenizer; AutoTokenizer.from_pretrained('Qwen/Qwen3-8B-Instruct')"

3. Качество OCR

Tesseract плохо работает с:

• Таблицами (преобразуйте в изображение по ячейкам)
• Рукописным текстом (нужен специальный OCR)
• Наклонным текстом (исправляйте skew перед OCR)

Для таблиц лучше использовать Camelot или Tabula. Но они требуют Java.

4. Размер чанков

1000 символов - компромисс. Для юридических документов лучше 1500-2000. Для чат-логов - 500-700.

Экспериментируйте:

# Тестируем разные размеры
for chunk_size in [500, 1000, 1500, 2000]:
    chunks = chunk_documents(pages, chunk_size=chunk_size)
    print(f"Чанков при {chunk_size}: {len(chunks)}")

Что делать когда все работает

Система готова. Что дальше?

Добавьте интерфейс: Gradio или Streamlit на 50 строчек кода. Или REST API с FastAPI.

Кэшируйте ответы: Похожие вопросы не должны генерироваться заново. Используйте Redis или даже sqlite.

Мониторинг качества: Записывайте вопросы и ответы. Раз в неделю проверяйте, не галлюцинирует ли модель.

Обновление документов: Добавьте функцию incremental update индекса.

Чего не хватает в этом гайде

Я сознательно упростил некоторые моменты:

1. Обработка исключений - в продакшене добавьте retry логику для HuggingFace загрузок

2. Логирование - используйте structlog или loguru вместо print

3. Векторизация в фоне - для больших документов делайте это асинхронно

4. Оценка качества - добавьте RAGAS или собственные метрики

Но для старта этого достаточно. Система работает. Отвечает на вопросы. Не требует интернета.

Вместо заключения: что изменится через год

На 13.02.2026 эта система актуальна. Но через год появятся:

• Модели размером 2Б с качеством как у 8Б сегодня
• Эмбеддинги с размером 256 вместо 1024
• Аппаратное ускорение поиска на NPU
• Нативные OCR в браузере

Суть останется той же: взять документы, разбить на смысловые куски, найти релевантные, передать модели. Просто каждый этап станет быстрее и дешевле.

Главное - понять принципы. А код всегда можно обновить.