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 с упрощённым APILocalEmbedder синглтон-эмбеддер на SentenceTransformerRAGFileWatcher реагирует на события файловой системы, разбивает текст на чанки, обновляет точки в QdrantPathWatcherManager оркестратор, который управляет несколькими папками, сериализует данные, восстанавливает отслеживание при стартеФайловый процессор обработчик текстовых файлов и 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()Сохранение конфигурации поиска в 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 - Разработка на основе ИИ.
#Сезон ИИ в разработке
