Кэширование ответов LLM в Redis: не платить за повторные запросы

Классификация отзывов, извлечение полей из документов, перевод однотипных строк — в таких пайплайнах одни и те же запросы повторяются, и без кэша за каждый повтор приходится платить. Кэширование ответов LLM в Redis решает это на стороне вашего приложения. Но сначала стоит понять, что уже делает шлюз Zukko, чтобы не строить лишнее: ниже — серверный кэш, случаи, когда нужен свой, и рабочий код на redis-py.

Что кэширует сам Zukko

В Zukko есть кэш по содержимому запроса (подробно — в разделе лимиты ключа, идемпотентность и кэш↗):

• Если повторить запрос с той же моделью и тем же телом при temperature: 0, второй ответ отдаётся из кэша бесплатно.
• При другой temperature кэш включается заголовком X-Zukko-Cache: true, выключается в любом случае — X-Zukko-Cache: false.
• Попадание видно по заголовку ответа X-Zukko-Cache: hit.
• Новые модели Claude (claude-sonnet-5, claude-opus-5-5) могут отклонить запрос с temperature, отличной от 1, ошибкой 400. Поэтому для них temperature не передавайте, а кэш включайте явно заголовком X-Zukko-Cache: true.
• Кэш отдельный у каждого аккаунта, ошибки не кэшируются, потоковые и обычные ответы хранятся раздельно, срок хранения — 7 дней.

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

Отдельно от кэша работает заголовок Idempotency-Key: повтор запроса с тем же ключом возвращает сохранённый ответ и не списывает деньги второй раз (ответ хранится 24 часа). Это защита от двойной оплаты при ретраях, а не кэш.

Когда нужен свой кэш в Redis

Если ваши повторы — это точные копии запросов и 7 дней вам достаточно, серверного кэша хватит. Свой Redis имеет смысл, когда:

• Входы почти одинаковые. Пользователи присылают один и тот же текст с разными пробелами, регистром, переносами. Нормализуйте вход перед построением ключа — серверный кэш так не умеет.
• Нужен срок дольше 7 дней или, наоборот, точный контроль, когда ответ устаревает.
• Нужна инвалидация по версии промпта. Поменяли инструкцию — старые ответы должны перестать использоваться.
• Важна задержка. Ответ из локального Redis приходит без сетевого запроса к шлюзу.
• Вы кэшируете результат обработки, а не сырой ответ: например, уже разобранный JSON.

Что не стоит кэшировать: творческие тексты, где ожидается разнообразие, ответы, зависящие от времени или внешних данных, и шаги агентов, где контекст меняется на каждом ходе.

Как строить ключ кэша

В ключ должно входить всё, что влияет на ответ: модель, все сообщения (включая системное), параметры генерации. Плюс префикс с версией — его меняют, когда меняется промпт.

import hashlib
import json
import re

PREFIX = "llm:v1:"

def normalize(text: str) -> str:
    return re.sub(r"\s+", " ", text).strip()

def cache_key(model: str, messages: list, params: dict) -> str:
    norm = [{"role": m["role"], "content": normalize(m["content"])} for m in messages]
    payload = json.dumps(
        {"model": model, "messages": norm, "params": params},
        sort_keys=True, ensure_ascii=False,
    )
    return PREFIX + hashlib.sha256(payload.encode("utf-8")).hexdigest()

sort_keys=True делает сериализацию стабильной: словари с одинаковым содержимым дадут одинаковую строку. Нормализация здесь только схлопывает пробелы; если для вашей задачи регистр не важен, добавьте .lower(). В модель при этом уходит исходный текст — нормализуется только ключ.

Кэширование ответов LLM в Redis: код на redis-py

Установка: pip install redis openai. Пример использует OpenAI SDK с базовым адресом https://api.zukko.pro/v1.

import json
import redis
from openai import OpenAI

r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
client = OpenAI(base_url="https://api.zukko.pro/v1", api_key="fetch_ваш_ключ")

TTL_SECONDS = 30 * 24 * 3600

def ask(messages: list, model: str = "claude-sonnet-5", **params) -> dict:
    params.setdefault("max_tokens", 512)
    key = cache_key(model, messages, params)

    cached = r.get(key)

Что здесь важно:

• decode_responses=True — Redis возвращает строки, а не байты, и json.loads работает напрямую.
• r.set(key, value, ex=...) записывает значение сразу со сроком жизни в секундах; эквивалент — r.setex(key, ttl, value).
• temperature не передаётся: для claude-sonnet-5 это безопаснее. Для GPT и GLM можно добавить temperature=0 в вызов ask(...) — он попадёт и в запрос, и в ключ кэша.
• extra_headers={"X-Zukko-Cache": "true"} явно включает серверный кэш Zukko, а with_raw_response даёт доступ к заголовкам ответа: в поле server_cache окажется hit, если ответ пришёл из этого кэша.
• Счётчики llm:stats:hit и llm:stats:miss показывают долю попаданий.

Для Anthropic SDK подход тот же: ключ строится так же, вызов — client.messages.create, текст ответа — в resp.content[0].text, а базовый адрес указывается как https://api.zukko.pro без /v1.

Сброс кэша без KEYS

Команда KEYS блокирует Redis на время обхода всей базы, поэтому в продакшене для поиска ключей используйте SCAN — в redis-py это scan_iter:

def invalidate(prefix: str = PREFIX) -> int:
    deleted = 0
    for key in r.scan_iter(match=prefix + "*", count=1000):
        deleted += r.delete(key)
    return deleted

Чаще всего удалять ничего не нужно: поменяйте PREFIX на llm:v2:, и старые записи просто истекут по TTL. Удалить один ответ можно через r.delete(cache_key(model, messages, params)).

Наблюдение за кэшем

hits = int(r.get("llm:stats:hit") or 0)
misses = int(r.get("llm:stats:miss") or 0)
total = hits + misses
print(f"Попаданий: {hits} из {total}")
print("Память:", r.info("memory")["used_memory_human"])
print("Ключей в базе:", r.dbsize())

Если доля попаданий близка к нулю, свой кэш не окупает сложность — оставьте только серверный. Чтобы Redis не разрастался бесконечно, задайте в его конфигурации лимит памяти и политику вытеснения (например, allkeys-lru) — подробнее в документации Redis.

Если одного сервера Redis мало, в redis-py есть RedisCluster (from redis.cluster import RedisCluster), подключение — по host и port любого узла или списку ClusterNode; остальной код не меняется.

Сколько можно сэкономить

Пример расчёта для claude-sonnet-5 ($0.3 за 1M входных и выходных токенов). Пусть пайплайн классифицирует 10 000 отзывов, в среднем 100 входных и 50 выходных токенов на запрос:

• стоимость одного запроса: (100 × 0.3 + 50 × 0.3) / 1 000 000 = $0.000045;
• без кэша: 10 000 × 0.000045 = $0.45;
• если после нормализации 40% отзывов совпали с уже обработанными, платных запросов 6 000: 6 000 × 0.000045 = $0.27.

Часть из этих 40% (точные копии) бесплатно закрыл бы и серверный кэш Zukko. Свой Redis добавляет попадания для почти одинаковых входов, хранение дольше 7 дней и ответ без сетевого запроса. Реальную долю повторов посчитайте по своим логам — от неё зависит вся выгода.

Создать отдельный ключ для пайплайна с дневным лимитом можно на странице API-ключи↗, а пример настройки клиента есть в руководстве по Python↗.

Частые вопросы

Нужен ли Redis, если у Zukko уже есть кэш?

Не всегда. Для точных повторов серверного кэша достаточно (он включается сам при temperature: 0 или заголовком X-Zukko-Cache: true). Redis нужен, когда входы отличаются мелочами, нужен срок больше 7 дней, версия промпта или минимальная задержка.

Можно ли кэшировать ответы при temperature больше 0?

Технически да, но тогда вы сознательно отказываетесь от разнообразия: пользователь всегда получит тот же вариант. На сервере Zukko это включается заголовком X-Zukko-Cache: true, в своём Redis — просто запись ответа. Для моделей Claude, которым temperature лучше не передавать, заголовок — единственный способ включить серверный кэш.

Можно ли хранить кэш в словаре Python вместо Redis?

Для скрипта или тестов — да. Но такой кэш пропадает при перезапуске и не разделяется между несколькими процессами или серверами.

Кэшировать ли потоковые ответы?

В свой Redis удобнее класть собранный целиком ответ после окончания стрима. На стороне Zukko потоковые и обычные ответы кэшируются раздельно.