Как проектировать distributed locking для LLM agents?

Краткий тезис

locking|Distributed locking (locking|распределённая блокировка) — критический механизм для координации параллельно работающих LLM-агентов, предотвращающий race conditions при доступе к общим ресурсам (базы данных, кэш, стейт-машина). Основные подходы — пессимистичные блокировки на базе алгоритма Redlock (Redis) и оптимистичные блокировки с версионированием. Ключевые проблемы: split-brain, корректный расчёт TTL, фальсификация блокировки (lock falsification). Лучшая практика — отдавать предпочтение оптимистичным блокировкам в сценариях с низкой конкуренцией и использовать идентификаторы блокировки с достаточным временем жизни.

1. Зачем нужны блокировки в контексте LLM-агентов

LLM-агенты — это автономные сущности, которые могут одновременно обновлять одно состояние: память (векторное хранилище), счётчики использованных токенов, разделяемые между агентами очереди задач. Без блокировки два агента могут прочитать одно значение, параллельно изменить его и записать — последняя запись перетрёт предыдущую, данные разрушатся.

Пример race condition

• Агент А читает количество оставшихся токенов: 1000.
• Агент Б читает то же значение: 1000.
• Агент А вычитает 200 и записывает 800.
• Агент Б вычитает 150 и записывает 850 (потеря агента А). Результат: реально израсходовано 350, а записано 150 — ошибка учёта бюджета.

locking|Distributed locking гарантирует, что в каждый момент только один агент выполняет критическую секцию (update ресурса).

2. Основные термины

Race condition — состояние, в котором результат операции зависит от порядка выполнения параллельных потоков/процессов. Приводит к непредсказуемому поведению.

Distributed lock — механизм, позволяющий процессу (агенту) в распределённой системе временно получить исключительный доступ к ресурсу. Реализуется через внешний координатор (Redis, ZooKeeper, etcd).

Critical section — участок кода, который не должен выполняться одновременно несколькими агентами.

TTL (Time-To-Live) — время жизни блокировки. Если агент удерживает блокировку дольше TTL, она автоматически снимается, чтобы предотвратить вечное ожидание (deadlock).

Lease — контракт на удержание блокировки на ограниченное время. Агент должен продлять lease (heartbeat), если не завершил операцию.

Split-brain — ситуация, когда два агента считают, что владеют одной и той же блокировкой (ложный захват).

Fencing token — монотонно возрастающий номер, выдаваемый при захвате блокировки. Сервер, защищающий ресурс, отклоняет запросы с устаревшим fencing token. Это решает проблему фальсификации блокировки.

3. Сценарии применения в AI-агентах

4. Критерии проектирования блокировки

При выборе механизма блокировки нужно ответить на вопросы:

1. Какой уровень конкуренции Если конфликты редкие — оптимистичная блокировка (дешевле, нет простоев).
2. Как долго может выполняться критическая секция? TTL должен превышать max execution time операции, иначе блокировка снимется преждевременно, и другой агент сможет зайти в секцию.
3. Допустима ли потеря блокировки Если агент упал без снятия блокировки, TTL автоматически освободит ресурс (защита от deadlock).
4. Есть ли поломка синхронизации часов В распределённой системе часы могут дрейфовать, что критично для алгоритма Redlock.

5. Реализация на базе Redis: алгоритм Redlock

5.1 Идея

Предложен Antirez (автор Redis). Использует N независимых Redis-инстансов. Клиент пытается захватить блокировку сразу на всех (всех или большинстве > N/2). Если удалось на большинстве — блокировка считается захваченной.

Шаги:

1. Получить текущее время t1.
2. На каждом Redis выполнить SET resource_name unique_id NX PX timeout (NX — only if not exists, PX — milliseconds TTL). unique_id — случайное значение (UUID) для безопасности.
3. Для каждого успешного SET запоминаем время t_set.
4. Если количество успешных SET >= N/2 + 1 и разница t1 - t_last_set меньше TTL, блокировка захвачена.
5. Если не захвачена — снять блокировку на всех инстансах (команда DEL с проверкой unique_id).

Пример кода на Python (redis-py):

import redis
import time
import uuid

class Redlock:
    def __init__(self, redis_nodes: list):
        self.redis_nodes = [redis.StrictRedis(host=node['host'], port=node['port']) for node in redis_nodes]
        self.quorum = len(redis_nodes) // 2 + 1
        self.ttl = 10000  # milliseconds

    def acquire(self, resource: str, timeout_ms: int = 10000) -> tuple[bool, str]:
        uuid_val = str(uuid.uuid4())
        start = int(time.time() * 1000)
        n_success = 0
        # Сбрасываем таймер на каждом узле
        for r in self.redis_nodes:
            try:
                if r.set(resource, uuid_val, nx=True, px=timeout_ms):
                    n_success += 1
            except Exception:
                pass
        elapsed = int(time.time() * 1000) - start
        if n_success >= self.quorum and elapsed < timeout_ms:
            return True, uuid_val
        else:
            # Rollback
            self.release(resource, uuid_val)
            return False, None

    def release(self, resource: str, uuid_val: str):
        # Проверяем, что блокировка наша, и удаляем
        script = """
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        """
        for r in self.redis_nodes:
            r.eval(script, 1, resource, uuid_val)

5.2 Проблемы Redlock (Martin Kleppmann и другие)

• Split-brain из-за дрейфа часов: если часы на узлах рассинхронизированы, агент может думать, что время ещё не вышло, а блокировка уже истекла (другой агент захватит её).
• GC pause (stop-the-world): JVM/CPython может делать длительные паузы GC. Агент может не заметить, что TTL истёк.
• Необходимость fencing token: Redlock не генерирует монотонные номера, поэтому если ресурс не защищён fencing token, два агента могут одновременно писать.

Рекомендация: используйте Redlock только в сценариях, где допустимы редкие отказы (best-effort), или дополняйте fencing token на стороне хранилища.

6. Оптимистичные блокировки (version-based)

Принцип: не блокировать ресурс на время операции, а перед записью проверять, изменился ли ресурс с момента чтения.

Как работает (на примере PostgreSQL):

• Таблица agent_state содержит поле version (монотонный счётчик).
• Агент читает запись вместе с version.
• Выполняет вычисления.
• Запись: UPDATE agent_state SET data = new_data, version = version + 1 WHERE id = xxx AND version = old_version.
• Если ROW_COUNT = 0 → конфликт, нужно повторить операцию (retry).

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

• Нет блокировок (не нужно удерживать блокировку, нет deadlock, не нужен TTL).
• Высокая производительность при низкой конкуренции.
• Легко масштабируется.

Недостатки

• При высокой конкуренции много retry.
• Не подходит для операций с побочными эффектами (например, вызов внешнего API).

Пример для Redis (CAS — compare-and-set):

import redis

r = redis.Redis()
# Условный ключ с версией
key = "agent_counter"
# Агент читает
counter = int(r.get(key) or 0)
version = int(r.get(f"{key}:version") or 0)
# Вычисляет новое значение
new_counter = counter + 1
new_version = version + 1
# CAS: атомарно обновляем, если версия не изменилась
lua = """
redis.call("SET", KEYS[1], ARGV[1])
redis.call("SET", KEYS[2], ARGV[2])
"""
# Но в Redis нет встроенной CAS на одном ключе с версией. Используем WATCH+TRANSACTION
with r.pipeline() as pipe:
    while True:
        try:
            pipe.watch(key, f"{key}:version")
            curr_counter = int(pipe.get(key) or 0)
            curr_version = int(pipe.get(f"{key}:version") or 0)
            if curr_version != version:
                raise Exception("Conflict")
            pipe.multi()
            pipe.set(key, new_counter)
            pipe.set(f"{key}:version", new_version)
            pipe.execute()
            break
        except redis.WatchError:
            continue

Вывод для большинства сценариев с AI-агентами (обновление памяти, счётчики) предпочтительна оптимистичная блокировка — она проще, безопаснее при падениях и не требует TTL.

7. Мониторинг и observability блокировок

При проектировании блокировок нужно предусмотреть метрики, иначе проблемы (deadlock, высокая задержка) останутся незамеченными.

Инструменты: Prometheus + Grafana для сбора и визуализации, OpenTelemetry для трассировки отдельных операций агентов.

8. Альтернативы Redis: Zookeeper / etcd

Если ваша инфраструктура уже использует ZooKeeper или etcd, блокировка на их основе может быть надёжнее, потому что они предоставляют строгую последовательность и fencing token.

ZooKeeper

• Создаём эфемерный (ephemeral) последовательный (sequential) узел, например /lock/lock-000000001.
• Агент читает все дочерние узлы /lock/. Если его узел самый маленький (минимальный sequence) — блокировка получена.
• При потере соединения эфемерный узел удаляется автоматически (защита от deadlock).
• Fencing token — это sequence номер узла.

etcd

• Использует API с lease и CAS (compare-and-swap) на ключе.
• Также выдаёт монотонный revision при создании ключа, который можно использовать как fencing token.

Сравнение

9. Лучшие практики для LLM-агентов

1. Использовать уникальный идентификатор блокировки — UUID или агентский token. Это предотвращает случайное снятие блокировки другим агентом.
2. Выставлять TTL с запасом — 2–3× больше типичного времени выполнения критической секции, но не настолько большой, чтобы блокировка долго держалась при падении.
3. Реализовать heartbeat — если операция может длиться дольше TTL, агент должен периодически обновлять TTL (SET ... PX <новое время>).
4. Предпочитать оптимистичные блокировки — они проще, не требуют управления временем и меньше нагружают координатор.
5. Оборачивать захват блокировки в контекстный менеджер (with-блок), чтобы гарантировать освобождение при исключениях.
6. Добавить fencing token на стороне ресурса (например, поле version в каждой записи, которая обновляется агентом). Тогда даже при ложном захвате блокировки старые данные будут отвергнуты.
7. Мониторить все операции с блокировками через OpenTelemetry spans.

Пример использования контекстного менеджера с оптимистичной блокировкой:

import redis
from contextlib import contextmanager

redis_client = redis.Redis()

@contextmanager
def optimistic_lock(resource_key, max_retries=3):
    version_key = f"{resource_key}:version"
    for attempt in range(max_retries):
        with redis_client.pipeline() as pipe:
            try:
                pipe.watch(resource_key, version_key)
                value = pipe.get(resource_key)
                version = int(pipe.get(version_key) or 0)
                yield value, version  # отдаём текущее значение и версию в блок
                # После завершения блока try — обновляем
                new_value, new_version = ...  # вычисляется в блоке with
                pipe.multi()
                pipe.set(resource_key, new_value)
                pipe.set(version_key, new_version)
                pipe.execute()
                break
            except redis.WatchError:
                if attempt == max_retries - 1:
                    raise
                continue

Пет-проект для закрепления

Задача Разработать систему безопасного обновления долговременной памяти (векторного хранилища) тремя LLM-агентами, работающими параллельно.

Инструменты

• Python (asyncio + aiohttp)
• Redis (как распределённый лог блокировок)
• TinyDB или локальный FAISS (для упрощения — SQLite с векторным расширением)
• docker-compose (Redis + три контейнера с агентами)

Шаги:

1. Создать хранилище ключ-значение с полем version (использовать SQLite с колонкой version INTEGER).
2. Реализовать оптимистичную блокировку Redis: при чтении записи сохранять её версию, при записи — CAS с проверкой версии.
3. Написать класс AgentWorker который в бесконечном цикле:
   • Читает новую задачу (например, add_document(doc_id, embedding)).
   • Захватывает блокировку на doc_id.
   • Эмулирует долгую операцию (time.sleep 0.5–1 сек).
   • Обновляет запись в SQLite (INSERT OR UPDATE) с проверкой версии.
   • При конфликте читает свежую версию и повторяет.
4. Запустить 3 экземпляра агентов параллельно (threading или multiprocessing).
5. Добавить мониторинг: счётчик успешных/конфликтных обновлений.

Ожидаемый результат

• Все документы будут добавлены без потерь (итоговое количество совпадает с количеством задач).
• В логах видны конфликты (WatchError) и повторные попытки.
• Система не падает при отказе одного агента (блокировка освобождается автоматически по TTL).

Связь с другими вопросами

Навигация

• Предыдущий: 827
• Следующий: 829
• Индекс: 00. Индекс разборов