Введение
Сегодня чат-боты и интеллектуальные ассистенты широко применяются в различных сферах: поддержка клиентов, корпоративные системы, поисковые сервисы и во многих других. Для их разработки часто используют архитектуру 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 для более продвинутого и управляемого кэширования
Реализация семантической маршрутизации запросов
Плюсы и минусы векторного кэша
Плюсы
Снижение задержек (latency): при повторных или схожих запросах можно избежать заново выполнения векторного поиска
Экономия ресурсов: снижается нагрузка на векторную базу данных
Минусы
Увеличения объёма кэша: при большом числе уникальных запросов кэш быстро увеличивается в размерах, необходимо определить оптимальную политику вытеснения
Дополнительные вычисления при вставке: каждая новая запись требует генерации эмбеддинга, обновления индекса и сохранения данных
Потенциальная устарелость результатов: если кэш не обновляется, ответы могут стать неактуальными по сравнению с обновлённой внешней базой данных
Выводы
Векторный кэш — это отличное решение, если вы хотите, чтобы ваша система быстрее отвечала и не тратила лишние ресурсы на похожие запросы.
Главное — не забывать следить за размером кэша, периодически очищать устаревшие данные и правильно настроить параметры кэширования, тогда он действительно будет полезен и поможет сделать систему быстрее и умнее.
А приходилось ли вам использовать векторный кэш или как-нибудь оптимизировать запросы в RAG пайплайнах? Делитесь в комментариях!