Context Caching: Экономия ресурсов при повторяющихся запросах

Введение: Проблема "Забывчивого" API

Добро пожаловать в четвертый модуль курса! Сегодня мы погружаемся в тему, которая отделяет просто работающие решения от экономически эффективных и быстрых архитектур. Речь пойдет о Context Caching (Кэшировании контекста) в Gemini API.

Давайте начнем с аналогии. Представьте, что вы нанимаете консультанта для анализа огромного юридического договора на 500 страниц.

• Без кэширования (стандартный подход): Каждый раз, когда вы задаете вопрос («Какие риски в пункте 3?», «Кто платит штрафы?»), вы вынуждены заново распечатывать все 500 страниц, передавать их консультанту, ждать, пока он их прочитает, и только потом получать ответ. Это долго, дорого (бумага, время) и нелепо.
• С кэшированием: Вы передаете договор консультанту один раз. Он кладет его на стол. Теперь вы просто задаете вопросы, а он сразу отвечает, глядя в лежащий перед ним документ.

В мире LLM «распечатка договора» — это передача токенов. Передавать одни и те же огромные файлы (книги, базы кода, видео) в каждом запросе — значит сжигать бюджет и увеличивать задержку (latency). Gemini решает эту проблему, позволяя вам «сохранить» контекст на серверах Google и ссылаться на него.

Экономика и Техника: Как это работает под капотом

Context Caching в Gemini — это не просто «память». Это специализированный механизм для работы с большим контекстом (long context window).

Когда вы отправляете запрос в LLM, процесс делится на этапы:

1. Обработка ввода (Input Processing): Модель токенизирует и «понимает» ваш текст/видео/аудио. Для огромных объемов данных это требует значительных вычислительных мощностей.
2. Генерация (Output Generation): Создание ответа.

В чем выгода?
Кэширование позволяет выполнить первый этап один раз. Результат обработки (эмбеддинги, внутренние состояния модели) сохраняется. При следующих запросах модель пропускает тяжелую фазу обработки входных данных.

Структура цены (Cost Benefit):
Google обычно тарифицирует это так (цифры могут меняться, важен принцип):

• Создание кэша: Стандартная цена за входные токены.
• Хранение кэша: Почасовая оплата за объем (как аренда места).
• Использование кэша: Огромная скидка на входные токены (так как они уже обработаны).

# Подготовка окружения
import os
import time
import google.generativeai as genai
from google.generativeai import caching
import datetime

# Настройка API ключа
os.environ["GOOGLE_API_KEY"] = "ВАШ_КЛЮЧ"
genai.configure(api_key=os.environ["GOOGLE_API_KEY"])

# Представим, что у нас есть большой файл (например, документация или книга)
# Для примера загрузим файл в File API (предварительный шаг)
# path_to_large_file = "manual.pdf" 
# file_ref = genai.upload_file(path_to_large_file)

# В реальном коде вы должны дождаться завершения обработки файла:
# while file_ref.state.name == "PROCESSING":
#     time.sleep(2)
#     file_ref = genai.get_file(file_ref.name)

print("Окружение настроено и готово к созданию кэша.")

Жизненный цикл Кэша (TTL и Expiration)

Кэшированный контент в Gemini не вечен. Это временный ресурс, и управление его жизненным циклом — задача архитектора.

Ключевое понятие здесь — TTL (Time To Live).

• По умолчанию: Если вы не укажете иное, кэш часто живет короткое время (например, 1 час), после чего автоматически удаляется, чтобы вы не платили за забытые данные.
• Продление: Вы можете обновлять TTL, если контекст все еще нужен.
• Явное удаление: Хороший тон — удалять кэш сразу, как только он перестал быть нужен, используя метод delete.

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

# Создание кэшированного контента

# Мы используем класс caching.CachedContent
cache = caching.CachedContent.create(
    model='models/gemini-1.5-flash-001', # Указываем модель, совместимую с кэшированием
    display_name='tech_support_manual_v1', # Удобное имя для нас
    system_instruction='Ты эксперт техподдержки. Отвечай строго по инструкции.',
    contents=[file_ref], # Ссылка на ранее загруженный файл (см. предыдущий блок)
    ttl=datetime.timedelta(minutes=15) # Кэш будет жить 15 минут
)

print(f"Кэш создан: {cache.name}")
print(f"Истекает: {cache.expire_time}")

# ВАЖНО: cache.name - это уникальный идентификатор ресурса на сервере Google.
# Он выглядит примерно так: 'cachedContents/1234567890'

Использование Кэша в запросах

Теперь, когда «книга лежит на столе консультанта», нам нужно объяснить SDK, что при обращении к модели нужно использовать именно этот подготовленный контекст.

В Python SDK это делается элегантно: мы инициализируем объект модели не просто по имени (строке), а используя метод from_cached_content. Это «привязывает» экземпляр модели к конкретному хранилищу знаний.

Обратите внимание: кэш привязан к конкретной версии модели. Вы не можете создать кэш для Gemini 1.5 Pro и использовать его с Gemini 1.5 Flash.

# Инициализация модели с использованием кэша
model = genai.GenerativeModel.from_cached_content(cached_content=cache)

# Теперь все запросы к 'model' автоматически видят контекст из кэша
response = model.generate_content("Как сбросить настройки устройства до заводских?")

print("Ответ модели:")
print(response.text)

# Можно проверить использование токенов
# Вы увидите, что cached_content_token_count > 0, а обычных input tokens мало.
print(f"Использовано токенов кэша: {response.usage_metadata.cached_content_token_count}")

Управление и Обновление

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

Как получить доступ к уже существующему кэшу? По его имени.

Также важно уметь управлять временем жизни. Если диалог с пользователем затянулся, или если ваша база знаний актуальна весь день, вам нужно обновлять TTL.

# Сценарий: Другой сервис подключается к существующему кэшу
cache_name = cache.name # В реальности вы бы взяли это из базы данных

# Получение объекта кэша по имени
existing_cache = caching.CachedContent.get(cache_name)

# Продление жизни кэша еще на 2 часа
existing_cache.update(ttl=datetime.timedelta(hours=2))
print(f"Новое время истечения: {existing_cache.expire_time}")

# Удаление кэша (когда рабочий день закончен или вышли новые инструкции)
existing_cache.delete()
print("Кэш удален.")

Архитектурные паттерны использования

Когда стоит внедрять Context Caching? Давайте разберем конкретные сценарии.

• Чат-бот по большой документации: У вас есть PDF на 1000 страниц. Пользователь задает 10 вопросов подряд. Без кэша вы платите за 1000 страниц × 10 раз. С кэшем — за 1000 страниц 1 раз + плата за хранение (копейки) + 10 маленьких промптов. Экономия колоссальная.
• Анализ видео: Загрузили часовой фильм. Спрашиваете: «Найди момент с погоней», затем «Кто был за рулем?». Видео тяжелое, кэширование здесь критично для скорости.
• Few-shot Prompting с большой базой примеров: Вы хотите обучить модель стилю ответов, предоставив 500 идеальных примеров диалогов. Это «статичный» контекст, который идеально кэшируется.
• Агенты с личностью: Если у вас сложная системная инструкция с описанием мира, правил и инвентаря персонажа, кэшируйте её.

Когда НЕ использовать:
Если контекст меняется при каждом запросе (например, вы каждый раз анализируете новую статью из новостей), кэширование бесполезно и даже вредно (тратится время на создание).

import google.generativeai as genai
from google.generativeai import caching
import datetime
import os

# 1. Имитация данных (в реальности это был бы upload_file)
large_policy_text = "ПРАВИЛА КОМПАНИИ... " * 1000  # Делаем текст длинным

# 2. Создание кэша
# Обратите внимание: contents может принимать и просто текст, не только файлы
cache = caching.CachedContent.create(
    model='models/gemini-1.5-flash-001',
    display_name='hr_policies',
    contents=[large_policy_text],
    ttl=datetime.timedelta(minutes=5)
)

# 3. Инициализация
model = genai.GenerativeModel.from_cached_content(cached_content=cache)

# 4. Запрос
response = model.generate_content("Сколько дней отпуска положено?")

print(response.text)
print("Cached tokens:", response.usage_metadata.cached_content_token_count)

# Очистка
cache.delete()