Синхронизация кэша LLM с векторным хранилищем для юридического чат-бота

Сделайте кэш LLM осведомлённым о корпусе: синхронизация с векторным хранилищем, счётчик документов, LangChain-хук и компромиссы для юридического чат-бота.

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

Суть проблемы

Обычно ответы кэшируются по тексту запроса и «подписи» модели. Это работает, пока база знаний неизменна. Если кэш не видит обновлений корпуса, он без сомнений вернёт то, что вчера было верно, а сегодня — уже неполно.

from langchain_core.caches import BaseCache, RETURN_VAL_TYPE
from typing import Any, Dict, Optional
class NaiveResponseCache(BaseCache):
    def __init__(self):
        super().__init__()
        self._store = {}
    def fetch(self, prompt: str, llm_fingerprint: str) -> Optional[RETURN_VAL_TYPE]:
        entry = self._store.get((prompt, llm_fingerprint))
        if entry:
            return entry["value"]
        return None
    def store(
        self,
        prompt: str,
        llm_fingerprint: str,
        value: RETURN_VAL_TYPE,
        meta: Dict[str, Any] = None,
    ) -> None:
        self._store[(prompt, llm_fingerprint)] = {
            "value": value,
            "meta": meta or {},
        }

Этот подход не знает, выросло ли векторное хранилище с N до N+k документов. В результате он продолжит отдавать ответы, игнорирующие новую информацию.

Что именно идёт не так

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

Практичный способ синхронизировать кэш с обновлениями

Простой приём — сделать кэш «осведомлённым о корпусе»: отслеживать число документов и заново генерировать кэшированные записи при изменении этого числа. Это базовая эвристика: если счётчик поменялся, база знаний изменилась, значит ответы в кэше нужно обновить. Ниже — минимальная реализация этой идеи.

from langchain_core.caches import BaseCache, RETURN_VAL_TYPE
from typing import Any, Dict, Optional
class CorpusVersionCache(BaseCache):
    def __init__(self, doc_total: int):
        super().__init__()
        self._entries = {}
        self._corpus_size = doc_total
    # Задать или обновить известный размер корпуса
    def set_corpus_size(self, doc_total: int):
        if self._corpus_size == doc_total:
            return
        self._corpus_size = doc_total
        for args, _prev in self._entries.items():
            query, llm_repr = args[0], args[1]
            value, meta = self.rebuild_entry(query, llm_repr)
            self.store(query, llm_repr, value, meta)
    def rebuild_entry(self, query: str, llm_repr: str):
        # Пересобрать ответ с учётом новой информации
        response = "New LLM Response"
        metadata = {}
        return response, metadata
    # Поиск в кэше
    def fetch(self, query: str, llm_repr: str) -> Optional[RETURN_VAL_TYPE]:
        cached = self._entries.get((query, llm_repr))
        if cached:
            return cached["value"]
        return None
    # Обновление кэша
    def store(
        self,
        query: str,
        llm_repr: str,
        value: RETURN_VAL_TYPE,
        metadata: Dict[str, Any] = None,
    ) -> None:
        self._entries[(query, llm_repr)] = {
            "value": value,
            "metadata": metadata or {},
        }

Чтобы подключить это к глобальному хуку кэша LangChain, настройте так:

from langchain.globals import set_llm_cache
cache = CorpusVersionCache(doc_total=0)
set_llm_cache(cache)

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

Компромиссы производительности

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

Почему это важно

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

Выводы

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