Реализация памяти агента (краткосрочная и долгосрочная)

60 минут Урок 27

Введение: Почему агентам нужна память?

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

Представьте, что вы нанимаете блестящего аналитика. Он эрудирован, мгновенно считает и знает всё, что есть в интернете. Но у него есть одна особенность: каждое утро, приходя на работу, он полностью забывает, о чем вы договаривались вчера, какие проекты вел и как вас зовут. Полезен ли такой сотрудник? Ситуативно — да. Но для долгосрочной работы — вряд ли.

LLM (Large Language Models) по своей природе stateless (без сохранения состояния). Каждый запрос для модели — как первый день жизни. Чтобы превратить модель в Агента, способного вести осмысленный диалог и решать сложные задачи во времени, нам нужно архитектурно добавить ему память.

В этом уроке мы разберем:

  • В чем разница между краткосрочной (Working Memory) и долгосрочной (Long-term Memory) памятью.
  • Как эффективно управлять контекстным окном Gemini, чтобы не разориться на токенах.
  • Как использовать векторные базы данных для создания «второго мозга» агента.
  • Как реализовать гибридную систему памяти на Python.

Анатомия памяти агента

В когнитивной науке память человека часто делят на кратковременную (рабочую) и долговременную. В архитектуре ИИ мы используем ту же аналогию.

1. Краткосрочная память (Short-term Memory / Context)

Это то, что агент «держит в голове» прямо сейчас. Технически — это содержимое контекстного окна (prompt), отправляемого в Gemini.

  • Что хранит: Текущий диалог, инструкции, недавние действия.
  • Ограничения: Объем контекста (хотя у Gemini 3 он огромен, он не бесконечен), стоимость, задержка (latency).
  • Жизненный цикл: Теряется после завершения сессии или переполнения окна.

2. Долгосрочная память (Long-term Memory)

Это «записная книжка» или архив агента. Это внешнее хранилище, откуда информация извлекается по мере необходимости.

  • Что хранит: Факты о пользователе, базу знаний компании, исторические данные, опыт прошлых сессий.
  • Реализация: Обычно через векторные базы данных (Vector Stores) и поиск по семантической близости (RAG).
  • Жизненный цикл: Персистентна (сохраняется навсегда).

Реализация краткосрочной памяти: Управление историей

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

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

  1. Скользящее окно (Sliding Window): Храним только последние N сообщений (например, 10). Старые забываются.
  2. Буфер токенов: Храним историю до достижения лимита токенов, затем удаляем старейшие.
  3. Суммаризация (Summarization): Когда история становится слишком длинной, мы просим саму модель «сжать» старую часть диалога в краткое резюме и добавляем его в начало контекста.

Давайте посмотрим, как реализовать базовую память с суммаризацией на Python.

python
import google.generativeai as genai
from collections import deque

# Предполагаем, что API ключ уже настроен
model = genai.GenerativeModel('gemini-3-pro')

class ConversationMemory:
    def __init__(self, max_len=10):
        # Используем deque для хранения последних сообщений
        self.history = deque(maxlen=max_len)
        self.summary = ""
    
    def add_message(self, role, text):
        self.history.append({"role": role, "parts": [text]})
    
    def get_context(self):
        # Возвращаем историю в формате, понятном Gemini
        return list(self.history)

    def summarize_memory(self):
        """
        Метод для сжатия истории. 
        В реальном проекте вызывается, когда токенов становится слишком много.
        """
        if len(self.history) < 2: return
        
        # Превращаем историю в текст для суммаризации
        history_text = "\n".join([f"{msg['role']}: {msg['parts'][0]}" for msg in self.history])
        
        prompt = f"Проанализируй следующий диалог и обнови текущее резюме беседы.\nТекущее резюме: {self.summary}\nДиалог: {history_text}\nНовое краткое резюме:"
        
        # Используем модель для генерации саммари (можно использовать flash-модель для экономии)
        response = model.generate_content(prompt)
        self.summary = response.text
        
        # Очищаем историю, оставляя только summary (в реальности можно оставить пару последних сообщений)
        self.history.clear()
        print(f"[Память оптимизирована]: {self.summary}")

# Пример использования
agent_mem = ConversationMemory(max_len=5)
agent_mem.add_message("user", "Привет, меня зовут Алексей.")
agent_mem.add_message("model", "Привет, Алексей! Чем могу помочь?")
agent_mem.add_message("user", "Я люблю программировать на Python.")

# В какой-то момент мы решаем сжать память
agent_mem.summarize_memory()

Долгосрочная память: Векторные базы и RAG

Краткосрочная память помогает поддерживать нить разговора. Но что, если пользователь вернется через неделю и спросит: «Помнишь, я рассказывал про свой проект? Что ты о нем думаешь?»

Здесь в игру вступает Semantic Search (Семантический поиск). Мы не можем искать по точному совпадению слов, так как пользователь может сформулировать мысль иначе. Мы используем Эмбеддинги (Embeddings).

Процесс работы долгосрочной памяти:

  1. Сохранение (Write): Когда происходит что-то важное, агент берет текст, превращает его в вектор (массив чисел) с помощью модели эмбеддингов Gemini и сохраняет в векторную БД.
  2. Извлечение (Read): Когда поступает новый запрос, мы тоже превращаем его в вектор и ищем в БД самые «близкие» векторы (по косинусному сходству).
  3. Генерация (Generate): Найденные куски информации добавляются в промпт как контекст.

В Gemini 3 API работа с эмбеддингами максимально упрощена. Давайте создадим класс `LongTermMemory`.

python
import numpy as np
# Для примера используем простую структуру, в проде - ChromaDB, Pinecone или pgvector

class SimpleVectorStore:
    def __init__(self):
        self.documents = []
        self.vectors = []

    def add(self, text, vector):
        self.documents.append(text)
        self.vectors.append(vector)

    def search(self, query_vector, top_k=3):
        if not self.vectors:
            return []
        
        # Вычисляем косинусное сходство (простая реализация через numpy)
        # A . B / (|A| * |B|)
        scores = []
        for doc_vec in self.vectors:
            score = np.dot(query_vector, doc_vec) / (np.linalg.norm(query_vector) * np.linalg.norm(doc_vec))
            scores.append(score)
        
        # Сортируем и берем топ-K результатов
        sorted_indices = np.argsort(scores)[::-1][:top_k]
        return [self.documents[i] for i in sorted_indices]

class Agent:
    def __init__(self):
        self.ltm = SimpleVectorStore()
        self.chat = model.start_chat()

    def memorize(self, text):
        # Получаем эмбеддинг через Gemini API
        embedding = genai.embed_content(
            model="models/text-embedding-004",
            content=text,
            task_type="retrieval_document"
        )['embedding']
        
        self.ltm.add(text, embedding)
        print(f"[Запомнил в LTM]: {text[:30]}...")

    def ask(self, query):
        # 1. Сначала ищем в памяти релевантную информацию
        query_embedding = genai.embed_content(
            model="models/text-embedding-004",
            content=query,
            task_type="retrieval_query"
        )['embedding']
        
        relevant_info = self.ltm.search(query_embedding)
        context_str = "\n".join(relevant_info)
        
        # 2. Формируем промпт с контекстом
        full_prompt = f"""
        Используй следующую информацию из памяти, если она полезна для ответа:
        ---
        {context_str}
        ---
        Вопрос пользователя: {query}
        """
        
        response = self.chat.send_message(full_prompt)
        return response.text

# Тестируем
my_agent = Agent()

# Этап обучения (запоминания)
my_agent.memorize("Пользователь работает DevOps инженером.")
my_agent.memorize("Любимый язык программирования пользователя - Go, но он ненавидит Java.")
my_agent.memorize("В компании используется Kubernetes.")

# Этап запроса (проверка памяти)
print("\nОтвет агента:", my_agent.ask("Стоит ли мне использовать Spring Boot для нового микросервиса?"))

Нюансы Enterprise решений

В приведенном выше коде мы создали игрушечную память. В реальных Enterprise системах на базе Gemini 3 нужно учитывать следующее:

  • Context Caching (Кэширование контекста): Уникальная фича Gemini. Если у вас есть огромная база знаний (например, документация на 1000 страниц), которую вы подгружаете в промпт, использование RAG может быть излишним. Вы можете загрузить весь документ в кэш Gemini и обращаться к нему дешевле, чем передавать его каждый раз заново. Это «промежуточная» память между STM и LTM.
  • Разделение памяти: Храните факты о пользователе («User Memory») отдельно от базы знаний («Knowledge Base»). Это улучшает качество поиска.
  • Рефлексия (Reflection): Агент не должен запоминать всё. После завершения диалога запустите фоновый процесс, где модель анализирует чат и решает: «Что из этого стоит сохранить в долгосрочную память?». Не засоряйте векторную базу шумом типа «Привет» или «Спасибо».

Упражнение

Создайте функцию `auto_archive`, которая принимает историю чата (список сообщений). Функция должна использовать Gemini для извлечения только ВАЖНЫХ фактов о пользователе (имя, предпочтения, даты) в виде списка строк, которые затем можно было бы сохранить в векторную базу данных.

Вопрос

Какой подход наиболее эффективен для оптимизации затрат при работе с очень длинной историей диалога в краткосрочной памяти, если важно сохранить суть беседы, но не точные формулировки?

Заключение

Реализация памяти — это то, что отличает чат-бота от настоящего цифрового ассистента. Комбинируя буфер сообщений для текущего контекста и векторный поиск для глобального опыта, вы создаете агента, который «знает» своего пользователя.

В следующем уроке мы разберем, как агенты могут использовать инструменты (Function Calling), чтобы не только помнить, но и действовать во внешнем мире.