Архитектура контекстного движка для RAG: как управлять памятью, компрессией и токенами в диалоговых системах на Python

Почему ваш RAG-бот забывает разговор через 10 минут

Вы построили идеальную RAG-систему. Она прекрасно отвечает на первый вопрос. На второй - уже скучнеет. К пятому - начинает нести околесицу, а к десятому предлагает 'уточнить запрос', хотя вы обсуждаете одну тему. Знакомо? Это не баг, это фундаментальный провал в управлении контекстом. Стандартное скользящее окно в 8K токенов работает как амнезиак: новое входит, старое вылетает. В статье о контекстном роте мы уже разбирали, почему простые решения не работают.

Контекстный движок: что это и зачем он вам

Контекстный движок - это прослойка между пользователем и LLM, которая управляет тем, что именно попадет в промпт. Не просто 'последние N сообщений', а интеллектуальная система, которая решает, что помнить, что сжать, а что выкинуть навсегда. В отличие от встроенных функций 'памяти' в локальных LLM, которые часто оказываются черным ящиком (об этом наш разбор в статье AI Memory - это обман?), здесь вы получаете полный контроль.

1 Три кита архитектуры

Движок состоит из трех взаимосвязанных модулей. Если вырвать один - вся конструкция рухнет.

Memory Manager: не просто список сообщений

Первый соблазн - хранить все в векторах. Второй - положиться на базу данных. Оба подхода убивают производительность. Memory Manager работает с двумя типами памяти: оперативной (последние 5-10 обменов в сыром виде) и долговременной (ключевые факты, решения, контекст сессии). Для долговременной памяти мы используем граф знаний, а не плоский список. Почему? Потому что связи между фактами важнее самих фактов. Статья о системах долговременной памяти подробно разбирает паттерны.

# Пример ядра Memory Manager на Python (2026)
# Используем актуальные библиотеки для работы с графами
import networkx as nx  # версия 3.3+
from typing import List, Dict, Optional
from dataclasses import dataclass
from datetime import datetime

@dataclass
class MemoryNode:
    id: str
    content: str
    entity_type: str  # 'fact', 'decision', 'user_preference', 'context'
    created_at: datetime
    weight: float  # важность узла, вычисляется динамически

class MemoryManager:
    def __init__(self):
        self.graph = nx.DiGraph()
        self.short_term_buffer: List[Dict] = []  # оперативная память
        self.max_short_term = 10  # сообщений
        
    def add_interaction(self, user_msg: str, ai_response: str):
        # Сохраняем в оперативную память
        self.short_term_buffer.append({
            'user': user_msg,
            'ai': ai_response,
            'timestamp': datetime.now()
        })
        if len(self.short_term_buffer) > self.max_short_term:
            self._compress_oldest()  # сжимаем самое старое
            
        # Извлекаем сущности и факты для долговременной памяти
        facts = self._extract_facts(user_msg, ai_response)
        for fact in facts:
            self._add_to_graph(fact)
    
    def _compress_oldest(self):
        # Самое интересное: сжатие через LLM
        # Используем легкую модель для суммирования
        old_interaction = self.short_term_buffer.pop(0)
        compressed = self._summarize_interaction(old_interaction)
        # Превращаем сжатое взаимодействие в узел графа
        memory_node = MemoryNode(
            id=f"compressed_{datetime.now().timestamp()}",
            content=compressed,
            entity_type='compressed_interaction',
            created_at=datetime.now(),
            weight=0.7  # сжатые воспоминания имеют меньший вес
        )
        self.graph.add_node(memory_node.id, data=memory_node)

Compression Engine: как сжать в 10 раз без потери смысла

Сжатие - это не удаление 'стоп-слов'. Это переформулирование сути. Классическая ошибка - использовать ту же LLM, что и для генерации ответов. Она дорогая и медленная. Compression Engine использует каскадный подход: сначала извлекает ключевые предложения (extractive summarization), затем перефразирует их абстрактно (abstractive summarization) с помощью маленькой модели. Важный нюанс: сжатые фрагменты должны сохранять ссылки на оригинальные сущности в графе памяти. Иначе вы потеряете контекстную связность.

# Ядро Compression Engine
import re
from transformers import pipeline  # версия 5.0+

class CompressionEngine:
    def __init__(self):
        # Инициализируем две модели: для извлечения и для абстракции
        # На 2026 год используем оптимальные модели из Hugging Face Hub
        self.extractor = pipeline(
            "text-classification",
            model="facebook/bart-large-mnli"  # или более новая версия
        )
        self.summarizer = pipeline(
            "summarization",
            model="google/pegasus-x-base"  # актуальная lightweight модель
        )
    
    def compress_interaction(self, interaction: Dict) -> str:
        """Сжимает одно взаимодействие (пара user+AI)."""
        full_text = f"User: {interaction['user']}\nAI: {interaction['ai']}"
        
        # Шаг 1: Извлекаем ключевые предложения
        sentences = self._split_into_sentences(full_text)
        if len(sentences) <= 2:
            return full_text  # нечего сжимать
            
        # Шаг 2: Оцениваем важность каждого предложения
        scores = []
        for sent in sentences:
            # Используем простую эвристику + модель
            score = self._score_sentence(sent)
            scores.append((sent, score))
        
        # Шаг 3: Берем топ-3 предложения по важности
        top_sentences = sorted(scores, key=lambda x: x[1], reverse=True)[:3]
        extractive_summary = ' '.join([s[0] for s in top_sentences])
        
        # Шаг 4: Абстрактное суммирование (если нужно сильнее сжать)
        if len(extractive_summary.split()) > 50:
            abstractive = self.summarizer(
                extractive_summary,
                max_length=30,
                min_length=10,
                do_sample=False
            )[0]['summary_text']
            return abstractive
        return extractive_summary
    
    def _score_sentence(self, sentence: str) -> float:
        """Комбинированная оценка важности предложения."""
        # Эвристики: длина, наличие именованных сущностей, вопросительные слова
        score = 0.0
        score += min(len(sentence.split()) / 20, 1.0)  # нормализованная длина
        
        # Проверяем на наличие важных маркеров
        important_markers = ['important', 'key', 'critical', 'decision', 'agreed', 'prefers']
        for marker in important_markers:
            if marker in sentence.lower():
                score += 0.3
        
        # Можно добавить модель для классификации важности
        return score

Этот подход дает сжатие в 3-10 раз в зависимости от избыточности текста. Критично: никогда не сжимайте последние 2-3 обмена - они должны оставаться в сыром виде для контекстуальной связности.

Reranking Gateway: что отправить в промпт прямо сейчас

Самая хитрая часть. У вас есть: оперативная память (последние сообщения), долговременная память (граф фактов), сжатые воспоминания. Какой набор отправить в LLM для генерации ответа? Reranking Gateway решает эту задачу через многоуровневую систему релевантности. Он не просто ищет семантическое сходство (как в наивном RAG), а учитывает:

• Временную близость (свежее важнее)
• Смысловую связность с текущим запросом
• Важность узла в графе (центральность)
• Частоту обращения к фрагменту памяти

Итоговый промпт формируется как слоеный пирог: самые релевантные фрагменты из долговременной памяти идут первыми, затем сжатый контекст, затем сырая оперативная память. Общая длина не должна превышать 70% от контекстного окна LLM (оставшиеся 30% - для генерации ответа). Если вы работаете с локальными моделями и сталкиваетесь с ограничениями, вам пригодится статья о расширении памяти локального AI.

class RerankingGateway:
    def __init__(self, memory_manager, compression_engine):
        self.mm = memory_manager
        self.ce = compression_engine
        
    def build_context(self, user_query: str, max_tokens: int = 6000) -> str:
        """Строит контекст для промпта из всех источников памяти."""
        context_parts = []
        
        # 1. Самые релевантные узлы из долговременной памяти (графа)
        relevant_nodes = self._retrieve_from_graph(user_query, limit=3)
        for node in relevant_nodes:
            context_parts.append(f"[Memory] {node['content']}")
        
        # 2. Сжатая история (кроме последних 3 обменов)
        compressed_history = self._get_compressed_history()
        context_parts.append(f"[Compressed History] {compressed_history}")
        
        # 3. Сырая оперативная память (последние 3-5 обменов)
        raw_recent = self._get_raw_recent(limit=5)
        for interaction in raw_recent:
            context_parts.append(f"User: {interaction['user']}")
            context_parts.append(f"AI: {interaction['ai']}")
        
        # 4. Текущий запрос
        context_parts.append(f"Current query: {user_query}")
        
        # Собираем всё, следя за лимитом токенов
        full_context = '\n\n'.join(context_parts)
        estimated_tokens = self._count_tokens(full_context)
        
        # Если превышаем лимит, итеративно убираем наименее важное
        while estimated_tokens > max_tokens and len(context_parts) > 1:
            # Удаляем самый старый сжатый фрагмент
            context_parts.pop(1)  # индекс 1 - сжатая история
            full_context = '\n\n'.join(context_parts)
            estimated_tokens = self._count_tokens(full_context)
            
        return full_context
    
    def _retrieve_from_graph(self, query: str, limit: int):
        """Извлекает узлы из графа, релевантные запросу."""
        # Используем комбинацию семантического поиска и анализа графа
        # 1. Эмбеддинг запроса
        query_embedding = self._get_embedding(query)
        
        # 2. Поиск по векторному индексу узлов (если он есть)
        # 3. Учет centrality в графе
        # Возвращаем топ-N узлов
        return []  # заглушка

С чем сравнивать? Альтернативы, которые разочаровывают

OpenAI Assistants с их файловой памятью? Удобно для прототипов, но непрозрачно и дорого для продакшена. Статья OpenAI Assistants против кастомной платформы показывает, где кастомное решение бьет готовое. Встроенная память в локальных LLM (Llama 3.2, Qwen2.5)? Часто это просто скрытый буфер, который чистится при перезагрузке. Скользящее окно контекста? Убивает долгосрочную связность. Memory OS и подобные концепции? Звучат круто, но на практике оказываются overengineering для большинства задач (подробнее в статье о бесконечном контексте и Memory OS).

Кому это нужно? (Спойлер: не всем)

Эта архитектура - не серебряная пуля. Она избыточна для чат-бота, который отвечает на вопросы про погоду. Но если вы строите:

• Кодинг-агента, который помнит всю сессию разработки (см. контекст-инжиниринг для coding-агентов)
• Персонального ассистента, который учится вашим предпочтениям
• Поддержку клиентов с многоходовыми кейсами
• Игровых NPC с длительной памятью (вам может помочь статья о борьбе с деградацией контекста в играх)

...тогда контекстный движок спасет ваш проект от распада через 15 минут диалога.

Начните с простой версии: Memory Manager на списках, сжатие через текстовые эвристики, реранкинг на основе TF-IDF. Добавляйте сложность постепенно, когда упретесь в ограничения. И помните: идеальная память AI - это не та, что помнит все, а та, что забывает неважное в правильный момент.

Подписаться на канал