Контекстный рот в AI-агентах: когда скользящее окно не спасает

Ваш AI-агент забывает всё через 20 сообщений. Это нормально? Нет.

Представьте: клиентская поддержка через чат-бота. Пользователь описывает проблему в 5 сообщениях, вы уточняете детали, получаете лог-файлы, анализируете. А потом агент спрашивает: "Какую именно проблему вы хотите решить?" Он забыл всё. Это и есть контекстный рот - когда LLM теряет ключевую информацию из-за переполнения контекстного окна.

В 2026 году у нас есть модели с контекстом до 1 млн токенов (Claude 3.5 Sonnet, GPT-4o Long, Gemini 2.0 Flash Thinking). Но проблема не исчезла. Она стала тоньше. Даже с большим окном агент путается в деталях, теряет нить диалога, повторяет вопросы.

Почему скользящее окно и базовая суммаризация проваливаются

Вы пробовали классические методы? Скользящее окно сохраняет последние N сообщений. Суммаризация сжимает историю. В теории работает. На практике - катастрофа.

Сценарий: поддержка SaaS-продукта. Диалог 30 сообщений:

• Сообщения 1-5: описание проблемы с авторизацией
• Сообщения 6-10: обмен учетными данными
• Сообщения 11-15: анализ логов
• Сообщения 16-20: предложение временного решения
• Сообщения 21-25: обсуждение постоянного фикса
• Сообщения 26-30: планирование обновления

Скользящее окно (последние 10 сообщений) отрезает корень проблемы. Суммаризация теряет технические детали. Результат: агент предлагает сбросить пароль, когда проблема в OAuth-токенах.

1Иерархическая суммаризация: когда одного уровня недостаточно

Базовая суммаризация работает как пресс: всё сжимает в один абзац. Иерархическая - как архив с папками. Вот как это выглядит в коде (используем Claude 3.5 Sonnet API 2026):

import anthropic
from typing import List, Dict

class HierarchicalSummarizer:
    def __init__(self, api_key: str):
        self.client = anthropic.Anthropic(api_key=api_key)
        
    def create_chunk_summary(self, messages: List[Dict]) -> str:
        """Суммаризация чанка из 5-7 сообщений"""
        prompt = f"""Суммаризируй этот фрагмент диалога технической поддержки.
        Выдели:
        1. Основную тему
        2. Ключевые факты
        3. Открытые вопросы
        4. Решенные проблемы
        
        Диалог:
        {"\n".join([f'{m["role"]}: {m["content"]}' for m in messages])}
        """
        
        response = self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=500,
            messages=[{"role": "user", "content": prompt}]
        )
        return response.content[0].text
    
    def create_hierarchical_summary(self, chunk_summaries: List[str]) -> str:
        """Создание иерархической сводки из чанков"""
        hierarchy_prompt = f"""Ты получишь несколько суммаризаций фрагментов длинного диалога.
        Создай структурированную сводку всего диалога в формате:
        
        ## Основная тема
        [тема]
        
        ## Хронология событий
        - [событие 1] → [событие 2] → [событие 3]
        
        ## Текущий статус
        - Решено: [список]
        - В процессе: [список]
        - Ожидает: [список]
        
        ## Критические детали (не удалять!)
        - [техническая деталь 1]
        - [техническая деталь 2]
        
        Фрагменты:
        {"\n---\n".join(chunk_summaries)}
        """
        
        response = self.client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=800,
            messages=[{"role": "user", "content": hierarchy_prompt}]
        )
        return response.content[0].text

Ключевая фишка: разделение на уровни. Чанковая суммаризация сохраняет детали. Иерархическая - структуру. Вместе они дают то, что я называю "семантической картой диалога".

2Архитектура с выделенной памятью: stateful memory pattern

Проблема большинства агентов: они treat память как побочный эффект. Правильный подход - memory-first архитектура. Взгляните на stateful memory - это меняет всё.

Вот упрощенная реализация:

from dataclasses import dataclass
from typing import Any, List, Optional
from datetime import datetime
import json

@dataclass
class MemoryItem:
    id: str
    content: str
    category: str  # 'fact', 'action', 'decision', 'user_preference'
    importance: float  # 0.0 to 1.0
    timestamp: datetime
    metadata: dict
    
class StatefulMemory:
    def __init__(self):
        self.memories: List[MemoryItem] = []
        self.working_memory: List[str] = []  # Краткосрочная память
        
    def add_memory(self, content: str, category: str, importance: float = 0.5):
        """Добавление с автоматической категоризацией"""
        memory = MemoryItem(
            id=f"mem_{len(self.memories)}",
            content=content,
            category=category,
            importance=importance,
            timestamp=datetime.now(),
            metadata={"access_count": 0}
        )
        self.memories.append(memory)
        
    def retrieve_relevant(self, query: str, top_k: int = 5) -> List[MemoryItem]:
        """Извлечение релевантных воспоминаний"""
        # Здесь должна быть семантическая поисковая система
        # Для простоты - фильтрация по категориям
        relevant = []
        query_lower = query.lower()
        
        for memory in sorted(self.memories, key=lambda x: -x.importance):
            if query_lower in memory.content.lower():
                relevant.append(memory)
            if memory.category in query_lower:
                relevant.append(memory)
            if len(relevant) >= top_k:
                break
                
        # Обновляем метаданные доступа
        for memory in relevant:
            memory.metadata["access_count"] += 1
            
        return relevant
    
    def compress_memories(self) -> str:
        """Сжатие старых воспоминаний в суммаризацию"""
        old_memories = [m for m in self.memories 
                       if (datetime.now() - m.timestamp).days > 1]
        
        if len(old_memories) < 3:
            return ""
            
        # Группируем по категориям
        memories_by_category = {}
        for memory in old_memories:
            if memory.category not in memories_by_category:
                memories_by_category[memory.category] = []
            memories_by_category[memory.category].append(memory.content)
        
        # Создаем структурированную суммаризацию
        summary_parts = []
        for category, contents in memories_by_category.items():
            if len(contents) > 2:  # Только если достаточно контента
                summary = f"{category}: {", ".join(contents[:3])}..."
                summary_parts.append(summary)
                
        # Удаляем сжатые воспоминания
        self.memories = [m for m in self.memories 
                        if (datetime.now() - m.timestamp).days <= 1]
        
        return "\n".join(summary_parts)

Эта архитектура решает главную проблему: смешение важной и неважной информации. Технические детали авторизации хранятся отдельно от предпочтений пользователя. Система знает, что "токен OAuth истек" важнее, чем "пользователь любит эмодзи".

3Динамическое управление контекстом: что отправлять в каждый запрос

Самая частая ошибка: отправлять всю историю в каждый запрос. Умные агенты 2026 года используют динамический контекст.

Алгоритм:

1. Определить тип текущего запроса (уточнение, новое действие, продолжение)
2. Извлечь релевантные воспоминания из stateful memory
3. Добавить критический контекст (последние 2-3 сообщения)
4. При необходимости добавить суммаризацию
5. Рассчитать приоритеты: что должно быть ближе к промпту

Реализация:

class DynamicContextManager:
    def __init__(self, memory: StatefulMemory):
        self.memory = memory
        self.conversation_history = []
        
    def build_context(self, user_message: str, message_type: str) -> str:
        """Динамическое построение контекста"""
        context_parts = []
        
        # 1. Всегда добавляем системный промпт
        system_prompt = """Ты - AI-агент технической поддержки. 
        У тебя есть доступ к истории диалога и памяти.
        Отвечай точно, учитывая контекст."""
        context_parts.append(f"System: {system_prompt}")
        
        # 2. Извлекаем релевантные воспоминания
        relevant_memories = self.memory.retrieve_relevant(user_message)
        if relevant_memories:
            memory_text = "\n".join([f"Memory [{m.category}]: {m.content}" 
                                     for m in relevant_memories])
            context_parts.append(f"Relevant Memories:\n{memory_text}")
        
        # 3. Добавляем историю в зависимости от типа сообщения
        if message_type == "clarification":
            # Для уточнений - только последние 5 сообщений
            recent = self.conversation_history[-5:] if len(self.conversation_history) > 5 \
                     else self.conversation_history
            context_parts.append(f"Recent conversation:\n{recent}")
        elif message_type == "new_action":
            # Для новых действий - суммаризация всей истории
            summary = self.create_history_summary()
            context_parts.append(f"Conversation summary:\n{summary}")
        else:
            # По умолчанию - последние 10 сообщений
            recent = self.conversation_history[-10:] if len(self.conversation_history) > 10 \
                     else self.conversation_history
            context_parts.append(f"Conversation history:\n{recent}")
        
        # 4. Добавляем текущее сообщение
        context_parts.append(f"User: {user_message}")
        
        return "\n\n".join(context_parts)
    
    def create_history_summary(self) -> str:
        """Создание суммаризации истории"""
        if len(self.conversation_history) < 8:
            return "\n".join(self.conversation_history)
            
        # Используем иерархическую суммаризацию
        chunks = [self.conversation_history[i:i+5] 
                 for i in range(0, len(self.conversation_history), 5)]
        
        # Здесь можно подключить LLM для суммаризации
        # Для примера - простая версия
        summary = f"Диалог из {len(self.conversation_history)} сообщений. "
        summary += f"Основные темы: {', '.join(set(self.extract_topics()))}"
        
        return summary
    
    def extract_topics(self) -> List[str]:
        """Извлечение тем из истории (упрощенно)"""
        topics = []
        keywords = ["error", "login", "payment", "bug", "feature", "help"]
        
        for message in self.conversation_history:
            for keyword in keywords:
                if keyword in message.lower():
                    topics.append(keyword)
                    
        return list(set(topics))

Этот подход экономит токены и улучшает качество. Агент не тратит контекст на нерелевантную историю.

Fine-tuning для устойчивости к контекстному роту

Архитектурные паттерны - это половина решения. Другая половина - обучение модели. В 2026 году fine-tuning для устойчивости к длинным контекстам стал стандартом.

Что нужно учить:

• Извлекать ключевую информацию из начала диалога
• Игнорировать повторяющиеся детали
• Поддерживать ссылки на ранее упомянутые сущности
• Распознавать, когда контекста недостаточно

Пример датасета для fine-tuning:

{
  "messages": [
    {
      "role": "user",
      "content": "У меня проблема с авторизацией в приложении. Токен не работает."
    },
    {
      "role": "assistant",
      "content": "Расскажите подробнее. Какой тип авторизации используете?"
    },
    // ... 20 сообщений диалога ...
    {
      "role": "user",
      "content": "И что теперь делать?"
    },
    {
      "role": "assistant",
      "content": "На основе нашего обсуждения: ваш OAuth-токен истек. Нужно обновить его в настройках приложения. Это решает проблему авторизации, которую вы описали в начале."
    }
  ],
  "metadata": {
    "retains_initial_context": true,
    "references_early_details": true,
    "handles_context_gap": false
  }
}

Интеграция с внешними системами: когда памяти всё равно мало

Иногда никакая архитектура не спасает. Диалог длится неделю, обсуждается 15 разных проблем. Здесь нужна интеграция с внешними системами.

Варианты:

1. Векторная база диалогов: каждый диалог - отдельный документ с embedding. Похожие диалоги извлекаются как reference.
2. Граф знаний: сущности и отношения из диалога сохраняются в графовой БД (Neo4j, Amazon Neptune).
3. Внешняя память через API: как в статье про контекст-инжиниринг.

Простая интеграция с векторной БД:

import chromadb
from sentence_transformers import SentenceTransformer

class VectorMemory:
    def __init__(self):
        self.client = chromadb.Client()
        self.collection = self.client.create_collection("dialog_memory")
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
        
    def store_dialog_chunk(self, dialog_id: str, messages: List[str], 
                          metadata: dict):
        """Сохранение чанка диалога"""
        text = "\n".join(messages)
        embedding = self.encoder.encode(text).tolist()
        
        self.collection.add(
            documents=[text],
            embeddings=[embedding],
            metadatas=[metadata],
            ids=[f"{dialog_id}_{len(messages)}"]
        )
        
    def find_similar_context(self, query: str, dialog_id: str, 
                           top_k: int = 3) -> List[str]:
        """Поиск похожих контекстов в истории"""
        query_embedding = self.encoder.encode(query).tolist()
        
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k,
            where={"dialog_id": {"$ne": dialog_id}}  # Исключаем текущий диалог
        )
        
        return results['documents'][0] if results['documents'] else []

Мониторинг и диагностика: как понять, что агент забывает

Контекстный рот коварен. Агент может казаться адекватным, но постепенно деградировать. Нужны метрики.

Инструмент для мониторинга:

class ContextRotMonitor:
    def __init__(self):
        self.metrics = {
            'consistency': [],
            'repetitions': [],
            'context_usage': []
        }
    
    def check_consistency(self, dialog_history: List[str]) -> float:
        """Проверка согласованности фактов"""
        if len(dialog_history) < 4:
            return 1.0
            
        # Извлекаем факты из первых и последних сообщений
        early_facts = self.extract_facts(dialog_history[:2])
        late_facts = self.extract_facts(dialog_history[-2:])
        
        # Сравниваем
        matches = sum(1 for fact in early_facts if fact in late_facts)
        total = max(len(early_facts), 1)
        
        return matches / total
    
    def detect_repetitions(self, current_response: str, 
                          history: List[str]) -> List[str]:
        """Обнаружение повторяющихся вопросов"""
        repetitions = []
        
        # Простой detection по ключевым фразам
        question_phrases = [
            "какая проблема", "расскажите подробнее",
            "что именно", "какие детали"
        ]
        
        for phrase in question_phrases:
            if phrase in current_response.lower():
                # Проверяем, задавался ли уже такой вопрос
                for past_message in history[-5:]:
                    if phrase in past_message.lower():
                        repetitions.append(phrase)
                        break
                        
        return repetitions
    
    def log_metrics(self, dialog_id: str):
        """Логирование метрик для анализа"""
        avg_consistency = sum(self.metrics['consistency']) / \
                         max(len(self.metrics['consistency']), 1)
        
        if avg_consistency < 0.7:
            print(f"ВНИМАНИЕ: Контекстный рот в диалоге {dialog_id}")
            print(f"Consistency score: {avg_consistency:.2f}")

Чего ждать в 2027: архитектурные сдвиги

Текущие методы работают, но они костыли. Настоящее решение требует изменений в архитектуре моделей.

Тренды:

• Differentiable memory: память как часть модели, обучаемая end-to-end
• Attention с приоритетами: механизм внимания, который учится сохранять важное
• Иерархические трансформеры: разные уровни для разных временных масштабов
• Нейрографическая память: хранение информации в графовых структурах внутри модели

Пока этого нет, используйте комбинацию: иерархическая суммаризация + stateful memory + динамический контекст. И не забывайте про agent skills - иногда проблема не в памяти, а в том, что агент просто не умеет работать с длинными диалогами.

Главный совет на 2026: перестаньте думать о контексте как о линейной последовательности токенов. Это семантическое пространство, где одни части важнее других. Учите агента различать важное и неважное. Иначе вы просто будете эффективно забывать.