Стратегии управления контекстным окном в диалоговых системах

Введение: Иллюзия бесконечной памяти

Приветствую, коллеги. Сегодня мы погружаемся в одну из самых архитектурно сложных тем при создании диалоговых систем — управление контекстным окном.

С выходом моделей семейства Gemini, обладающих контекстным окном в 1-2 миллиона токенов, у многих разработчиков возник соблазн отказаться от стратегий управления памятью. Казалось бы: «Зачем мне что-то оптимизировать, если я могу просто отправлять всю историю переписки с момента сотворения мира?»

Это опасная иллюзия. И вот почему:

1. Стоимость (Cost): Даже с дешевыми моделями, отправка мегабайтов текста в каждом запросе быстро истощит бюджет Enterprise-проекта.
2. Задержка (Latency): Чем больше контекст на входе (Input), тем дольше модель его обрабатывает (Time to First Token). Для чат-бота ожидание в 10 секунд недопустимо.
3. Фокус внимания (Attention Drift): Модели, даже самые мощные, могут «теряться» в огромных массивах неструктурированных данных, уделяя внимание нерелевантным деталям из начала беседы.

В этом уроке мы разберем, как эволюционировали методы памяти и как правильно использовать возможности Gemini 3, включая Context Caching, для построения эффективных систем.

Стратегия 1: Скользящее окно (Sliding Window)

Это классический метод, пришедший к нам из эпохи малых контекстных окон (4k-8k токенов). Суть проста: мы храним только $N$ последних сообщений или последние $K$ токенов. Как только лимит превышен, самые старые сообщения «выпадают» из памяти.

Принцип работы FIFO

Работает по принципу очереди (First-In, First-Out). Это идеальное решение для сценариев, где важна непосредственная нить разговора, а детали десятиминутной давности теряют актуальность (например, уточнение параметров заказа).

Преимущества:
✅ Предсказуемое потребление токенов.
✅ Низкая задержка.
✅ Простота реализации.

Недостатки:
❌ Полная потеря контекста при длительных беседах (амнезия).
❌ Если пользователь ссылается на что-то, сказанное в начале, бот его не поймет.

from collections import deque
from typing import List, Dict

class SlidingWindowMemory:
    def __init__(self, max_messages: int = 10):
        # Используем deque для эффективного добавления и удаления
        self.history = deque(maxlen=max_messages)

    def add_message(self, role: str, content: str):
        self.history.append({"role": role, "content": content})

    def get_context(self) -> List[Dict[str, str]]:
        # Возвращаем список для отправки в API
        return list(self.history)

# Пример использования
memory = SlidingWindowMemory(max_messages=4)
memory.add_message("user", "Привет, меня зовут Алекс.")
memory.add_message("model", "Привет, Алекс! Чем могу помочь?")
memory.add_message("user", "Какая погода в Лондоне?")
memory.add_message("model", "В Лондоне сейчас дождь.")

# Если добавить еще одно сообщение, первое ("Привет, меня зовут Алекс") исчезнет
memory.add_message("user", "А в Париже?")

print(memory.get_context())
# Результат не содержит имени Алекса, модель "забыла" его.

Стратегия 2: Суммаризация и "Сводная память"

Чтобы решить проблему амнезии метода «Скользящего окна», мы применяем технику суммаризации. Идея заключается в том, чтобы периодически сжимать старую часть диалога в краткое резюме (summary) и подавать его как «системный контекст» для следующих шагов.

Как это работает в Gemini

Мы разделяем промпт на три части:

1. System Instruction: Инструкции по поведению + Текущее резюме разговора.
2. Short-term History: Несколько последних сообщений в сыром виде (для поддержания живого стиля общения).
3. New User Query: Новый вопрос.

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

import google.generativeai as genai

# Псевдокод логики суммаризации

def update_summary(current_summary, recent_history, model):
    prompt = f"""
    Текущее резюме разговора: {current_summary}
    
    Недавний диалог:
    {recent_history}
    
    Обнови резюме, включив новые ключевые детали из недавнего диалога.
    Сохраняй краткость. Не теряй имена и цифры.
    """
    response = model.generate_content(prompt)
    return response.text

# Пример структуры промпта при запросе
final_prompt = [
    {"role": "user", "parts": [f"Системная память: {summary_text}"]},
    # ... последние 5 сообщений ...
    {"role": "user", "parts": ["Новый вопрос пользователя"]}
]

Стратегия 3: Context Caching в Gemini (Enterprise уровень)

Это «киллер-фича» современных моделей Google. Если ваш бот должен знать содержание огромной технической документации, свода законов или истории проекта (например, 500,000 токенов), отправлять это каждый раз — безумие.

Context Caching позволяет загрузить этот контекст один раз, получить специальный ID (токен кэша) и ссылаться на него в последующих запросах. Кэш живет определенное время (TTL).

Когда использовать?

• Анализ книг/документов: Пользователь задает много вопросов по одному PDF файлу.
• Ролевые боты с длинной легендой: Глубокое описание мира и правил.
• Длительные сессии обучения: Где контекст накапливается часами.

Это снижает стоимость (кэшированные токены дешевле) и радикально снижает задержку, так как модели не нужно заново «переваривать» префикс.

Стратегия 4: Гибридная память (Архитектура Мозга)

Самые продвинутые системы не полагаются на один метод. Они имитируют человеческую память, разделяя её на типы:

1. Сенсорная память (Raw Buffer): Последние 2-3 сообщения. Точность 100%.
2. Семантическая память (RAG): База знаний (Vector DB), откуда мы подтягиваем факты, релевантные текущему вопросу.
3. Эпизодическая память (Summaries): Сжатая история того, что происходило в этой конкретной сессии.

При поступлении запроса, оркестратор формирует контекст, собирая данные из всех трех источников. Это позволяет вести диалоги, которые длятся днями, не выходя за пределы разумного окна токенов и сохраняя точность.

python
class ConversationManager:
    def __init__(self):
        self.buffer = [] # Список словарей
        self.buffer_size = 4
        self.long_term_memory = "Start of conversation."

    def add_turn(self, user_text, model_text):
        # Добавляем новые сообщения
        self.buffer.append({"role": "user", "content": user_text})
        self.buffer.append({"role": "model", "content": model_text})

        # Проверка переполнения
        self._manage_buffer()

    def _manage_buffer(self):
        while len(self.buffer) > self.buffer_size:
            # Берем самую старую пару (user + model)
            old_user = self.buffer.pop(0)
            old_model = self.buffer.pop(0)
            
            # 'Архивируем' в долгосрочную память
            # В реальности здесь был бы вызов LLM для суммаризации
            archive_text = f"\n[Archived] U: {old_user['content']} | M: {old_model['content']}"
            self.long_term_memory += archive_text
            print(f"LOG: Archived messages to LTM.")

    def get_full_context(self):
        return {
            "system_context": self.long_term_memory,
            "active_dialogue": self.buffer
        }

# Тестирование
mgr = ConversationManager()
mgr.add_turn("1. Привет", "1. Привет!")
mgr.add_turn("2. Как дела?", "2. Отлично.")
# Сейчас буфер полон (4 сообщения). Следующий шаг вызовет архивацию первых двух.
mgr.add_turn("3. Что такое API?", "3. Интерфейс программирования.")

ctx = mgr.get_full_context()
print("LTM:", ctx['system_context'])
print("Buffer len:", len(ctx['active_dialogue'])) # Должно быть 4