ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Настроить prompt caching

1. Цель задачи

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

Ключевой результат Скрипт/библиотека, измерение latency до/после, ускорение ≥10x на наборе тестовых запросов.

2. Исходные данные

Если нет реального API — симулируем:

1. Используем transformers + небольшую модель (например, t5-small) для токенизации и прямого прохода forward.
2. Вместо сетевого вызова делаем вызов модели локально — время замеряется.
3. Шаблоны: "Ответь на вопрос: {question}. Контекст: {context}".

3. Технологический стек

4. Этапы выполнения

Этап 1: Создание базового pipeline без кэша (1 час)

Действия

1. Написать класс PromptProcessor:

   • метод compile(template: str, **kwargs) -> str: подстановка переменных (через f‑strings или Jinja2).
   • метод tokenize(prompt: str) -> list[int]: преобразование строки в список токенов (использовать tiktoken.encoding_for_model("gpt-4")).
   • метод send_to_llm(prompt: str) -> str: эмуляция / реальный call|вызов API (заглушка с time.sleep(0.1) для симуляции).

2. Написать тестовые шаблоны и 50 вариантов подстановок (например, вопросы + контексты).

3. Замерить время обработки всего набора без кэша — получить baseline.

import time
import tiktoken

class PromptProcessor:
    def __init__(self, model="gpt-4"):
        self.encoding = tiktoken.encoding_for_model(model)

    def compile(self, template, **kwargs):
        return template.format(**kwargs)

    def tokenize(self, prompt):
        return self.encoding.encode(prompt)

    def send(self, prompt):
        # эмуляция LLM вызова
        time.sleep(0.1)
        return "OK"

Ожидаемый результат этапа Рабочий pipeline, измеренное baseline latency (например, 10.5 ms на запрос или 500 ms с учётом эмуляции).

Этап 2: Внедрение in-memory кэша (1 час)

Действия

1. Определить ключ кэша: хэш от (шаблон + сортированные kwargs). Использовать hashlib.sha256.
2. Использовать cachetools.LRUCache с maxsize=1000.
3. Модифицировать compile — проверять кэш до выполнения подстановки и токенизации.
4. Реализовать invalidate(template, **kwargs) для принудительной очистки.

from cachetools import LRUCache
import hashlib

class CachedPromptProcessor(PromptProcessor):
    def __init__(self, maxsize=1000):
        super().__init__()
        self.cache = LRUCache(maxsize=maxsize)

    def _make_key(self, template, kwargs):
        data = template + str(sorted(kwargs.items()))
        return hashlib.sha256(data.encode()).hexdigest()

    def compile_and_tokenize(self, template, **kwargs):
        key = self._make_key(template, kwargs)
        if key in self.cache:
            return self.cache[key]
        prompt = self.compile(template, **kwargs)
        tokens = self.tokenize(prompt)
        self.cache[key] = tokens
        return tokens

5. Запустить тот же набор запросов (50), замерить время. Обратить внимание: если все запросы уникальны, ускорения не будет. Для демонстрации нужно создать повторяющиеся комбинации (например, 10 уникальных шаблонов × 5 повторов = 50 запросов).

Ожидаемый результат этапа При повторных запросах с теми же аргументами latency падает до ~0 (практически только overhead кэша). Общее время выполнения набора с 10 уникальными шаблонами × 5 повторов снижается в 5-10 раз.

Этап 3: Интеграция с Redis (1.5 часа)

Действия

1. Развернуть Redis (через Docker: docker run -d -p 6379:6379 redis/redis-stack).
2. Написать класс RedisPromptCache с методами get(key), set(key, value, ttl=3600).
3. Сериализовать список токенов в JSON (или использовать pickle, но осторожно).
4. Подключить к CachedPromptProcessor — при compile_and_tokenize сначала проверять Redis, потом локальный LRU.
5. Написать тест на пропускную способность в многопоточном режиме (concurrent.futures) — измерить throughput.

import redis
import json

class RedisPromptCache:
    def __init__(self, host='localhost', port=6379, db=0, ttl=3600):
        self.client = redis.Redis(host=host, port=port, db=db)
        self.ttl = ttl

    def get(self, key):
        data = self.client.get(key)
        return json.loads(data) if data else None

    def set(self, key, value):
        self.client.setex(key, self.ttl, json.dumps(value))

# Использование в CachedPromptProcessor
self.redis_cache = RedisPromptCache()

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

Этап 4: Измерение производительности и подготовка отчёта (1 час)

Действия

1. Подготовить бенчмарк-скрипт benchmark.py:
   • Генерирует N запросов (например, 1000), из которых 80% повторяются (20 уникальных).
   • Запускает последовательно без кэша, с in-memory кэшем, с Redis кэшем.
   • Выводит метрики: общее время, среднее latency, median, P95, P99, hit rate.
2. Убедиться, что ускорение ≥10x на тесте с повторными запросами.
3. Написать одностраничный отчёт в Markdown с графиками (например, bar chart latency до/после).

Ожидаемый результат этапа Файл benchmark_results.md с таблицами и выводом.

Этап 5: Обработка инвалидации и edge-кейсов (1 час)

Действия

1. Реализовать механизм принудительной инвалидации: если меняется сам шаблон (новая версия), все ключи с этим шаблоном должны быть сброшены.
   • Хранить в кэше дополнительно версию шаблона.
2. Обработать случай, когда подстановки содержат большие объёмы (контекст до 32k токенов) — кэшировать не весь промпт, а только результат токенизации (список ID).
3. Добавить логирование hit/miss для мониторинга.

def compile_and_tokenize(self, template, template_version, **kwargs):
    key_data = template_version + template + str(sorted(kwargs.items()))
    key = hashlib.sha256(key_data.encode()).hexdigest()
    ...

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

5. Критерии приемки (Definition of Done)

• Реализован класс CachedPromptProcessor с LRU-кэшем.
• Реализована интеграция с Redis.
• Бенчмарк показывает ускорение ≥10x на данных с повторными запросами (80% повторений).
• Время ответа с кэшем (in-memory) < 1 ms.
• Кэширование не ломается при передаче аргументов разных типов (int, str, None).
• Есть механизм инвалидации по версии шаблона.
• В коде есть логирование hit/miss с уровнем DEBUG.
• Написаны юнит-тесты для ключевых функций (создание ключа, кэширование, инвалидация).
• Документация README.md с инструкцией по запуску и примером.
• Отчёт benchmark_results.md с таблицей latency до/после.

6. Ожидаемый результат

Основной артефакт Репозиторий (папка) со следующими файлами:

• prompt_cache.py — реализация кэша (in-memory + Redis).
• benchmark.py — скрипт для замера производительности.
• benchmark_results.md — отчёт с результатами.
• requirements.txt — зависимости (tiktoken, redis, cachetools, jinja2).
• test_prompt_cache.py — юнит-тесты (pytest).

Содержание отчёта (benchmark_results.md):

Дополнительный результат Навык проектирования кэша для LLM‑пайплайнов, понимание метрик hit rate и latency.

7. Возможные сложности и их решение

8. Бюджет времени (оценка)

Примечание Для первого раза рекомендуется выделить 6-7 часов с учётом настройки окружения и отладки.

9. Связанные вопросы из базы знаний

10. Чек-лист самопроверки

• Я реализовал in-memory кэш и проверил, что повторные запросы обрабатываются мгновенно.
• Я настроил Redis локально и успешно протестировал set/get.
• Я выполнил бенчмарк на наборе с 80% повторений и получил ускорение ≥10x.
• Я написал хотя бы 3 юнит-теста (создание ключа, hit/miss, инвалидация).
• Я задокументировал в README, как запустить бенчмарк и интерпретировать результаты.