Зашариваем память: строим систему для LLM с нуля как в Mem0

Зачем LLM вообще нужна память?

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

Проблема в том, что у LLM нет долговременной памяти. Каждый запрос - это чистый лист. Контекстное окно ограничено, и все, что выходит за его пределы, безвозвратно теряется. RAG частично решает проблему, но он больше про поиск по документам, а не про запоминание диалога.

Что мы будем строить и как это работает

Наша система будет состоять из трех ключевых компонентов:

1. Векторное хранилище для семантического поиска похожих воспоминаний
2. Суммаризатор для сжатия длинных диалогов в ключевые факты
3. Извлекатель сущностей для структурирования информации

Архитектура простая, но эффективная. Когда пользователь задает вопрос, мы:

• Ищем похожие воспоминания в векторной БД
• Извлекаем релевантные факты
• Добавляем их в промпт к LLM
• Обновляем память новыми фактами из ответа

Это похоже на системы долговременной памяти, но с конкретной реализацией.

1Настраиваем окружение и устанавливаем зависимости

Начнем с базового набора библиотек. Нам понадобятся:

• ChromaDB для векторного поиска (самая простая в настройке)
• Sentence Transformers для эмбеддингов
• LangChain для удобной работы с LLM (опционально, но сильно упрощает)
• spaCy для извлечения сущностей

# Создаем виртуальное окружение и устанавливаем зависимости
python -m venv mem0_env
source mem0_env/bin/activate  # На Windows: mem0_env\Scripts\activate

pip install chromadb sentence-transformers langchain langchain-community
pip install spacy
python -m spacy download en_core_web_sm  # Модель для английского

2Создаем базовый класс памяти

Начнем с каркаса. Определим, что такое "воспоминание" и как мы будем его хранить.

from datetime import datetime
from typing import List, Dict, Any, Optional
import json

class Memory:
    """Класс для представления одного воспоминания"""
    def __init__(self, 
                 content: str, 
                 metadata: Optional[Dict] = None,
                 timestamp: Optional[str] = None):
        self.content = content  # Текст воспоминания
        self.metadata = metadata or {}
        self.timestamp = timestamp or datetime.now().isoformat()
        self.embedding = None  # Векторное представление
    
    def to_dict(self) -> Dict:
        return {
            "content": self.content,
            "metadata": self.metadata,
            "timestamp": self.timestamp
        }
    
    @classmethod
    def from_dict(cls, data: Dict) -> 'Memory':
        return cls(
            content=data["content"],
            metadata=data.get("metadata", {}),
            timestamp=data.get("timestamp")
        )

Теперь создадим основной класс системы памяти. Он будет отвечать за все операции: добавление, поиск, обновление.

class MemorySystem:
    """Основная система памяти"""
    
    def __init__(self, 
                 embedding_model: str = "all-MiniLM-L6-v2",
                 persist_directory: str = "./mem0_data"):
        self.embedding_model_name = embedding_model
        self.persist_directory = persist_directory
        self.memories: List[Memory] = []
        self._init_embedding_model()
        self._init_vector_store()
    
    def _init_embedding_model(self):
        """Инициализируем модель для создания эмбеддингов"""
        from sentence_transformers import SentenceTransformer
        self.embedder = SentenceTransformer(self.embedding_model_name)
    
    def _init_vector_store(self):
        """Инициализируем ChromaDB для векторного поиска"""
        import chromadb
        from chromadb.config import Settings
        
        self.chroma_client = chromadb.Client(Settings(
            persist_directory=self.persist_directory,
            anonymized_telemetry=False
        ))
        
        # Создаем или получаем коллекцию
        try:
            self.collection = self.chroma_client.get_collection("memories")
        except:
            self.collection = self.chroma_client.create_collection("memories")

3Реализуем добавление и поиск воспоминаний

Самое интересное. Научим систему запоминать и находить информацию.

    def add_memory(self, content: str, metadata: Optional[Dict] = None) -> str:
        """Добавляем новое воспоминание"""
        memory = Memory(content, metadata)
        
        # Создаем эмбеддинг
        embedding = self.embedder.encode([content])[0]
        memory.embedding = embedding.tolist()
        
        # Добавляем в список
        self.memories.append(memory)
        
        # Сохраняем в ChromaDB
        memory_id = str(len(self.memories))
        self.collection.add(
            embeddings=[embedding.tolist()],
            documents=[content],
            metadatas=[metadata or {}],
            ids=[memory_id]
        )
        
        return memory_id
    
    def search_memories(self, query: str, k: int = 5) -> List[Memory]:
        """Ищем похожие воспоминания по семантическому сходству"""
        # Создаем эмбеддинг для запроса
        query_embedding = self.embedder.encode([query])[0]
        
        # Ищем в ChromaDB
        results = self.collection.query(
            query_embeddings=[query_embedding.tolist()],
            n_results=k
        )
        
        # Восстанавливаем объекты Memory
        retrieved_memories = []
        for i in range(len(results["ids"][0])):
            memory = Memory(
                content=results["documents"][0][i],
                metadata=results["metadatas"][0][i],
            )
            retrieved_memories.append(memory)
        
        return retrieved_memories

Проверим работу базовой системы:

# Тестируем
memory_system = MemorySystem()

# Добавляем воспоминания
memory_system.add_memory("Пользователь любит кофе и работает программистом")
memory_system.add_memory("У пользователя аллергия на арахис")
memory_system.add_memory("Пользователь планирует отпуск в Испании в июле")

# Ищем
results = memory_system.search_memories("Что пользователь любит пить?")
for memory in results:
    print(f"- {memory.content}")

4Добавляем суммаризацию длинных диалогов

Проблема: пользователь отправляет сообщение на 500 слов. Хранить его целиком - расточительно. Нужно выделить суть.

Есть два подхода:

1. Использовать отдельную модель для суммаризации (быстро, но требует GPU)
2. Использовать саму LLM через промпты (точнее, но медленнее)

Реализуем второй вариант, так как он более гибкий. Подключим LangChain для работы с LLM.

from langchain.llms import Ollama  # Для локальных моделей
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain

class Summarizer:
    """Класс для суммаризации длинного текста"""
    
    def __init__(self, model_name: str = "llama3.2"):
        # Используем Ollama для локальных моделей
        self.llm = Ollama(model=model_name, temperature=0.1)
        
        self.summary_prompt = PromptTemplate(
            input_variables=["text"],
            template="""Извлеки ключевые факты из следующего текста. 
            Перечисли только конкретные факты, которые могут быть полезны для будущих разговоров.
            Текст: {text}
            Ключевые факты:"""
        )
        self.chain = LLMChain(llm=self.llm, prompt=self.summary_prompt)
    
    def summarize(self, text: str) -> str:
        """Суммаризируем текст"""
        if len(text.split()) < 50:  # Короткие тексты не суммаризируем
            return text
        
        try:
            summary = self.chain.run(text=text)
            return summary.strip()
        except Exception as e:
            print(f"Ошибка суммаризации: {e}")
            # Возвращаем первые 100 слов в случае ошибки
            return " ".join(text.split()[:100])

5Извлекаем структурированную информацию

Сырой текст - это хорошо. Структурированные данные - лучше. Научим систему извлекать сущности: имена, даты, предпочтения, факты.

import spacy
from typing import List, Dict

class EntityExtractor:
    """Извлекаем структурированные сущности из текста"""
    
    def __init__(self):
        # Загружаем модель spaCy
        self.nlp = spacy.load("en_core_web_sm")
        
        # Определяем интересующие нас типы сущностей
        self.relevant_entities = {
            "PERSON", "ORG", "GPE", "DATE", 
            "PRODUCT", "EVENT", "NORP"
        }
    
    def extract(self, text: str) -> Dict[str, List[str]]:
        """Извлекаем сущности из текста"""
        doc = self.nlp(text)
        
        entities = {
            "people": [],
            "organizations": [],
            "locations": [],
            "dates": [],
            "misc": []
        }
        
        for ent in doc.ents:
            if ent.label_ in self.relevant_entities:
                # Классифицируем по типам
                if ent.label_ == "PERSON":
                    entities["people"].append(ent.text)
                elif ent.label_ in ["ORG", "PRODUCT"]:
                    entities["organizations"].append(ent.text)
                elif ent.label_ in ["GPE", "LOC"]:
                    entities["locations"].append(ent.text)
                elif ent.label_ == "DATE":
                    entities["dates"].append(ent.text)
                else:
                    entities["misc"].append(ent.text)
        
        # Удаляем дубликаты
        for key in entities:
            entities[key] = list(set(entities[key]))
        
        return entities

Теперь интегрируем все компоненты в основную систему:

class EnhancedMemorySystem(MemorySystem):
    """Улучшенная система памяти с суммаризацией и извлечением сущностей"""
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.summarizer = Summarizer()
        self.entity_extractor = EntityExtractor()
        
    def add_memory_enhanced(self, content: str, 
                           metadata: Optional[Dict] = None) -> Dict:
        """Добавляем воспоминание с суммаризацией и извлечением сущностей"""
        
        # Суммаризируем если текст длинный
        processed_content = self.summarizer.summarize(content)
        
        # Извлекаем сущности
        entities = self.entity_extractor.extract(content)
        
        # Объединяем метаданные
        enhanced_metadata = metadata or {}
        enhanced_metadata.update({
            "original_length": len(content.split()),
            "processed_length": len(processed_content.split()),
            "entities": entities,
            "is_summarized": len(content.split()) > len(processed_content.split())
        })
        
        # Добавляем в память
        memory_id = self.add_memory(processed_content, enhanced_metadata)
        
        return {
            "memory_id": memory_id,
            "processed_content": processed_content,
            "entities": entities
        }

6Интегрируем с LLM и создаем полноценный чат

Теперь свяжем нашу систему памяти с LLM. Создадим чат-бота, который помнит разговор.

class ChatWithMemory:
    """Чат-бот с долговременной памятью"""
    
    def __init__(self, 
                 memory_system: EnhancedMemorySystem,
                 llm_model: str = "llama3.2"):
        self.memory = memory_system
        self.llm = Ollama(model=llm_model, temperature=0.7)
        
        self.chat_prompt = PromptTemplate(
            input_variables=["memories", "question", "chat_history"],
            template="""Ты - полезный ассистент с памятью о прошлых разговорах.
            
            Вот что ты помнишь о пользователе:
            {memories}
            
            История текущего разговора:
            {chat_history}
            
            Текущий вопрос: {question}
            
            Ответь, учитывая воспоминания о пользователе."""
        )
        self.chain = LLMChain(llm=self.llm, prompt=self.chat_prompt)
        
        self.current_chat_history = []
    
    def chat(self, user_input: str) -> str:
        """Обрабатываем пользовательский ввод"""
        
        # 1. Ищем релевантные воспоминания
        relevant_memories = self.memory.search_memories(user_input, k=3)
        
        # 2. Форматируем воспоминания для промпта
        memories_text = "\n".join([
            f"- {mem.content}" for mem in relevant_memories
        ])
        
        # 3. Форматируем историю чата
        chat_history_text = "\n".join([
            f"Пользователь: {q}\nАссистент: {a}" 
            for q, a in self.current_chat_history[-5:]  # Последние 5 пар
        ])
        
        # 4. Генерируем ответ
        response = self.chain.run(
            memories=memories_text,
            question=user_input,
            chat_history=chat_history_text
        )
        
        # 5. Сохраняем текущий обмен в историю
        self.current_chat_history.append((user_input, response))
        
        # 6. Извлекаем и сохраняем новые факты из ответа пользователя
        # (Упрощенно - сохраняем сам вопрос как факт)
        self.memory.add_memory_enhanced(user_input, {
            "type": "user_statement",
            "response_to": response[:50] + "..." if len(response) > 50 else response
        })
        
        return response

Где это все сломается и как починить

Реализация выше - это MVP. В продакшене вас ждут подводные камни.

Что дальше? Куда развивать систему

Базовая система работает. Но это только начало. Вот что можно добавить:

• Категоризация воспоминаний. Автоматически определять тип информации (личные предпочтения, рабочие вопросы, развлечения)
• Приоритет фактов. Некоторые факты важнее других. Аллергия на арахис важнее любимого цвета.
• Автоматическое забывание. Устаревшая информация должна удаляться или архивироваться.
• Мультимодальность. Запоминать не только текст, но и изображения, аудио, структурированные данные.

Самое главное - тестировать. Запустите систему на реальных диалогах. Посмотрите, какие воспоминания она сохраняет, какие находит. Отлаживайте промпты для суммаризации - от них зависит 80% качества.

И помните: идеальной системы памяти не существует. Каждая LLM, каждый use case требуют своей настройки. Избегайте основных ошибок, экспериментируйте с разными подходами.

Ваш чат-бот наконец-то перестанет страдать склерозом. И это только начало.