Память — краеугольный камень интеллектуального агента. Без неё агент не способен учитывать контекст предыдущих взаимодействий, персонализировать ответы и накапливать знания. В отличие от классических детерминированных систем, где память сводится к хранению состояния в переменных, AI-агенты оперируют тремя качественно разными уровнями памяти, каждый со своей физической природой и временным горизонтом.

Рабочая память (Working Memory) — это то, что языковая модель «видит» в момент генерации: контекстное окно, ограниченное 4K–200K токенов в зависимости от модели. Это самый быстрый, но и самый дорогой уровень — каждый токен в контексте оплачивается и замедляет инференс. Краткосрочная память (Short-term) — история сообщений текущей сессии, обычно хранящаяся в Redis или PostgreSQL. Она переживает один вызов LLM, но теряется между сессиями. Долгосрочная память (Long-term) — это векторная база данных, где каждое значимое взаимодействие или факт сохраняется в виде эмбеддинга и может быть найдено по семантической близости через месяцы.

╔══════════════════════════════════════════════════════════════════╗
║          🧠  ТРИ УРОВНЯ ПАМЯТИ AI-АГЕНТА                        ║
╠══════════╦══════════════════╦══════════════╦═════════════════════╣
║ Уровень  ║ Горизонт         ║ Хранилище    ║ Объём / Стоимость   ║
╠══════════╬══════════════════╬══════════════╬═════════════════════╣
║ WORKING  ║ секунды-минуты   ║ RAM (inference)║ 4K–200K токенов   ║
║ SHORT    ║ часы-дни         ║ Redis / PG    ║ ~10K сообщений     ║
║ LONG     ║ недели-годы      ║ Vector DB     ║ миллионы записей   ║
╚══════════╩══════════════════╩══════════════╩═════════════════════╝

# Ключевой принцип: каждый уровень — компромисс между скоростью,
# стоимостью и глубиной хранения. Архитектура агента должна
# оркестрировать все три уровня как единый конвейер.

class MemoryTier:
    """Абстракция одного уровня памяти."""
    def __init__(self, name, ttl, capacity, cost_per_unit):
        self.name = name             # working | short | long
        self.ttl = ttl               # время жизни записи в секундах
        self.capacity = capacity       # максимальный объём
        self.cost = cost_per_unit     # стоимость хранения единицы

WORKING = MemoryTier("working", ttl=300,  capacity="128K tokens", cost="$0.01/1K tokens")
SHORT  = MemoryTier("short",   ttl=86400, capacity="10K msgs",     cost="~$0.00/msg")
LONG   = MemoryTier("long",    ttl=float('inf'), capacity="unlimited", cost="~$0.0001/record")

Рабочая память — это самое «горячее» и дорогое место. Каждый токен, попадающий в контекстное окно, увеличивает latency (квадратично для attention-механизма) и стоимость. Искусство управления рабочей памятью — в балансе: не потерять важный контекст, но и не перегрузить окно шумом. Для GPT-4o контекстное окно достигает 128K токенов, но эффективная длина ответа резко падает после ~64K — модель «забывает» середину диалога (феномен lost-in-the-middle).

Практические стратегии: (1) резервировать 30% окна под ответ модели, (2) использовать sliding window с сохранением system prompt, (3) суммировать старые сообщения вместо их удаления — компрессия вместо обрезки. Для подсчёта токенов критически использовать тот же токенизатор, что и модель (tiktoken для OpenAI, sentencepiece для opensource моделей).

pip install tiktoken

import tiktoken
from openai import OpenAI

client = OpenAI()
enc = tiktoken.encoding_for_model("gpt-4o")

class WorkingMemoryManager:
    """Управление контекстным окном с компрессией и приоритизацией."""

    def __init__(self, model="gpt-4o", max_tokens=96000, reserve_ratio=0.25):
        self.enc = tiktoken.encoding_for_model(model)
        self.model = model
        self.limit = max_tokens
        self.reserve = int(max_tokens * reserve_ratio)  # ~24K под ответ

    def count_tokens(self, text):
        """Точный подсчёт токенов."""
        return len(self.enc.encode(text))

    def prepare_context(self, system_prompt, history, new_query,
                       injected_memories=None):
        """Формирует оптимальное контекстное окно с компрессией."""
        available = self.limit - self.reserve

        # System prompt — нерушимый
        sys_tokens = self.count_tokens(system_prompt)
        available -= sys_tokens

        # Инжектим долгосрочные воспоминания (RAG)
        if injected_memories:
            mem_text = "\n".join(f"• {m}" for m in injected_memories[:5])
            available -= self.count_tokens(mem_text)

        # Новый запрос — обязательно
        available -= self.count_tokens(new_query)

        # История: старые сообщения суммируем, новые — as-is
        context_msgs = []
        spent = 0
        for msg in reversed(history):
            t = self.count_tokens(msg["content"])
            if spent + t > available * 0.7:
                # Старые сообщения — компрессия вместо обрезки
                summary = self._summarize_messages(history[:history.index(msg)])
                context_msgs.insert(0, {"role": "system", "content": f"[Резюме диалога] {summary}"})
                break
            context_msgs.insert(0, msg)
            spent += t

        return context_msgs

    def _summarize_messages(self, messages):
        """Компрессия: cheap модель суммирует старый контекст."""
        dialogue = "\n".join(f"{m['role']}: {m['content'][:200]}" for m in messages[-20:])
        resp = client.chat.completions.create(
            model="gpt-4o-mini",  # дешёвая модель для суммаризации
            messages=[{"role":"user", "content":f"Summarise this dialogue in 3 bullet points in Russian:\n{dialogue}"}],
            max_tokens=200
        )
        return resp.choices[0].message.content

Краткосрочная память — это связующее звено между вызовами LLM в рамках одной сессии. Когда агент делает 5 последовательных tool-call'ов, каждый вызов LLM должен «помнить» результаты предыдущих. Redis идеален для этой роли: атомарные операции, TTL на ключи, встроенные структуры (list для истории, hash для метаданных сессии). Важно: short-term память должна быть сериализуемой — при перезапуске сервера сессия восстанавливается из Redis.

Паттерн: append-only log сообщений в Redis List + периодическая компрессия в сводку. Для production-агентов с сотнями одновременных сессий необходимо шардирование по session_id и LRU-эвикция старых сессий. Альтернативы: PostgreSQL (если нужна SQL-аналитика по истории), Dragonfly (совместим с Redis, но в 25 раз быстрее на больших нагрузках).

pip install redis[hiredis] orjson

import redis, orjson, time
from datetime import datetime, timedelta

r = redis.Redis(host='localhost', port=6379, decode_responses=True,
                      socket_keepalive=True, health_check_interval=30)

class SessionMemory:
    """Production-ready сессионная память с компрессией и лимитами."""

    def __init__(self, session_id, ttl_hours=24, max_messages=500):
        self.sid = session_id
        self.key_msg = f"sess:{session_id}:msgs"
        self.key_meta = f"sess:{session_id}:meta"
        self.key_summary = f"sess:{session_id}:summary"
        self.ttl = ttl_hours * 3600
        self.max_msgs = max_messages

    def add_message(self, role, content, metadata=None):
        """Добавить сообщение с атомарным ограничением длины."""
        msg = {
            "role": role,
            "content": content[:8000],  # защита от раздувания
            "ts": datetime.now().isoformat(),
            "meta": metadata or {}
        }
        pipe = r.pipeline()
        pipe.rpush(self.key_msg, orjson.dumps(msg))
        pipe.ltrim(self.key_msg, -self.max_msgs, -1)  # держим только последние N
        pipe.expire(self.key_msg, self.ttl)
        pipe.hset(self.key_meta, mapping={"last_activity": str(time.time())})
        pipe.execute()

    def get_history(self, limit=50, since=None):
        """Получить историю с фильтрацией по времени."""
        raw = r.lrange(self.key_msg, -limit, -1)
        msgs = [orjson.loads(m) for m in raw]
        if since:
            msgs = [m for m in msgs if m["ts"] >= since]
        return [{"role": m["role"], "content": m["content"]} for m in msgs]

    def store_summary(self, summary_text):
        """Сохранить сжатую сводку диалога."""
        r.setex(self.key_summary, self.ttl, summary_text)

    def get_summary(self):
        """Получить сводку предыдущего контекста."""
        return r.get(self.key_summary)

    def is_active(self, timeout_sec=1800):
        """Проверить, активна ли сессия."""
        last = r.hget(self.key_meta, "last_activity")
        if not last: return False
        return (time.time() - float(last)) < timeout_sec

ChromaDB — open-source векторная база данных, оптимизированная для AI-приложений. В отличие от Pinecone (managed cloud), ChromaDB можно развернуть локально за 5 минут. Она поддерживает два режима: in-memory (для тестов) и persistent (на диске через DuckDB). Для агента с долгосрочной памятью ChromaDB хранит каждое значимое взаимодействие как (id, embedding, document, metadata)-кортеж. При новом запросе агент семантически ищет top-K релевантных воспоминаний и инжектит их в system prompt.

Ключевые метрики: размерность эмбеддинга (OpenAI ada-002 = 1536, text-embedding-3-large = 3072), расстояние (косинусное, L2, IP), стратегия индексации (HNSW для скорости, flat для точности). Важно: ChromaDB не масштабируется горизонтально из коробки — для production с >1M записей рассмотрите Qdrant (Rust, gRPC-native) или Pinecone.

pip install chromadb openai

import chromadb, hashlib, json
from datetime import datetime
from openai import OpenAI

client = OpenAI()

class ChromaLongTermMemory:
    """Долгосрочная память на ChromaDB с автоматической дедупликацией."""

    def __init__(self, persist_dir="./agent_ltm", collection_name="memories"):
        self.chroma = chromadb.PersistentClient(path=persist_dir)
        self.coll = self.chroma.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"}  # косинусное расстояние
        )
        self.emb_model = "text-embedding-3-small"  # дешевле ada-002

    def _embed(self, text):
        """OpenAI text-embedding-3-small: 1536-dim, $0.02/1M tokens."""
        r = client.embeddings.create(model=self.emb_model, input=text)
        return r.data[0].embedding

    def remember(self, fact, category="general", importance=1.0):
        """Сохранить факт с дедупликацией по хешу."""
        fact_id = f"mem_{hashlib.md5(fact.encode()).hexdigest()[:12]}"

        # Проверяем, не существует ли уже такой факт
        existing = self.coll.get(ids=[fact_id])
        if existing['ids']:
            self.coll.update(ids=[fact_id], metadatas=[{"last_seen": datetime.now().isoformat()}])
            return fact_id  # обновили timestamp, не дублируем

        self.coll.add(
            ids=[fact_id],
            documents=[fact],
            embeddings=[self._embed(fact)],
            metadatas=[{
                "category": category,
                "importance": importance,
                "created": datetime.now().isoformat(),
                "access_count": 0
            }]
        )
        return fact_id

    def recall(self, query, n=5, category_filter=None, min_importance=0.0):
        """Семантический поиск релевантных воспоминаний."""
        where_clause = {}
        if category_filter:
            where_clause["category"] = category_filter
        if min_importance > 0:
            where_clause["importance"] = {"$gte": min_importance}

        results = self.coll.query(
            query_embeddings=[self._embed(query)],
            n_results=n,
            where=where_clause if where_clause else None,
            include=["documents", "metadatas", "distances"]
        )
        return list(zip(results['documents'][0], results['distances'][0]))

    def forget_old(self, days=90, min_access=1):
        """Удаление устаревших/неиспользуемых воспоминаний."""
        cutoff = datetime.now() - timedelta(days=days)
        all_data = self.coll.get(include=["metadatas"])
        to_delete = []
        for i, (mid, meta) in enumerate(zip(all_data['ids'], all_data['metadatas'])):
            created = datetime.fromisoformat(meta['created']) if meta else cutoff
            if created < cutoff and meta.get('access_count', 0) < min_access:
                to_delete.append(mid)
        if to_delete:
            self.coll.delete(ids=to_delete)
            print(f"🧹 Forgot {len(to_delete)} stale memories")

Pinecone — полностью управляемая векторная база данных с serverless-архитектурой. В отличие от ChromaDB, которую вы администрируете сами, Pinecone берёт на себя репликацию, шардирование, бэкапы и мониторинг. Вы платите за поды (pod-based) или за потребление (serverless). Для production-агента с >100K пользователей Pinecone часто оказывается дешевле, чем содержание DevOps-инженера для self-hosted решения.

Ключевые особенности: metadata filtering (фильтрация по категориям, датам, важности без ущерба скорости), namespaces (изоляция данных разных пользователей в одном индексе), real-time upserts (изменения видны в следующем запросе). Стоимость: $0.096/час за pod на 1M 1536-dim векторов (~$70/мес). Serverless: $0.33/1M запросов на чтение + $2.00/1M на запись.

pip install pinecone-client openai

import os, time
from pinecone import Pinecone, ServerlessSpec
from openai import OpenAI

client = OpenAI()
pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))

class PineconeMemory:
    """Production long-term memory на Pinecone Serverless."""

    def __init__(self, index_name="agent-memory", dim=1536):
        # Создать serverless индекс, если не существует
        if index_name not in pc.list_indexes().names():
            pc.create_index(
                name=index_name,
                dimension=dim,
                metric="cosine",
                spec=ServerlessSpec(cloud="aws", region="us-east-1")
            )
            time.sleep(5)  # ждём инициализацию

        self.index = pc.Index(index_name)
        self.emb_model = "text-embedding-3-small"

    def _embed(self, texts):
        """Batched эмбеддинг (Pinecone позволяет upsert пачками)."""
        if isinstance(texts, str): texts = [texts]
        r = client.embeddings.create(model=self.emb_model, input=texts)
        return [d.embedding for d in r.data]

    def remember(self, user_id, fact, category="general"):
        """Сохранить факт в namespace пользователя."""
        vec_id = f"{user_id}_{hash(fact) % 10**10}"
        self.index.upsert(
            vectors=[{
                "id": vec_id,
                "values": self._embed(fact)[0],
                "metadata": {
                    "fact": fact[:500],
                    "category": category,
                    "user_id": user_id,
                    "ts": int(time.time())
                }
            }],
            namespace=f"user_{user_id}"  # изоляция данных пользователей
        )

    def recall(self, user_id, query, top_k=5, category=None):
        """Семантический поиск с metadata-фильтром."""
        filters = {"user_id": {"$eq": user_id}}
        if category:
            filters["category"] = {"$eq": category}

        results = self.index.query(
            vector=self._embed(query)[0],
            top_k=top_k,
            filter=filters,
            namespace=f"user_{user_id}",
            include_metadata=True
        )
        return [(m['metadata']['fact'], m['score']) for m in results['matches']]

    def get_stats(self):
        """Статистика индекса."""
        return self.index.describe_index_stats()

Mem0 (Memory Zero) — это не просто векторная база данных, а интеллектуальный слой памяти, который автоматически извлекает, структурирует и обновляет факты из диалогов. В отличие от ChromaDB/Pinecone, куда вы сами решаете что сохранять, Mem0 анализирует каждое сообщение с помощью LLM и решает: является ли это новым фактом? Обновляет ли он существующий? Стоит ли его забыть? Это память «человеческого» уровня — не дословное запоминание, а извлечение сути.

Mem0 поддерживает несколько бэкендов: локальный (Qdrant), managed cloud, и кастомные. Метод add() принимает сырое сообщение, внутренний LLM извлекает structured memory, а search() находит релевантные факты по семантическому запросу. Идеально для персонализированных агентов, которые должны помнить предпочтения пользователя, контекст проекта, прошлые решения.

pip install mem0ai

from mem0 import Memory

class Mem0AgentMemory:
    """Интеллектуальная память на Mem0 с автоматическим извлечением фактов."""

    def __init__(self, user_id, llm_provider="openai"):
        self.user_id = user_id

        # Конфигурация: бэкенд Qdrant (бесплатный локальный)
        config = {
            "vector_store": {
                "provider": "qdrant",
                "config": {
                    "host": "localhost",
                    "port": 6333,
                }
            },
            "llm": {
                "provider": llm_provider,
                "config": {
                    "model": "gpt-4o-mini",  # дешёвая модель для извлечения фактов
                }
            },
            "embedder": {
                "provider": "openai",
                "config": {
                    "model": "text-embedding-3-small"
                }
            }
        }
        self.memory = Memory.from_config(config)

    def add_from_message(self, message, role="user"):
        """Mem0 сам решит, является ли сообщение фактом."""
        result = self.memory.add(
            messages=[{"role": role, "content": message}],
            user_id=self.user_id
        )
        return result  # список извлечённых/обновлённых фактов

    def search(self, query, limit=5):
        """Семантический поиск релевантных воспоминаний."""
        results = self.memory.search(
            query=query,
            user_id=self.user_id,
            limit=limit
        )
        return [(r['memory'], r.get('score', 0)) for r in results]

    def get_all(self):
        """Получить все факты о пользователе."""
        return self.memory.get_all(user_id=self.user_id)

    def delete(self, memory_id):
        """Удалить конкретный факт."""
        self.memory.delete(memory_id=memory_id)

# Пример использования Mem0 в агенте
mem = Mem0AgentMemory(user_id="user_123")

# Агент общается — Mem0 сам извлекает факты
mem.add_from_message("Меня зовут Алексей, я Python-разработчик из Москвы")
# → Извлечёт: {"name": "Алексей", "role": "Python developer", "location": "Москва"}

mem.add_from_message("Я переехал в Санкт-Петербург")
# → Обновит location на "Санкт-Петербург", а не создаст дубликат

# Поиск контекста для нового запроса
context = mem.search("Где живёт Алексей и чем занимается?")
# → [("Алексей — Python разработчик из Санкт-Петербурга", 0.94)]

Выбор стека памяти зависит от масштаба, требований к приватности и бюджета. Для прототипа или личного агента: ChromaDB (persistent) + Redis + Mem0. Вы получаете полноценную трёхуровневую память за $0 на инфраструктуру, только платите за API-вызовы. Для стартапа с 1K–10K пользователей: Pinecone (serverless) заменяет ChromaDB, давая managed масштабирование без DevOps. Для enterprise с жёсткими требованиями к данным: self-hosted Qdrant на Kubernetes + собственный embedding-сервер (например, text-embeddings-inference от HuggingFace).

Важнейший архитектурный принцип: не смешивайте уровни. Рабочая память — это всегда RAM инференс-сервера. Краткосрочная — быстрая key-value БД (Redis/Dragonfly). Долгосрочная — векторная БД. Попытка использовать ChromaDB как краткосрочную память приведёт к latency >100ms на каждый вызов, что убьёт UX. И наоборот — хранение всей истории в контекстном окне быстро приведёт к банкротству на API-токенах.

╔══════════════════════════════════════════════════════════════════════╗
║           🧠  КАКУЮ ВЕКТОРНУЮ БД ВЫБРАТЬ ДЛЯ ПАМЯТИ?               ║
╠═══════════╦═══════════════╦══════════════╦══════════════╦════════════╣
║ Решение   ║ Тип           ║ Масштаб       ║ Стоимость    ║ Когда      ║
╠═══════════╬═══════════════╬══════════════╬══════════════╬════════════╣
║ ChromaDB  ║ Self-hosted   ║ До 1M вект.  ║ $0           ║ Прототип   ║
║ Qdrant    ║ Self-hosted   ║ До 100M      ║ $0 (OSS)     ║ Production ║
║ Pinecone  ║ Managed Cloud ║ Не ограничен ║ ~$70/мес     ║ Стартап    ║
║ Weaviate  ║ Self/Managed  ║ До 1B        ║ $0 / $25/мес ║ Гибрид     ║
║ Mem0      ║ Слой над БД   ║ Зависит от БД║ $0 (OSS)     ║ Факты      ║
╚═══════════╩═══════════════╩══════════════╩══════════════╩════════════╝

# Финальная архитектура памяти агента:
#
#  User Message
#       │
#       ▼
#  ┌─────────────────────────────────────────┐
#  │ 1. Mem0 / Ручное извлечение фактов      │  ← извлекаем суть
#  └──────────────┬──────────────────────────┘
#                 ▼
#  ┌─────────────────────────────────────────┐
#  │ 2. Long-term search (Pinecone/ChromaDB) │  ← ищем похожий опыт
#  └──────────────┬──────────────────────────┘
#                 ▼
#  ┌─────────────────────────────────────────┐
#  │ 3. Short-term load (Redis session)      │  ← подгружаем историю
#  └──────────────┬──────────────────────────┘
#                 ▼
#  ┌─────────────────────────────────────────┐
#  │ 4. Working Memory (LLM context)         │  ← всё попадает в окно
#  └──────────────┬──────────────────────────┘
#                 ▼
#  ┌─────────────────────────────────────────┐
#  │ 5. LLM Response + Save to all tiers     │  ← сохраняем опыт
#  └─────────────────────────────────────────┘