Базовый RAG-компонент для локального семантического поиска на Питоне

quad_rag_core — лёгкое Python-ядро для локального RAG, которое автоматически отслеживает изменения в папках, индексирует их в Qdrant и поддерживает эмбеддинги в актуальном состоянии. Изначально проект задумывался как инструмент MCP (Model Context Protocol), но стал универсальной основой для системы локального семантического поиска.

Зачем это нужно

В процессе работы с кодовой базой через LLM-агентов и при необходимости локального семантического поиска по файлам проекта обнаружилась проблема. Инструменты агентской разработки вроде Kilo Code предоставляют встроенную функцию семантического поиска, но в компании заявляют что в будущем эта функциональность может стать платной. Сразу задумался о том чтобы сделать свою подсистему поиска. Простые запросы к MCP-серверу на поиск и обновление тут не подойдут - система поиска должна иметь полный контроль над контекстом - она должна автоматически узнавать, что файл удалён, функция изменена или добавлен новый документ, без необходимости перезапуска индексации.

От идеи к архитектуре

В начале планировался простой MCP-сервер, который принимает команды поиска и обновления, индексирует текстовые файлы и PDF, использует Qdrant как векторное хранилище и эмбеддит локально.

В ходе проектирования стало понятно: вся логика отслеживания файлов, парсинга, чанкинга и синхронизации с Qdrant — это переиспользуемое ядро, а не часть MCP-протокола.

Так появился quad_rag_core — отдельный Python-модуль, который не знает ничего про MCP или другие внешние интерфейсы, но готов к ним подключаться.

Архитектура: компонент RAG как watch-сервис

Самая важная особенность quad_rag_core — автоматический жизненный цикл индекса.

Вы говорите системе следить за папкой. Она создаёт коллекцию в Qdrant, сканирует файлы и индексирует их. Затем запускается watchdog, который отлавливает события файловой системы: создание, изменение, перемещение, удаление.

При любом изменении старые чанки удаляются, новые пересчитываются и вставляются. Даже при перезапуске система восстанавливает состояние из метаданных в Qdrant.

Это значит, что система всегда работает с актуальной базой знаний без ручного управления.

Компоненты ядра

QdrantManager обёртка над qdrant-client с упрощённым API
LocalEmbedder синглтон-эмбеддер на SentenceTransformer
RAGFileWatcher реагирует на события файловой системы, разбивает текст на чанки, обновляет точки в Qdrant
PathWatcherManager оркестратор, который управляет несколькими папками, сериализует данные, восстанавливает отслеживание при старте
Файловый процессор обработчик текстовых файлов и PDF с тремя бэкендами.

в составе модулей:

quad_rag_core/ ├── qdrant_manager.py # работа с Qdrant ├── embedder.py # nomic-embed-text ├── reranker.py # BGE reranker ├── file_processor.py # чанкинг + PDF ├── path_watcher.py # watchdog ├── path_manager.py # оркестрация ├── config.py # настройки по умолчанию └── utils.py # хеши, MIME, нормализация

Singleton Pattern для AI-моделей

Одно из ключевых архитектурных решений — использование паттерна Singleton для LocalEmbedder.

class LocalEmbedder: _instance = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) cls._instance.model = SentenceTransformer( "nomic-ai/nomic-embed-text-v2-moe", trust_remote_code=True, device="cuda" if torch.cuda.is_available() else "cpu" ) return cls._instance

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

• Эффективное использование GPU памяти — модель загружается только один раз

• Быстродействие — последующие вызовы не требуют загрузки модели

• Thread-safety — все потоки используют один и тот же экземпляр модели

Тот же подход применён к LocalReranker с моделью BAAI/bge-reranker-v2-m3.

Dual-Prompt Embedding

Модель nomic-embed-text-v2-moe обучена с разными инструкциями для разных типов текста. При использовании prompt_name="passage" модель понимает, что это фрагмент документа для индексации. При prompt_name="query" модель создаёт эмбеддинг, оптимизированный для поиска по индексированным чанкам.

def embed_document(self, text: str) -> List[float]: """Embed a document (for indexing).""" vector = self.model.encode( text, prompt_name="passage", # Для документов convert_to_tensor=False, show_progress_bar=False, normalize_embeddings=True ) return vector.tolist() def embed_query(self, text: str) -> List[float]: """Embed a search query (for inference).""" vector = self.model.encode( text, prompt_name="query", # Для запросов convert_to_tensor=False, show_progress_bar=False, normalize_embeddings=True ) return vector.tolist()

State Persistence в Qdrant

Конфигурация watcher-компонента, следящего за папкой, хранится как специальная точка метаданных в каждой коллекции Qdrant с фиксированным UUID.

WATCHER_METADATA_ID = str(uuid.UUID("f0f0f0f0f0-0000-0000-0000-000000000001")) # При создании ватчера meta_point = PointStruct( id=WATCHER_METADATA_ID, vector=[0.0] * 768, payload={ "watcher_config": { "folder_path": folder_path, "content_types": content_types or ["text"], "collection_prefix": self.collection_prefix } } )

Что это дает:

1. Автоматическое восстановление при перезапуске сервиса PathWatcherManager считывает все коллекции и восстанавливает watcher'ы из метаданных

2. Позволяет хранить метаданные централизованно в единой базе данных, значит не нужен отдельный конфиг-файл или другая база.

3. Простота миграции — при переносе на другой сервер достаточно скопировать Qdrant данные.

def _restore_from_qdrant(self): """Восстанавливает наблюдателей из метаданных в Qdrant.""" collections = self.qdrant_manager.client.get_collections().collections for col in collections: meta = self.qdrant_manager.client.retrieve( collection_name=col.name, ids=[WATCHER_METADATA_ID], with_payload=True, with_vectors=False ) config = meta[0].payload.get("watcher_config") watcher = RAGFileWatcher( folder_path=config["folder_path"], collection_name=col.name, ... )

Многокомпонентная обработка pdf-файлов

Для извлечения текста из PDF используется стратегия с тремя бэкендами и fallback-механизмом.

def _extract_text_from_pdf(file_path: str) -> str: # Попытка 1: PyPDF2 try: text = _extract_pdf_PyPDF2(file_path) if text.strip(): return text except Exception as e: print(f"[DEBUG] PyPDF2 failed for {file_path}: {e}") # Попытка 2: fitz try: text = _extract_pdf_fitz(file_path) if text.strip(): return text except Exception as e: print(f"[DEBUG] fitz failed for {file_path}: {e}") # Попытка 3: pdfplumber try: text = _extract_pdf_pdfplumber(file_path) if text.strip(): return text except Exception as e: print(f"[DEBUG] pdfplumber failed for {file_path}: {e}") return ""

При этом обеспечивается надежность - если один бэкенд падает, система продолжает работать со следующим.

Обход подводных камней: потоко-безопасность и конфликты с наблюдением папок

Для работы с ватчерами и отслеживания прогресса используем блокировки потоков.

class PathWatcherManager: def __init__(self, ...): self.watchers: Dict[str, RAGFileWatcher] = {} self._lock = threading.Lock() class RAGFileWatcher: def __init__(self, ...): self._progress_lock = threading.Lock()

Значит несколько потоков могут одновременно вызывать методы, а блокировки предотвращают ситуацию гонок и система корректно работает при параллельной обработке файлов.

При попытке добавить папку для отслеживания система проверяет конфликты с уже наблюдаемыми путями.

def _check_conflict(self, new_path: str) -> List[str]: """Returns list of conflicting paths.""" new_path = self._normalize_path(new_path) conflicts = [] for watched in self.watchers: if (new_path.startswith(watched + os.sep) or watched.startswith(new_path + os.sep)): conflicts.append(watched) return conflicts

Какие проблемы это решает:
- Предотвращение дублирования — нельзя отслеживать вложенные папки /project, и /project/src одновременно, то есть мы избегаем ситуаций, когда один файл обрабатывается несколькими ватчерами.
- Пользователь получает ясные сообщения об ошибках и сразу понимает, что пошло не так.

Отложенная обработка

При изменении файла система не обрабатывает его мгновенно, а ждёт заданный интервал времени, по умолчанию 0.5 секунды. Интервал можно настроить под свой стиль и не обрабатывать файл, пока он активно изменяется.

def _delayed_process(self, file_path: str, delay: float = 0.5): """Waits until file stops changing, then processes it.""" time.sleep(delay) self._process_file(file_path)

Конфигурация

Всё, что касается чанкинга, фильтрации, порогов поиска, процент перекрытия чанков вынесено в единый файл конфигурации config.py

CHUNK_SIZE_WORDS = 150 CHUNK_OVERLAP_RATIO = 0.15 SEARCH_SCORE_THRESHOLD = 0.150 CHUNK_CHARACTERS_PREVIEW = 100 RERANK_SCORE_THRESHOLD = 0.35 TEXT_FILE_EXTENSIONS = { '.py', '.js', '.md', '.json', '.txt', '.yaml', '.yml', ... }

Реранкер

Первый этап поиска — векторный. Но он не всегда точен, особенно когда запрос семантически далёк от формулировки в коде или документации. Поэтому добавлен локальный кросс-энкодер реранкер (компонентLocalReranker).

class LocalReranker: _instance = None def __new__(cls): if cls._instance is None: cls._instance = super().__new__(cls) cls._instance.model = CrossEncoder( "BAAI/bge-reranker-v2-m3", max_length=512, device="cuda" if torch.cuda.is_available() else "cpu" ) return cls._instance def rerank(self, query: str, chunks: List[str], top_k: int = 10): """Возвращает список (чанк, score), отсортированный по релевантности.""" pairs = [[query, chunk] for chunk in chunks] scores = self.model.predict(pairs, batch_size=32, show_progress_bar=False) results = list(zip(chunks, scores)) results.sort(key=lambda x: x[1], reverse=True) return results[:top_k]

Реранкер принимает топ-N чанков от Qdrant, оценивает их парную релевантность к запросу и возвращает отсортированный список с новыми скорами. Это повышает точность первых результатов, особенно для сложных запросов.

Реранкер реализован как синглтон, чтобы не загружать GPU память при частых вызовах.

Интеграция

Ядро не зависит, например, от MCP-сервера реализующего внешний интерфейс семантического поиска, но сравнительно легко с ним интегрируется. Конечно, схема обработки данных внешнего сервиса может быть разной. Предварительно устанавливаем quad_rag_core как компонент Питона. Тогда, схематично, MCP-инструменты LLM могут выглядеть так:

from quad_rag_core.path_manager import PathWatcherManager from quad_rag_core.qdrant_manager import QdrantManager from quad_rag_core.embedder import LocalEmbedder pm = PathWatcherManager( QdrantManager(host="localhost", port=6333), LocalEmbedder() ) @mcp.tool() def watch_folder(path: str): pm.watch_folder(path) @mcp.tool() def search(query: str, folder: str): vector = LocalEmbedder().embed_query(query) results = QdrantManager().search(f"rag_{folder}", vector) reranked = LocalReranker().rerank(query, [r.payload for r in results]) return reranked

То есть нашему внешнему сервису достаточно всего двух функций
- отслеживать локальную папку в новом или уже существующем индексе (watch_folder)
- и, собственно, семантический поиск в этой папке (search).

Такой компонентный подход позволяет использовать одно и то же ядро для MCP-сервера, FastAPI-веб-интерфейса, CLI-утилиты или любого другого интерфейса.

Производительность

Система автоматически использует GPU, если доступен CUDA.

device="cuda" if torch.cuda.is_available() else "cpu"

Реранкер поддерживает пакетную обработку для ускорения.

scores = self.model.predict(pairs, batch_size=32, show_progress_bar=False)

Оптимизации индексации включают чанкинг с перекрытием, удаление старых чанков перед вставкой новых и отложенную обработку файлов.

Выводы и результаты

Создан проект (исходники на Gitgub) локального, автономного ядра RAG, которое работает в фоне и автоматически поддерживает индекс векторной БД в актуальном состоянии, например, для систем поиска, которые понимают кодовую базу здесь и сейчас.

Основные преимущества — автоматическое отслеживание изменений в папке, локальная работа без облачных API, поддержка множества форматов файлов и простая интеграция с любыми протоколами. Система сохраняет состояние в Qdrant для автоматического поддержания индекса и применяет multi-backend подход для максимальной совместимости с PDF.

Ядро подходит для MCP-серверов LLM-агентов, веб-интерфейсов документации, CLI-утилит для разработчиков и чат-ботов. Архитектура позволяет расширять функциональность и адаптировать под конкретные задачи.

Особенности проекта включают централизованную конфигурацию, кросс-платформенную работу и защиту от конфликтов путей отслеживания. В дальнейшем можно добавить поддержку дополнительных форматов файлов, внешнюю конфигурацию и метрики для мониторинга.

Quad_rag_core это open-source ядро, которое можно интегрировать с разными внешними интерфейсами RAG, расширить под свои форматы и запустить на локальном компьютере. Протестировал модуль на двух обертках - MCP сервере для LLM и локальном поиске с веб-интерфейсом. В обоих случаях система продемонстрировала высокую скорость реакции - изменения в исходных файлах обновлялись в индексе практически моментально и поиск показывал актуальные результаты. Все три подсистемы разрабатывались с помощью LLM.

Вообще, хочется заметить, что с развитием ИИ роль человека программиста, похоже, трансформируется в подобие менеджера команды AI-разработчиков цифрового контента. "Контент" здесь - в широком смысле, включая и разработку компьютерных программ. Причем, уровень подготовки и квалификация такого менеджера в программистских дисциплинах должны быть очень высокими, ведь его роль будет заключаться в налаживании полного цикла разработки ПО на виртуальной фабрике, сотрудниками которой будут роботы. Почему? Потому что скорость разработки у ИИ-программиста на порядок выше чем у человека и угнаться за ним будет не реально. Искусство программирования перейдет в искусство управления такой "фабрикой". Появятся грейды и уровни сертификации таких менеджеров, а термин CAD (Computer Aided Design) трансформируется в AIAD - Artificial Intelligence Aided Design - Разработка на основе ИИ.