Управление токенами: Context Caching и оптимизация затрат

Введение: Экономика токенов в эпоху больших контекстов

Приветствую на уроке, который может сэкономить вашей компании тысячи долларов и существенно ускорить работу ваших приложений. Мы переходим к одной из самых важных тем в инженерии промптов и архитектуре LLM-приложений — управлению токенами и Context Caching.

Давайте будем честны: когда Gemini 1.5 Pro представила контекстное окно в 1-2 миллиона токенов, все мы испытали восторг. Возможность «скормить» модели целую книгу, огромную кодовую базу или историю переписки за год кажется магией. Но у этой магии есть цена. И речь не только о деньгах, но и о латентности (задержке).

Каждый раз, когда вы отправляете запрос модели, вы платите за input tokens (входящие токены). Если у вас есть системный промпт на 50 000 токенов (например, подробная документация продукта), и вы отправляете его с каждым сообщением пользователя, вы платите за эти 50 000 токенов снова и снова. Это неэффективно. Это медленно. Это дорого.

В этом уроке мы разберем архитектурный паттерн Context Caching (Кэширование контекста) в Gemini API. Это инструмент, который меняет правила игры, позволяя хранить статический контекст на стороне Google и ссылаться на него, вместо того чтобы передавать его заново.

Как работает Context Caching: Под капотом

Традиционная модель взаимодействия с LLM выглядит как Stateless (без сохранения состояния). Каждый запрос — это новая жизнь. Чтобы модель знала контекст, вы должны передать его целиком.

Context Caching превращает часть вашего взаимодействия в Stateful. Вот как это работает физически и экономически:

1. Загрузка: Вы отправляете большой объем данных (файлы, тексты, видео) в API Gemini один раз.
2. Кэширование: Google обрабатывает эти токены и сохраняет их эмбеддинги/внутреннее представление в высокоскоростной памяти (RAM/VRAM своих серверов).
3. Ключ: Вы получаете уникальный идентификатор (имя кэша).
4. Использование: В следующих запросах вы просто передаете этот идентификатор и новый промпт пользователя. Модель «вспоминает» загруженные данные мгновенно.

В чем выгода?
Вы перестаете платить полную стоимость за обработку входящих токенов (Input processing). Вместо этого появляется новая метрика — Cached Input Tokens (Кэшированные входящие токены), которые стоят значительно дешевле обычных. Однако, добавляется стоимость хранения (Storage cost) за час существования кэша.

Эмпирическое правило: Кэширование в Gemini становится выгодным, если ваш контекст превышает ~32,000 токенов, и вы планируете обратиться к нему десятки раз.

# Пример базовой настройки окружения для работы с кэшированием
import os
import google.generativeai as genai
from google.generativeai import caching
import datetime

# Настройка API ключа
# Убедитесь, что ваш ключ имеет доступ к бета-функциям, если вы используете ранние версии
os.environ["GOOGLE_API_KEY"] = "ВАШ_КЛЮЧ_GEMINI"
genai.configure(api_key=os.environ["GOOGLE_API_KEY"])

# Проверяем доступные модели, поддерживающие кэширование
# Обычно это модели версии 1.5 (Pro и Flash)
print("Готовность к работе с Gemini 1.5 Pro")

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

Процесс создания кэша отличается от обычной генерации. Мы не вызываем generate_content сразу. Мы создаем объект кэша. Важно понимать понятие TTL (Time To Live). Кэш не вечен. Вы должны явно указать, сколько он должен жить. По умолчанию это может быть 1 час, но для продакшн-систем вы будете настраивать это время под рабочую сессию агента.

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

# Загрузка файла в File API (предварительный этап)
# Предположим, у нас есть файл 'gemini_api_docs.pdf'
doc_file = genai.upload_file(path='gemini_api_docs.pdf')

# Ждем завершения обработки файла (важно для PDF/Video)
import time
while doc_file.state.name == "PROCESSING":
    print('.', end='')
    time.sleep(2)
    doc_file = genai.get_file(doc_file.name)

print(f"\nФайл обработан: {doc_file.uri}")

# СОЗДАНИЕ КЭША
# Мы определяем 'system_instruction' внутри кэша, чтобы не передавать её каждый раз
cache = caching.CachedContent.create(
    model='models/gemini-1.5-pro-001',
    display_name='tech_support_v1', # Удобное имя для отладки
    system_instruction=(
        "Ты - эксперт по Gemini API. "
        "Используй предоставленную документацию для ответов на вопросы."
    ),
    contents=[doc_file], # Сюда передаем наш тяжелый контекст
    ttl=datetime.timedelta(minutes=60) # Кэш будет жить 60 минут
)

print(f"Кэш создан. Имя ресурса: {cache.name}")

Жизненный цикл и оптимизация затрат

Давайте поговорим о деньгах и архитектуре. Кэширование — это не всегда кнопка «сделать дешевле». Это инструмент балансировки.

Структура затрат (Примерная логика):

• Создание кэша: Вы платите полную стоимость input-токенов один раз (как за обычный запрос).
• Хранение: Вы платите почасовую ставку за хранение гигабайтов/токенов (сравнимо с арендой сервера).
• Использование: При запросе к кэшу вы платите значительно меньше (скидка может достигать 90% и более по сравнению с обычным input), так как модели не нужно заново «читать» текст.

Когда НЕ стоит использовать кэширование:

1. Короткий контекст: Если ваш промпт меньше 32к токенов, накладные расходы на создание кэша могут не окупиться.
2. Одноразовые запросы: Если вы загружаете книгу, задаете 1 вопрос и уходите — кэширование убыточно. Вы заплатите за создание + минимум за час хранения (зависит от тарификации), но не получите выгоды от дешевых запросов.

Паттерн «Общий ресурс»

Идеальный сценарий использования — это Retrieval Augmented Generation (RAG) без векторной базы данных для документов среднего размера, или долгоживущие сессии анализа кода. Вы создаете кэш с кодовой базой утром, и вся команда разработчиков делает к нему запросы в течение дня, обновляя TTL.

# ИСПОЛЬЗОВАНИЕ КЭША
# Обратите внимание: мы инициализируем модель, указывая cached_content

model = genai.GenerativeModel.from_cached_content(cached_content=cache)

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

print(f"Ответ модели: {response.text}")
print(f"Использовано токенов (usage_metadata): {response.usage_metadata}")

# В метаданных вы увидите разделение на cached_content_token_count
# Это подтверждение того, что вы платите по сниженному тарифу.

Управление состоянием: Обновление и Удаление

Кэш в Gemini неизменяем (immutable) по своему содержанию. Если вы нашли опечатку в загруженном PDF, вы не можете «подправить» кэш. Вам нужно удалить старый и создать новый. Однако, вы можете и должны управлять временем жизни (TTL).

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

Важное замечание по безопасности: Кэшированный контент привязан к вашему проекту. Убедитесь, что вы не кэшируете конфиденциальные данные (PII), если к этому кэшу будут иметь доступ пользователи с разными уровнями доступа, так как модель «знает» всё, что в кэше.

# УПРАВЛЕНИЕ ЖИЗНЕННЫМ ЦИКЛОМ

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

# 2. Получение информации о существующем кэше (по имени)
# Полезно, если кэш создал один сервис, а использует другой
cache_name = cache.name # строка вида 'cachedContents/12345...'
restored_cache = caching.CachedContent.get(cache_name)

# 3. Явное удаление
# ВСЕГДА удаляйте кэш, если он больше не нужен, чтобы не платить за хранение
# (хотя TTL удалит его автоматически, ручное удаление экономичнее)
restored_cache.delete()
print("Кэш удален.")

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

# Предполагаем, что api_key уже настроен
# Создадим фиктивный большой текст для примера (в реальности - загрузка файла)
fake_large_text = "Длинный текст... " * 10000 

def run_optimization_task():
    print("1. Создание кэша...")
    
    # Создаем кэш
    cache = caching.CachedContent.create(
        model='models/gemini-1.5-flash-001', # Flash дешевле для тестов
        display_name='war_and_peace_test',
        system_instruction="Отвечай кратко и по сути.",
        contents=[fake_large_text],
        ttl=datetime.timedelta(minutes=30)
    )
    
    print(f"Кэш создан: {cache.name}")

    # Подключаем модель к кэшу
    model = genai.GenerativeModel.from_cached_content(cached_content=cache)

    questions = [
        "О чем этот текст в двух словах?",
        "Какова главная идея?",
        "Назови ключевых персонажей."
    ]

    try:
        for i, q in enumerate(questions):
            print(f"\nВопрос {i+1}: {q}")
            response = model.generate_content(q)
            print(f"Ответ: {response.text}")
            
            # Анализ токенов
            usage = response.usage_metadata
            print(f"METRICS: Cached: {usage.cached_content_token_count}, Input: {usage.prompt_token_count}")
            
    except Exception as e:
        print(f"Ошибка: {e}")
    finally:
        # Удаление кэша в блоке finally гарантирует, что мы не будем платить лишнее при ошибках
        print("\n4. Удаление кэша...")
        cache.delete()
        print("Готово.")

# run_optimization_task() # Раскомментировать для запуска