PM: Нам нужно актуализировать базу знаний для ИИ-ассистента,
там изменилась инструкция по смене пароля.
DevOps: Не проблема, сейчас запущу скрипт, через два часа всё обновится.
Предупреди Заказчика о недоступности сервиса.

Знакомая ситуация? Полная зачистка векторной базы и реиндексация всех имеющихся документов с остановкой сервиса - решение простое и надёжное, но «прощается» только на этапе прототипа.

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

Нам потребуется Qdrant. На мой взгляд это отличный движок для векторного поиска с открытым исходным кодом, удобным API и интеграцией. Установить его можно через docker compose или использовать бинарник. У меня версия 1.14.1, не самая новая, но привычка. Если захотите более свежую, не забудьте про версию qdrant-клиента в окружении.

В репозитории туториала есть ноутбук с кодом и тестами, requirements.txt для установки окружения и короткая инструкция в README.

Чтобы упростить код я отказался от сегментации документов, в туториале один документ равен одному чанку. Как добавить сегментацию в код расскажу в конце статьи, в разделе «Замечания по масштабированию».

Для экспериментов я обычно использую тексты в жанре киберпанк. На этот раз тестовые документы будут содержать фрагменты синтетического законопроекта Директива 401. «О лицензировании корпоративного биологического присутствия и коммерческой эксплуатации белковых носителей». Заранее прошу не удивляться формулировкам.

Архитектура и логика

Система инкрементального обновления это ETL-скрипт на python. Qdrant используется через экземпляр класса QdrantVectorStore из LangChain. Ещё есть рекурсивная функция и две таблицы на pandas. И конечно хеширование. В качестве хеша - md5 дайджест полного пути к файлу с именем. Такой вариант позволяет не загружать сам документ в процессе обхода файловой системы.

Логика работы основана на статусах документа: new, update, delete. Статусы хранятся в мастер-таблице. После успешной загрузки документа в хранилище статус меняется на load, а при удалении документа запись о нём удаляется из мастер-таблицы.

Начнём с векторного хранилища. Инициализируем клиент. Векторы в Qdrant хранятся в именованных коллекциях. Клиент нужен чтобы создать коллекцию или удалить. Далее определяем эмбеддинг и создаём экземпляр векторного хранилища.

ETL-процесс включает четыре шага.

Первый шаг - рекурсивный обход директорий с документами. На входе - путь к корневой папке, на выходе таблица с перечнем файлов и их атрибутами: полный путь к файлу, имя файла, хеш-имя и дата модификации, а также поле для хранения статуса, пока пустое. Можно отказаться от рекурсии и сделать обход на базе функции os.walk. Вариант есть в ноутбуке, он закомментирован.

Код для рекурсивного обхода директорий

lst = list()
def list_dir(root: str, lst: List[Any]) -> None:
    """ Recursive traversing over directories """
    for node in os.listdir(root):
        if os.path.isdir(os.path.join(root, node)):
            node = os.path.join(root, node)
            list_dir(node, lst)  # Recursion
        else:
            path = os.path.join(root, node)
            path = path.replace('\\', '/')
            full_path = path.replace(f'{ROOT_DOC_PATH}/', '')
            files_dict = {
                'full_path': full_path,
                'file_name': node,
                'hash_name': md5(full_path.encode('utf-8')).hexdigest(),
                'modified': datetime.fromtimestamp(os.path.getmtime(path)),
                'status': ' ',
            }
            lst.append(files_dict)
    return None

  def scan_dir(directory: str) -> bool:
    """ Recursive scan over directories (file folder) """
    print('Files scanning...')
    lst = list()
    # os.walk implementation is here
    list_dir(directory, lst)  # comment this if os.walk used
    if lst:
        files_data = pd.DataFrame(lst)
        files_data.to_csv(os.path.join(STORAGE_PATH,  'files_data.csv'), sep=';', index=False)
        print(f'Recursive traversing Complete, total files: {files_data.shape[0]}')       
        return True
    else:
        print(f'Recursive traversing Complete. No documents found')       
        return False

Второй шаг - расчёт инкремента обновления или так называемой дельты. На входе таблица с результатами сканирования и мастер таблица. Если это первый запуск, то мастер таблицы ещё нет, и она создаётся копированием таблицы с результатами сканирования. На выходе — мастер-таблица с проставленными статусами (new, update, delete). Статусы определяются по следующей схеме:

1. Строим два списка с хеш-именами, new_hashes на основе сканирования, old_hashes на базе мастер-таблицы:

new_hashes = list(files_data.loc[:, 'hash_name'])
old_hashes = list(master_data.loc[:, 'hash_name'])

2. Помечаем в мастер-таблице статусом delete все документы, хеш-имена которых отсутствуют в списке new_hashes

master_data.loc[~master_data.hash_name.isin(new_hashes), 'status'] = 'delete'

3. Обходим мастер-таблицу в цикле по строкам, пропуская документы со статусом delete, сравниваем даты модификаций для документов с одинаковым хеш-именем из таблицы файлов и мастер-таблицы. Если дата в таблице файлов больше даты в мастер-таблице, устанавливаем статус update и записываем новую дату в мастер-таблицу.

4. Делаем выборку из таблицы файлов для документов, хеш-имена которых отсутствуют в мастер-таблице, устанавливаем для них статус new и добавляем этот срез в мастер-таблицу.

new_data = files_data.loc[~files_data.hash_name.isin(old_hashes), :].copy()
new_data.status = 'new'
master_data = pd.concat([master_data, new_data], axis=0, ignore_index=True)

Весь код расчёта дельты

def create_delta() -> bool:
    """ Updating delta calculation"""
    print('\nIncrement calculation...')
    
    try:  # Attempt to load the scan data
        files_data = pd.read_csv(os.path.join(STORAGE_PATH, 'files_data.csv'), sep=';')
    except FileNotFoundError:
        print(f'Files data not found, exit')
        return False
    
    try:  # Attempt to load master data
        master_data = pd.read_csv(os.path.join(STORAGE_PATH, 'master_data.csv'), sep=';')
    except FileNotFoundError:
        print(f'Master data not found, create new')
        master_data = files_data.copy()  # Copy master data from scan data
        master_data.status = 'new'
        master_data.to_csv(f'{STORAGE_PATH}/master_data.csv', sep=';', index=False)
    
    new_hashes = list(files_data.loc[:, 'hash_name'])
    old_hashes = list(master_data.loc[:, 'hash_name'])

    # Check for deleted files
    master_data.loc[~master_data.hash_name.isin(new_hashes), 'status'] = 'delete'
        
    # Check for updated files
    for i in master_data.index:
        if master_data.loc[i, 'status'] == 'delete':
            continue
    
        master_mod_date = datetime.fromisoformat(master_data.loc[i, 'modified'])
        status = master_data.loc[i, 'status']
        hash_name = master_data.loc[i, 'hash_name']
        file_mod_date_str = files_data.loc[files_data.hash_name == hash_name, 'modified'].values[0]
        file_mod_date = datetime.fromisoformat(file_mod_date_str)
    
        if file_mod_date > master_mod_date:
            master_data.loc[i, 'status'] = 'update'
            master_data.loc[i, 'modified'] = file_mod_date_str

    # Check for new files
    new_data = files_data.loc[~files_data.hash_name.isin(old_hashes), :].copy()
    new_data.status = 'new'
    master_data = pd.concat([master_data, new_data], axis=0, ignore_index=True)

    master_data.to_csv(f'{STORAGE_PATH}/master_data.csv', sep=';', index=False)
    print('Current master data saved')

    print(f"Deleted files: {master_data[master_data.status == 'delete'].shape[0]}")
    print(f"Updated files: {master_data[master_data.status == 'update'].shape[0]}")
    print(f"New files: {master_data[master_data.status == 'new'].shape[0]}")
    return True

Третий шаг - загрузка новых и обновляемых документов. На входе - мастер-таблица. Если в ней нет записей со статусами new, update, delete, процесс останавливается.

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

Код загрузки новых и обновленных документов

def download_docs() -> bool:
    """ Create list of new and updated documents """
    print('\nLoad docs...')
    try:
        master_data = pd.read_csv(os.path.join(STORAGE_PATH, 'master_data.csv'), sep=';')
    except FileNotFoundError:
        print(f'Master data not found, exit')
        return False

    # Check delta
    delta_size = master_data[master_data.status.isin(['new', 'update', 'delete'])].shape[0]
    if not delta_size:
        print(f'There is nothing to update, exit')
        return False
    
    docs = []
    for i in master_data.index:
        status = master_data.loc[i, 'status']
        full_path = master_data.loc[i, 'full_path']
        if status in ['new', 'update']:
            
            # Loading document(s) text
            with open(os.path.join(ROOT_DOC_PATH, full_path), 'r', encoding='utf-8') as handle:
                doc_text = handle.read()
                doc_title = ''
                if re.search(r'# (.*)\n', doc_text):
                    doc_title = re.search(r'# (.*)\n', doc_text).group(1)
            
            # Chunking should be here
            # splits = ...
            doc = Document(
                page_content=doc_text,
                metadata={
                    'title': doc_title,
                    'hash_name': master_data.loc[i, 'hash_name'],
                    'mod_date': master_data.loc[i, 'modified'],
                    'size': len(doc_text),
                    'status': master_data.loc[i, 'status'],
                }
            )
            docs.append(doc)  # Comment this if splits are used
            # docs.extend(splits)  # if chunking is used
    with open(os.path.join(STORAGE_PATH, 'documens.pkl'), 'wb') as handle:  # Save in binary
        pickle.dump(docs, handle)
    print(f"New and updated docs: {len(docs)}")
    return True

Четвертый шаг - обновление коллекции векторного хранилища. На входе - мастер-таблица и бинарный файл с контейнерами документов. Документы удаляются или добавляются в хранилище по их ID, в качестве которого используется хеш-имя из мастер-таблицы. Если документ с таким хеш-именем уже есть в коллекции, он заново векторизируется и сохраняется на место старого вектора. Так работает обновление по документу. Превращение документа в вектор реализовано внутри объекта vector_store.

Код обновления векторного хранилища

def vstore_update(vector_store: QdrantVectorStore) -> bool:
    """ Update the vectors """
    print('\nUpdating the vector store...')
    try:
        master_data = pd.read_csv(os.path.join(STORAGE_PATH, 'master_data.csv'), sep=';')
    except FileNotFoundError:
        print(f'Master data not found, exit')
        return False
    try:
        with open(os.path.join(STORAGE_PATH, 'documens.pkl'), 'rb') as handle:
            docs = pickle.load(handle)
    except FileNotFoundError:
        print(f'Documents not found, exit')
        return False

    # Create a lookup table to find a document by its hash-name.
    lookup = dict()
    for i in range(len(docs)):
        doc = docs[i]
        lookup[doc.metadata.get('hash_name', 'NF')] = i
    
    deleting = 0
    loading = 0
    for i in master_data.index:
        status = master_data.loc[i, 'status']
        hash_name = master_data.loc[i, 'hash_name']

        # Deleting
        if status == 'delete':
            # If chunking is used:
            # Don't forget to remove the splits for the updated document, 
            # otherwise the content will be duplicated.
            # Here is a cycle through document splits, 
            # or you can use a batch of splits if the documents are small in size.
            try:
                vector_store.delete(ids=[hash_name])
                master_data.loc[i, 'status'] = 'delete'
                deleting += 1
            except Exception as ex:
                print(f"Error while deleting: {ex}")
            
        # Adding new and updating vectors
        if status in ['new', 'update']:
            # Here is a cycle through document splits if chunking is used...
            try:
                doc = docs[lookup[hash_name]]
                vector_store.add_documents(documents=[doc], ids=[hash_name])
                master_data.loc[i, 'status'] = 'load'
                loading += 1
            except Exception as ex:
                print(f"Error while updating: {ex}")

    # Drop records with "delete" status
    master_data = master_data.loc[master_data.status != 'delete', :]
    master_data.to_csv(f'{STORAGE_PATH}/master_data.csv', sep=';', index=False)
    print('Current master data saved')
    print(f"Deleted files: {deleting}")
    print(f"Loaded files: {loading}")
    return True

Собственно, ETL-скрипт это тоже функция, которая последовательно запускает описанные шаги, если шаг возвращает «ложь», процесс останавливается.

Код ETL функции

def updating_script(vector_store: QdrantVectorStore) -> bool:
    scan_state = scan_dir(directory=ROOT_DOC_PATH)
    if not scan_state:
        return False

    delta_state = create_delta()
    if not delta_state:
        return False

    download_state = download_docs()
    if not download_state:
        return False

    update_state = vstore_update(vector_store=vector_store)
    if not download_state:
        return False
    return True

Давайте протестируем то, что у нас получилось.

Сценарий I. Первоначальная загрузка документов

Создадим два документа, один из которых сохраним в корневой папке, а второй в папке yaf (yet another folder) внутри корневой. Так сможем проверить обход директорий. Для демонстрации я подготовил функцию, которая возвращает метаданные первых десяти векторов в коллекции:

def collection_scroll(client: QdrantClient, collection_name: str = "Collection_D401") -> None:
    points, _ = client.scroll(
        collection_name="Collection_D401",
        limit=10, # Number of points to retrieve per page
        offset=None,
        with_payload=True,
        with_vectors=False
    )
    for p in points:
        for k, v in p.payload.get('metadata').items():
            print(f"{k}: {v}")
        print()

Запускаем скрипт, получаем вот такой лог:

Files scanning...
Recursive traversing Complete, total files: 2

Increment calculation...
Master data not found, create new
Current master data saved
Deleted files: 0
Updated files: 0
New files: 2

Load docs...
New and updated docs: 2

Updating the vector store...
Current master data saved
Deleted files: 0
Loaded files: 2
CPU times: total: 3.44 s
Wall time: 701 ms
True

Смотрим на коллекцию:

collection_scroll(client=qdrant_client)

title: Статья 15. Запрет на самолечение и неавторизованную регенерацию
hash_name: 2121124bad94b5b0b8a6641d72d41292
mod_date: 2026-05-23 13:49:06.028412
size: 537
status: new

title: Статья 14. Запрет на несанкционированное саморазвитие и био-взлом
hash_name: 62fc20a3e20bb7b23efac3fe3ee05d08
mod_date: 2026-05-23 13:49:06.027412
size: 601
status: new

Этот сценарий можно запустить повторно (без повторной загрузки текстов), процесс должен остановиться с сообщением «Nothing to update, exit».

Лог скрипта

Files scanning...
Recursive traversing Complete, total files: 2

Increment calculation...
Current master data saved
Deleted files: 0
Updated files: 0
New files: 0

Load docs...
There is nothing to update, exit

False

Как всё вернуть в исходное состояние, если потребуется?

- через клиент удалите коллекцию Collection_D401 в Qdrant
- в проектной папке удалите директории data (корневая папка) и storage (папка с промежуточными результатами ETL)

Сценарий II. Новый документ

Создаем новый документ и сохраняем его в ту же папку yaf. Запускаем скрипт.

Лог скрипта

Files scanning...
Recursive traversing Complete, total files: 3

Increment calculation...
Current master data saved
Deleted files: 0
Updated files: 0
New files: 1

Load docs...
New and updated docs: 1

Updating the vector store...
Current master data saved
Deleted files: 0
Loaded files: 1
CPU times: total: 1.58 s
Wall time: 340 ms

True

Сценарий III. Удаляем документ

Давайте удалим документ из корневой папки:

os.remove(os.path.join(ROOT_DOC_PATH, "D401A14.md"))

Лог скрипта

Files scanning...
Recursive traversing Complete, total files: 2

Increment calculation...
Current master data saved
Deleted files: 1
Updated files: 0
New files: 0

Load docs...
New and updated docs: 0

Updating the vector store...
Current master data saved
Deleted files: 1
Loaded files: 0
CPU times: total: 31.2 ms
Wall time: 80.9 ms

True

Сценарий IV. Обновление документа

Добавим текста в D401A16.md и запустим скрипт.

До обновления:
title: Статья 16. Регламент фильтрации сенсорного восприятия и управления доступом
hash_name: 193226b849538bae28eae4e9e716dce5
mod_date: 2026-05-20 13:02:05.476812
size: 563
status: new

После обновления:
title: Статья 16. Регламент фильтрации сенсорного восприятия и управления доступом
hash_name: 193226b849538bae28eae4e9e716dce5
mod_date: 2026-05-20 13:14:33.882040
size: 1002
status: update

Видим, что время модификации, размер и статус изменились.

Лог скрипта

Files scanning...
Recursive traversing Complete, total files: 2

Increment calculation...
Current master data saved
Deleted files: 0
Updated files: 1
New files: 0

Load docs...
New and updated docs: 1

Updating the vector store...
Current master data saved
Deleted files: 0
Loaded files: 1
CPU times: total: 2.66 s
Wall time: 506 ms

True

Замечания по масштабированию

1. Для обхода папок файловой системы я предпочитаю использовать os.walk. Рекурсивная функция поможет, если вместо файловой системы будет облачное хранилище. Главное, чтобы клиент хранилища поддерживал функции list_dir, is_dir или is_file. Тогда изменения в коде рекурсивной функции будут минимальны. Проверено на NextCloud и YAD. Также потребуется скорректировать операции по загрузке документов.

2. Код работает только c текстами в формате txt или markdown. Парсинг и конвертация - отдельная большая тема. Потребуется дополнительный шаг в ETL-скрипте сразу после калькуляции дельты. Я рекомендую конвертировать новые и обновленные документы в markdown, а затем сохранять их в отдельную папку c хеш-именем вместо названия файла (это можно и нужно делать в несколько потоков). И уже из этой папки загружать на следующем шаге.

3. Функции не используют внешние переменные в основном скрипте, всё взаимодействие организовано через файловую систему, то есть первая функция сохраняет файл, следующая его считывает. Это сделано специально, чтобы можно было быстро построить направленный ациклический граф (DAG), и реализовать ETL в AirFlow.

4. Про чанкинг - я пометил комментариями соответствующие места в функции загрузки документов, где должен быть реализован этот механизм. Нужно учитывать, что каждый фрагмент должен иметь свой уникальный id, обычно это uuid4 из библиотеки uuid. Фрагменты связаны с документом через хеш-имя в метаданных. При обновлении фрагментов одного и того же документа у них будут другие id и векторы в итоге задублируются. Поэтому придётся добавить в процедуру обновления полное удаление фрагментов обновляемого документа.

5. Удалять фрагменты удобно через функцию фильтрации метаданных Qdrant (пример фильтрации есть в конце ноутбука). Добавлять новые можно по одному, если эмбеддинг работает на CPU, списком или батчами, если есть поддержка CUDA.

6. В коде ETL-скрипта не предусмотрена проверка дублирования документов. Если один и тот же документ записать в разные папки, обе копии будут загружены. Проблема может быть решена разными способами. Самый простой - проверять на дублирование прямо в ретривере. Вариант посложнее - отслеживать уникальность файлов при сканировании.

7. В реальной системе лучше использовать asyncio и асинхронные функции для работы с векторным хранилищем (aadd_documents, adelete). Класс QdrantVectorStore поддерживает асинхронное взаимодействие.

Инкрементальное обновление векторной базы по отдельным документам занимает около секунды на документ (у меня ~300 мс на чанк с эмбеддингом deepvk/USER-bge-m3 на CPU). Мы работаем только с инкрементом, поэтому экономим часы компьюта на парсинг и векторизацию всего массива документов.

При высокой динамике обновления (например, логи или сообщения в чате) существует ненулевая вероятность, что в момент обновления релевантные векторы будут отсутствовать в коллекции. Но технически ничего не сломается. Просто LLM выдаст некорректный ответ. Так как векторное хранилище «не зачищается», вопрос остановки сервиса переходит в категорию оценки качества ответа. Я знаю несколько проектов, где сервис не останавливают, а обновление проводят в периоды наименьшей нагрузки, запуская ежедневно по планировщику.

А как в ваших проектах устроено обновление RAG-системы? Останавливаете сервис?
Напишите пожалуйста в комментариях.

Полезные ссылки:
LangChain + Qdrant
Фильтрация в Qdrant
Концепция DAG в Airflow
Код туториала

RAG без downtime: настраиваем инкрементальное обновление документов на Qdrant и LangChain