Реализация долговременной памяти агента (Memory Stores)

Введение: Проблема «Амнезии» у Искусственного Интеллекта

Добро пожаловать в шестой модуль. Сегодня мы поговорим о том, что отличает простого чат-бота от настоящего цифрового помощника: о памяти. Представьте, что вы наняли гениального аналитика. Он знает всё на свете, цитирует Шекспира и пишет код на Python с закрытыми глазами. Но есть одна проблема: каждый раз, когда он выходит из комнаты и возвращается (то есть, начинается новая сессия), он забывает, как вас зовут, над каким проектом вы работали вчера и что вы просили не делать.

Это классическая проблема LLM (Large Language Models). По своей природе они stateless (без сохранения состояния). Каждое обращение к API для модели — это чистый лист. В предыдущих версиях мы решали это передачей всей истории переписки в каждом запросе. Но что делать, если история переписки занимает 500 страниц? А если нам нужно помнить информацию, полученную месяц назад?

В эпоху Gemini 3 и огромных контекстных окон (1M+ токенов) соблазн просто «закинуть всё в контекст» велик. Однако, для построения действительно автономных агентов нам нужны структурированные хранилища памяти (Memory Stores). В этом уроке мы разберем, как превратить кратковременный буфер обмена в долговременную базу знаний.

Типы памяти агента

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

• Кратковременная (Short-term): Это текущее контекстное окно. То, что происходит "здесь и сейчас". В Gemini 3 оно огромно, но оно дорогое и очищается после завершения сессии.
• Долговременная процедурная (Long-term Procedural): Это инструкции, System Prompts и Few-Shot примеры. Это "навыки" агента.
• Долговременная эпизодическая (Long-term Episodic): Это то, о чем мы говорим сегодня. Хранилище фактов, прошлых диалогов и документов. Реализуется через RAG (Retrieval-Augmented Generation) и векторные базы данных.

Ключевая технология здесь — Векторные Эмбеддинги (Embeddings). Мы не ищем текст по ключевым словам (как Ctrl+F). Мы превращаем смысл текста в список чисел (вектор) и ищем похожие смыслы.

import google.generativeai as genai
import numpy as np
import os

# Настройка API (предполагается, что ключ в переменных окружения)
genai.configure(api_key=os.environ["GEMINI_API_KEY"])

# Пример: Как машина "видит" смысл
# Мы используем модель text-embedding-004 для преобразования текста в векторы
def get_embedding(text):
    result = genai.embed_content(
        model="models/text-embedding-004",
        content=text,
        task_type="retrieval_document",
        title="Lesson Example"
    )
    return result['embedding']

text1 = "Кот сидит на коврике"
text2 = "Кошка отдыхает на полу"
text3 = "Ядерная физика сложна"

vec1 = get_embedding(text1)
vec2 = get_embedding(text2)
vec3 = get_embedding(text3)

# Вычисляем косинусное сходство (простая математика векторов)
def cosine_similarity(v1, v2):
    return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))

print(f"Сходство Кота и Кошки: {cosine_similarity(vec1, vec2):.4f}")
print(f"Сходство Кота и Физики: {cosine_similarity(vec1, vec3):.4f}")

# Ожидаемый результат:
# Сходство Кота и Кошки будет высоким (близко к 0.8-0.9)
# Сходство Кота и Физики будет низким (близко к 0.1-0.2)

Архитектура Memory Store

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

1. Save (Zapominanie): Принимает информацию, создает эмбеддинг и сохраняет в базу.
2. Retrieve (Vspominanie): Принимает запрос агента (например, "Что пользователь говорил о своих предпочтениях?"), превращает его в вектор и находит самые близкие записи в базе.
3. Inject (Vnedrenie): Вставляет найденную информацию в промпт перед отправкой в LLM.

Для реальных проектов мы бы использовали ChromaDB, Pinecone или Qdrant. В рамках урока мы создадим упрощенную in-memory версию, чтобы понять суть, не отвлекаясь на настройку баз данных.

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

    def add_memory(self, text):
        """Сохраняет факт в память"""
        embedding = genai.embed_content(
            model="models/text-embedding-004",
            content=text,
            task_type="retrieval_document"
        )['embedding']
        
        self.documents.append(text)
        self.vectors.append(embedding)
        print(f"[Memory] Сохранено: {text[:30]}...")

    def retrieve(self, query, top_k=2):
        """Ищет релевантные факты"""
        if not self.documents:
            return []
            
        query_embedding = genai.embed_content(
            model="models/text-embedding-004",
            content=query,
            task_type="retrieval_query"
        )['embedding']
        
        # Вычисляем сходство со всеми документами
        similarities = []
        for doc_vec in self.vectors:
            sim = np.dot(query_embedding, doc_vec) / (np.linalg.norm(query_embedding) * np.linalg.norm(doc_vec))
            similarities.append(sim)
            
        # Сортируем и берем топ-K лучших
        # (zip объединяет списки, sorted сортирует по сходству)
        results = sorted(zip(self.documents, similarities), key=lambda x: x[1], reverse=True)
        
        return [doc for doc, score in results[:top_k] if score > 0.6] # Фильтр по порогу релевантности

Интеграция с агентом Gemini

Теперь самое интересное. У нас есть "мозг" (Gemini) и "блокнот" (MemoryStore). Нужно научить агента пользоваться блокнотом.

Есть два способа это сделать:

1. Прозрачный (Implicit): Мы (разработчики) скрыто от агента ищем информацию по каждому запросу пользователя и добавляем её в контекст. Агент даже не знает, что он что-то "вспоминает", он просто видит информацию в промпте.
2. Инструментальный (Explicit / Tool Use): Мы даем агенту инструмент (Function Calling) search_memory() и save_memory(). Агент сам решает, когда нужно что-то запомнить, а когда — поискать.

Второй подход более характерен для автономных агентов, так как он снижает шум в контексте. Агент не читает всю память подряд, а ищет точечно.

Ниже пример Прозрачного подхода, так как он надежнее для базовых сценариев чата.

# Инициализация памяти
agent_memory = SimpleVectorMemory()

# Загрузим начальные знания (имитация прошлых бесед)
agent_memory.add_memory("Пользователя зовут Алексей.")
agent_memory.add_memory("Алексей работает DevOps инженером.")
agent_memory.add_memory("Алексей предпочитает объяснения на Python, а не на Bash.")

model = genai.GenerativeModel('gemini-1.5-pro-latest')

def chat_with_memory(user_input):
    # 1. Поиск релевантного контекста
    relevant_facts = agent_memory.retrieve(user_input)
    
    context_str = ""
    if relevant_facts:
        context_str = "\nНАЙДЕННАЯ ИНФОРМАЦИЯ ИЗ ПАМЯТИ:\n" + "\n".join(f"- {fact}" for fact in relevant_facts)
    
    # 2. Формирование промпта
    full_prompt = f"""
    Ты умный ассистент. Используй контекст ниже, чтобы лучше ответить пользователю.
    Если в контексте нет нужной информации, отвечай как обычно.
    
    {context_str}
    
    ВОПРОС ПОЛЬЗОВАТЕЛЯ: {user_input}
    """
    
    # 3. Генерация ответа
    response = model.generate_content(full_prompt)
    return response.text

# Тестируем
print("Bot:", chat_with_memory("Как мне лучше автоматизировать деплой?"))
# Ожидается, что бот предложит решение на Python, зная предпочтения Алексея.

Особенность Gemini: Context Caching

В экосистеме Google Gemini есть уникальная функция, которая является гибридом между краткосрочной и долгосрочной памятью — Context Caching.

Представьте, что у вас есть огромная документация (например, весь кодекс законов или техническая документация на 2000 страниц). Векторный поиск (RAG) может выдернуть куски, но иногда нужно, чтобы модель видела всё сразу для глобального анализа. Загружать 1 миллион токенов в каждом запросе — безумно дорого и медленно.

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

Примечание: Это не замена векторной базе (которая хранит бесконечное количество мелких фактов), это инструмент для работы с "тяжелым" активным контекстом.

python
def update_memory_if_needed(user_text, memory_store):
    check_model = genai.GenerativeModel('gemini-1.5-flash') # Используем Flash для скорости и экономии
    
    prompt = f"""
    Проанализируй сообщение пользователя.
    Содержит ли оно долгосрочный факт о пользователе (имя, профессия, локация, предпочтения, питомцы)?
    Ответь ТОЛЬКО JSON: {{"is_fact": true, "fact_text": "извлеченный факт"}} или {{"is_fact": false}}
    
    Сообщение: {user_text}
    """
    
    response = check_model.generate_content(prompt)
    # В реальном коде нужен парсинг JSON (json.loads), здесь упрощено
    import json
    try:
        data = json.loads(response.text.strip().replace('```json', '').replace('```', ''))
        if data.get('is_fact'):
            memory_store.add_memory(data['fact_text'])
            print(f"[Auto-Memory] Запомнил: {data['fact_text']}")
    exceptException as e:
        print("Ошибка парсинга:", e)

# Пример использования
# update_memory_if_needed("Кстати, я теперь веган", agent_memory)