Векторный кэш: делаем умные ответы еще быстрее

Введение

Сегодня чат-боты и интеллектуальные ассистенты широко применяются в различных сферах: поддержка клиентов, корпоративные системы, поисковые сервисы и во многих других. Для их разработки часто используют архитектуру Retrieval-Augmented Generation (RAG), которая объединяет генерацию ответа с поиском данных во внешних источниках. Такой подход помогает ботам и ассистентам давать более точные и актуальные ответы. Но на практике оказывается, что RAG сталкивается с проблемой повторяющихся запросов, из-за которой система многократно выполняет одни и те же вычисления, повышая нагрузку и время отклика.

Всем привет! Меня зовут Вадим, я Data Scientist в компании Raft, и в этой статье мы разберемся, что такое векторный кэш и как его использовать. Давайте начнем!

Краткий обзор RAG

Перед началом знакомства с векторным кэшем давайте кратко рассмотрим, как базово работает система RAG. В целом её можно поделить на 2 большие части:

Индексация данных (Data Indexing): на этом этапе происходит сбор, обработка и преобразование документов в векторные представления (эмбеддинги), которые вносятся в векторную базу данных для дальнейшего поиска по ним.
Поиск и генерация ответа (Data Retrieval & Generation): на этом этапе пользователь вводит свой запрос, система преобразует его в вектор, по которому ищет top-K наиболее релевантных фрагментов (chunks) для формирования контекста. Далее, опираясь на запрос пользователя, заранее заготовленные инструкции и информацию из векторной базы данных, LLM формирует конечный ответ.

Определение проблемы

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

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

На практике же оказывается, что около 30% всех пользовательских запросов семантически похожи между собой, и система фактически заново извлекает одни и те же или очень близкие данные.

Давайте рассмотрим, как механизм векторного кэша может помочь избежать повторных вычислений и повысить эффективность работы системы.

Что такое векторный кэш и как его готовить?

Как работает векторный кэш?

По сути, векторный кэш — это дополнительный уровень хранения, который позволяет сохранять уже сгенерированные ответы для запросов, которые «семантически» (то есть по смыслу) похожи между собой. Вместо того чтобы каждый раз заново искать и пересобирать ответ, система сначала проверяет: не встречался ли уже похожий запрос? Если да, то можно быстро вернуть готовый ответ из кэша, минуя лишние вычисления. В ином случае необходимо будет обратиться к векторной базе данных, провести все необходимые вычисления, а затем вставить ответ в хранилище кэша.

На практике это работает примерно так:

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

Важные компоненты

Для стабильной и эффективной работы векторного кэша необходимо учитывать следующие компоненты:

Выбор эмбеддингов: подбираем модель, которая умеет хорошо улавливать смысл запросов для нашей задачи.
Метрика семантической близости: решаем, как измерять «похожесть» векторов, чаще всего это косинусное сходство или евклидово расстояние.
Выбор порога сходства: задаем порог, при котором два запроса считаются достаточно похожими, чтобы использовать кэш. Подбирается экспериментально, но обычно такой порог меньше 0.5.
Политика наполнения и обновления кэша: определяем, как контролировать размер кэша и актуальность данных. На практике часто используется политика TTL, которая определяет время жизни кэша, но также есть и другие, такие как LRU, FIFO, LFU и так далее.

Схемы реализации

Реализовывать векторный кэш можно различными способами, давайте посмотрим на некоторые из них.

Реализация с нуля

Для понимания принципа работы векторного кэша рассмотрим пример его реализации с сохранением в json файле (также можно сохранять и в других представлениях, например в виде коллекций в векторной базе данных).

Для начала нам необходимо инициализировать класс работы с кэшем — SemanticCache и указать путь к файлу для хранения кэша, порог и максимальное числом запросов, которые мы можем хранить в файле. Здесь я использую простую стратегию вытеснения FIFO (First-In, First-Out) — старые записи будут удаляться первыми.

class SemanticCache:     def __init__(self, json_file: str = "cache_file.json", threshold: float = 0.35,                  max_response: int = 100, eviction_policy: Optional[str] = None, nprobe: int = 8):         """         Инициализация семантического кэша.          Args:             json_file (str): Путь к JSON файлу для хранения кэша.             threshold (float): Порог Евклидова расстояния, ниже которого считаем запросы похожими.             max_response (int): Максимальное количество записей в кэше.             eviction_policy (str, optional): Политика вытеснения (например, 'FIFO').             nprobe (int): Количество кластеров для поиска в Faiss.         """         # Инициализируем Faiss-индекс и энкодер         self.index, self.encoder = init_cache()          self.threshold = threshold         self.json_file = json_file         self.max_response = max_response         self.eviction_policy = eviction_policy         self.nprobe = nprobe          # Загружаем кэш из файла или создаём пустой, если файл не существует         self.cache = retrieve_cache(self.json_file)

Дальше я определяю метод search_in_cache, задача которого проверить, есть ли в кэше похожий запрос. В нем кодируется новый запрос в эмбеддинг, а затем с помощью Faiss ищется ближайший вектор. Если расстояние до него меньше установленного порога, считаем, что запрос достаточно похож, и возвращаем готовый ответ из кэша, в ином случае — возвращаем None.

def _search_in_cache(self, embedding: List[float]) -> Optional[str]:         """         Ищет похожий запрос в Faiss-индексе.          Args:             embedding (List[float]): Векторное представление нового запроса.          Returns:             Optional[str]: Найденный ответ из кэша, либо None, если подходящего ответа нет.         """         # Устанавливаем число кластеров для поиска         self.index.nprobe = self.nprobe          # Выполняем поиск ближайшего соседа         D, I = self.index.search(embedding, 1)         distance = D[0][0]         index = I[0][0]          # Проверяем, подходит ли найденный результат под пороговое значение расстояния         if index >= 0 and distance <= self.threshold:             return self.cache['response_text'][index]          # Если подходящего результата не найдено         return None

Чтобы кэш не рос бесконечно, определяется метод _evict_if_needed, в котором реализована логика удаления старых записей. Как только число записей превышает max_response, мы просто удаляем несколько самых первых (самых старых) элементов.

def _evict_if_needed(self):         """         Проверяет, не превышает ли размер кэша максимальное количество записей,         и при необходимости удаляет старые записи.         """         overflow = len(self.cache["questions"]) - self.max_response         if overflow > 0:             # Удаляем старые элементы из всех списков, чтобы кэш оставался синхронизированным             self.cache["questions"] = self.cache["questions"][overflow:]             self.cache["embeddings"] = self.cache["embeddings"][overflow:]             self.cache["answers"] = self.cache["answers"][overflow:]             self.cache["response_text"] = self.cache["response_text"][overflow:]

Наконец, основной метод ask объединяет всю логику вместе:

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

def ask(self, question: str, k: int) -> str:         """         Получает ответ из кэша или, если в кэше не найдено, из внешней базы (например, chromaDB).          Args:             question (str): Запрос пользователя.             k (int): Количество документов для получения из базы при промахе кэша.          Returns:             str: Текст ответа.         """         try:             # Кодируем запрос в вектор             embedding = self.encoder.encode([question])              # Пробуем найти ответ в кэше             cached_response = self._search_in_cache(embedding)              if cached_response:                 # Если нашли — возвращаем его                 return cached_response              # Если не нашли — делаем запрос к внешней базе             answers = query_database(question, k)              # Склеиваем тексты документов в единый ответ             response_text = "".join(doc.page_content for doc in answers)              # Добавляем новый запрос, ответ и эмбеддинг в кэш             self.cache['questions'].append(question)             self.cache['embeddings'].append(embedding[0].tolist())             self.cache['answers'].append(response_text)             self.cache['response_text'].append(response_text)              # Добавляем новый эмбеддинг в Faiss-индекс             self.index.add(embedding)              # Проверяем, не переполнился ли кэш, и при необходимости удаляем старые записи             if len(self.cache["questions"]) > self.max_response:                 self._evict_if_needed()              # Сохраняем обновленный кэш в файл             store_cache(self.json_file, self.cache)             return response_text          except Exception as e:             raise RuntimeError(f"Error in 'ask' method: {e}")

Таким образом данная реализация позволяет максимально гибко настроить систему кэширования и увеличить производительность системы.

Но зачем нам писать всё с нуля? Если есть возможность использовать уже готовые решения, например Redis.

Использование Redis

Redis — это высокопроизводительное хранилище данных в памяти (in-memory database), которое поддерживает структуру «ключ–значение» и множество дополнительных типов данных: списки, множества, хеши, упорядоченные множества и другие. Благодаря работе в оперативной памяти и продуманной архитектуре он обеспечивает быструю обработку запросов с минимальной задержкой.

Сейчас Redis активно развивается и как часть экосистемы GenAI, среди его функционала есть работа с векторным (семантическим) кэшем. Для его реализации можно использовать библиотеку redisvl с необходимым функционалом. Для начала необходимо создать объект для работы с кэшем, класс SemanticCache, в котором необходимо установить следующие параметры:

название индекса в Redis,
адрес подключения,
модель для векторизации текста,
порог семантической близости (distance_threshold),
время жизни кэша (ttl).

from redisvl.extensions.cache.llm import SemanticCache from redisvl.utils.vectorize import HFTextVectorizer  llmcache = SemanticCache(    name="llmcache",                                          # название поискового индекса    redis_url="redis://localhost:6379",                       # URL для подключения к Redis    distance_threshold=0.1,                                   # пороговое значение семантической близости для кэша    vectorizer=HFTextVectorizer("redis/langcache-embed-v1"),  # модель эмбеддингов    overwrite=True,    ttl= 60 * 60 * 24,   # время жизни записей в кэше (1 день) )

Далее реализовываем аналогичную логику работы с векторным кэшем, используя готовое решение от Redis.

def process_query(question, k, vector_store, llmcache):      # Проверяем, есть ли уже готовый ответ в кэше для данного запроса     response = llmcache.check(question)      if response:         # Если кэш сработал, выводим найденный ответ         print(f"Cache hit: {response}")     else:         # Если в кэше не найдено (cache miss)         print(f"Cache miss: {response}")          # Выполняем семантический поиск по векторной базе с топ-k результатами         results = vector_store.similarity_search(query, k=k)          # Объединяем тексты найденных документов в один контекст         context = "\n".join([result.page_content for result in results])          # Сохраняем в кэш: запрос и полученный контекст как ответ         llmcache.store(             prompt=question,             response=context,         )          # Выводим сформированный контекст         print(context)

Дополнительные возможности Redis

Кроме представленного механизма кэширования у Redis также другие AI инструменты:

Векторная база данных, которая по измерениям производительности разработчиков обходит другие подобные решения
Интеграция с популярными фреймворками для разработки GenAI приложений: LangChain, LlamaIndex
Закрытый продукт LangCache для более продвинутого и управляемого кэширования
Реализация семантической маршрутизации запросов

Векторная база данных Redis в сравнении с другими популярными решениями

Плюсы и минусы векторного кэша

Плюсы

Снижение задержек (latency): при повторных или схожих запросах можно избежать заново выполнения векторного поиска
Экономия ресурсов: снижается нагрузка на векторную базу данных

Минусы

Увеличения объёма кэша: при большом числе уникальных запросов кэш быстро увеличивается в размерах, необходимо определить оптимальную политику вытеснения
Дополнительные вычисления при вставке: каждая новая запись требует генерации эмбеддинга, обновления индекса и сохранения данных
Потенциальная устарелость результатов: если кэш не обновляется, ответы могут стать неактуальными по сравнению с обновлённой внешней базой данных

Выводы

Векторный кэш — это отличное решение, если вы хотите, чтобы ваша система быстрее отвечала и не тратила лишние ресурсы на похожие запросы.

Главное — не забывать следить за размером кэша, периодически очищать устаревшие данные и правильно настроить параметры кэширования, тогда он действительно будет полезен и поможет сделать систему быстрее и умнее.

А приходилось ли вам использовать векторный кэш или как-нибудь оптимизировать запросы в RAG пайплайнах? Делитесь в комментариях!

Полезные материалы

ссылка на оригинал статьи https://habr.com/ru/articles/930788/