Кэш пуст, а 20 одинаковых запросов уже ушли в LLM: singleflight на Python без общей отмены

Кэш хранит готовые ответы, но не мешает нескольким запросам одновременно получить cache miss и запустить одну и ту же генерацию. Разбираю singleflight для read-only LLM-вызовов: один общий запрос, отдельные дедлайны клиентов и проверка, что отмена первого клиента не ломает остальных.

Почему семафор не решает эту задачу

Представим учебный сценарий: несколько пользователей одного tenant запрашивают краткое резюме одной версии общего документа. Контекст, права доступа и настройки генерации совпадают. Готового ответа пока нет.

Если каждый обработчик делает «проверить кэш → вызвать LLM → записать ответ», все могут пройти первую проверку до появления результата. Даже семафор на четыре слота только растянет дубли во времени: оставшиеся запросы всё равно дождутся своей очереди, если внутри никто повторно не проверит результат.

Singleflight хранит не ответ с TTL, а ещё выполняющуюся задачу. Первый обработчик создаёт её, остальные с тем же ключом ждут тот же результат. После завершения запись удаляется. Следующий вызов снова запустит работу — если отдельно не добавить обычный кэш.

Это не провайдерский prompt caching и не идемпотентность внешнего действия. Здесь мы подавляем дубли незавершённого вычисления внутри своего приложения.

Главное условие — право разделить один результат

Одинаковый текст запроса ещё не означает одинаковую работу. Ключ должен учитывать границу tenant и доступа, провайдера и маршрут, точную версию модели, параметры генерации, system prompt, версии документов или retrieval snapshot и остальные входы, влияющие на результат. Хеширование неполного ключа не делает его безопасным.

Для первого внедрения я бы ограничился генерацией по одному неизменяемому документу в одной области доступа. Авторизация каждого клиента происходит до присоединения к общей задаче. Если права различаются или меняются — запросы не объединяем без отдельного решения об изоляции.

Ещё одно продуктовое условие: все участники действительно согласны получить один и тот же ответ. Если пользователь просит «ещё один вариант», объединение уничтожит смысл запроса, даже при одинаковом prompt. Отдельные творческие попытки лучше исключить из singleflight.

Минимальная реализация: Python 3.11+

Объект создаётся при старте приложения внутри работающего event loop и используется всеми обработчиками этого процесса. Не создавайте его заново на каждый HTTP-запрос.

Ниже возвращается строка — неизменяемый результат. Для изменяемых словарей или SDK-объектов понадобятся независимые копии либо явный неизменяемый контракт. factory — функция, создающая awaitable; для уже занятого ключа функция нового участника не вызывается. Поэтому равенство ключей обязано означать эквивалентность всех входов.

import asyncio import math from collections.abc import Awaitable, Callable, Hashable class Busy(RuntimeError): pass class SingleFlight: def __init__(self, max_keys=128, job_timeout=20.0): if not isinstance(max_keys, int) or max_keys < 1: raise ValueError('max_keys must be a positive integer') if not math.isfinite(job_timeout) or job_timeout <= 0: raise ValueError('job_timeout must be finite and positive') self._loop = asyncio.get_running_loop() self._max_keys = max_keys self._job_timeout = job_timeout self._tasks = {} self._closed = False async def do(self, key: Hashable, factory: Callable[[], Awaitable[str]]) -> str: if asyncio.get_running_loop() is not self._loop: raise RuntimeError('use only in the original event loop') if self._closed: raise RuntimeError('SingleFlight is closed') task = self._tasks.get(key) if task is not None and task.done(): self._finished(key, task) task = None if task is None: if len(self._tasks) >= self._max_keys: raise Busy('too many distinct in-flight keys') deadline = self._loop.time() + self._job_timeout task = asyncio.create_task(self._run(factory, deadline)) self._tasks[key] = task task.add_done_callback(lambda done: self._finished(key, done)) return await asyncio.shield(task) async def _run(self, factory, deadline): async with asyncio.timeout_at(deadline): return await factory() def _finished(self, key, task): if self._tasks.get(key) is task: self._tasks.pop(key) if not task.cancelled(): task.exception() # retrieve even if every waiter has left async def aclose(self): self._closed = True tasks = tuple(self._tasks.values()) for task in tasks: task.cancel() await asyncio.gather(*tasks, return_exceptions=True)

Почему здесь shield, а не просто await task

При прямом await отмена ожидающего обработчика может уйти в общую задачу. asyncio.shield оставляет отмену у конкретного клиента: он получит CancelledError, но другие участники продолжат ждать. Сильную ссылку на общую Task сохраняет словарь.

У общей работы свой абсолютный deadline, рассчитанный при создании. У каждого клиента — свой deadline ожидания, который можно задать снаружи через asyncio.timeout_at(client_deadline). Короткий бюджет первого клиента не должен становиться бюджетом всех остальных. Если внутри factory есть ожидание семафора, оно входит в общий бюджет; перед отправкой внешнего запроса также стоит явно проверить, что deadline ещё не истёк.

В примере осознанно нет счётчика участников: даже если ушли все, задача продолжит работу до своего лимита. Это проще по жизненному циклу, но может потратить деньги на никому уже не нужный ответ. Для дорогих генераций понадобится отдельная политика отмены последнего участника. И даже отмена локальной Task не доказывает, что провайдер остановил вычисление или не выставит счёт.

Между поиском ключа и сохранением Task нет await. Для одного event loop со стандартным планированием этого достаточно; из других потоков или процессов так обращаться нельзя. Пример не рассчитан на пользовательские eager/re-entrant task factories.

Финальный callback убирает только свою Task: поздний callback старой задачи не должен удалить новую запись того же ключа. Вызов task.exception() забирает исключение, даже если уже никто не ждёт; при реальном внедрении здесь нужна также метрика или безопасное логирование ошибки, без prompt и секретов.

Тест: первый клиент ушёл, остальные дождались

Этот тест можно поставить ниже класса и запустить одним файлом. Здесь нет обращения к платному API: upstream заменён управляемой корутиной. Event удерживает её до присоединения участников; sleep(0) только отдаёт ход планировщику, а не имитирует сетевую задержку.

async def demo(): sf = SingleFlight() started, release = asyncio.Event(), asyncio.Event() calls = 0 async def upstream(): nonlocal calls calls += 1 started.set() await release.wait() return 'ready' # This short key is ONLY for a local test, not production. first = asyncio.create_task(sf.do('demo-only', upstream)) await started.wait() followers = [ asyncio.create_task(sf.do('demo-only', upstream)) for _ in range(19) ] await asyncio.sleep(0) # let followers join first.cancel() cancelled = await asyncio.gather(first, return_exceptions=True) assert isinstance(cancelled[0], asyncio.CancelledError) assert calls == 1 release.set() assert await asyncio.gather(*followers) == ['ready'] * 19 assert calls == 1 assert await sf.do('demo-only', upstream) == 'ready' assert calls == 2 # not a result cache await sf.aclose() print('OK: coalescing, isolated cancellation, no result cache') asyncio.run(demo())

Что проверено — и чего тест не доказывает

В локальном запуске прошли проверки: одна upstream-корутина на 20 участников; отмена первого клиента; отсутствие кэша готовых ответов; разные ключи; лимит уникальных ключей с присоединением к уже существующему; общая ошибка и новый вызов после неё; отдельный таймаут клиента; таймаут общей работы и очистка записи; запрет новых вызовов после aclose. Это тесты механики asyncio, не замер экономии на реальном LLM API.

Что нужно добавить перед продакшеном

Лимит max_keys ограничивает разные незавершённые задачи, а не число ожидающих клиентов. Один горячий ключ всё ещё способен собрать слишком много ожиданий. Нужны входные rate limits, ограничение участников и память под ответы. Busy следует обрабатывать как перегрузку, а не немедленно повторять запрос в цикле.

Важна работа с contextvars: create_task по умолчанию копирует контекст создателя. Передавайте в factory явные неизменяемые входы и корректно выбранные credentials; нельзя случайно подхватить current_user первого обработчика и выдать результат остальным. Ключ не заменяет авторизацию.

Дедлайн asyncio кооперативный: блокирующий код или подавленный CancelledError могут нарушить ожидаемое поведение. У HTTP-клиента должны быть собственные ограничения соединения и чтения. При остановке приложения вызовите aclose из lifecycle-hook и дождитесь завершения; для некорректного upstream потребуется отдельная политика остановки.

Это только один процесс и один event loop. При нескольких worker каждый создаст собственную генерацию. Распределённое объединение — отдельный протокол с lease, владением результатом и восстановлением после падения, а не просто Redis-lock вокруг сети.

Пример не объединяет SSE-потоки и не исполняет tool calls. Для streaming нужны буферизация, правила позднего подключения и backpressure. Для платежа или записи в CRM требуется идемпотентность операции, а не общий ответ нескольких клиентов. Общая ошибка также не лечится автоматически: синхронная волна повторов снова может ударить в провайдера.

Практический вывод

Я бы начинал с одного read-only маршрута, где дубли действительно встречаются и общий ответ допустим. Измерять стоит число созданных общих задач, число присоединившихся клиентов, ошибки, перегрузку, время ожидания и работы без оставшихся участников. Экономию считать по фактическим вызовам и расходам провайдера: один локальный factory не гарантирует один HTTP-запрос, если SDK сам делает retries. Singleflight не ускоряет единственную генерацию — он не даёт соседним запросам оплатить её дубли.

Первоисточники: Python — asyncio.shield, asyncio.timeout_at и create_task (https://docs.python.org/3/library/asyncio-task.html); описание duplicate suppression в Go singleflight (https://pkg.go.dev/golang.org/x/sync/singleflight).

У вас чаще встречаются повторные запросы к одному документу или почти все prompt уникальны? Где вы провели бы границу между полезным объединением и потерей независимых вариантов ответа?