Графовая когнитивная память для долгоживущих ИИ-агентов: архитектура на SQLite с кодом и гибридным поиском

Почему ваш агент страдает амнезией?

Ваш ИИ-агент помнит последние десять сообщений. Может, двадцать. Потом контекстное окно заканчивается, и он начинает вести себя как золотая рыбка - спрашивает то, что вы уже обсуждали час назад. Стандартный RAG? Это архив, а не память. Он не знает, что вчера важнее позавчера, не видит связей между разговорами о бюджете и новом фиче. Агент живет в плоском мире без времени и контекста.

Долгоживущий агент - это не тот, кто много помнит. Это тот, кто помнит правильно. Связывает идеи, забывает мусор, вспоминает нужное в нужный момент. Как человеческий мозг, а не как гугл-док.

Графы, SQLite и немного магии забывания

Решение - графовая когнитивная память. Каждое воспоминание - узел. Связи между ними - ребра. Но хранить это в Neo4j для маленького агента - overkill. Мы возьмем SQLite. Да, тот самый, что в вашем телефоне. На 2026 год SQLite 3.47.1 (или новее) тянет гигабайты данных, умеет в полнотекстовый поиск FTS5 и работает быстрее, чем вы думаете.

Три кита нашей архитектуры:

• Графовая структура - связи "причина-следствие", "ссылается на", "противоречит".
• Гибридный поиск - FTS5 для ключевых слов + векторные эмбеддинги (используем sentence-transformers/all-MiniLM-L12-v2 или новее, на 2026 актуальна версия 3.x) для смысла.
• Кривая забывания Эббингауза - вес воспоминания = exp(-Δt / λ). Δt - время с последнего доступа, λ - коэффициент затухания (настраиваемый).

Режем код: от схемы до гибридного поиска

Теория - это скучно. Вот схема БД, которая работает сегодня. Копируйте, не стесняйтесь.

1 Схема базы: память как граф

-- memories.sql
-- SQLite 3.47.1+ с включенным FTS5 и расширением JSON1

CREATE TABLE IF NOT EXISTS memories (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    content TEXT NOT NULL,               -- исходный текст
    embedding BLOB,                      -- векторное представление
    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
    last_accessed DATETIME DEFAULT CURRENT_TIMESTAMP,
    access_count INTEGER DEFAULT 1,
    importance REAL DEFAULT 1.0,         -- пользовательская важность
    decay_lambda REAL DEFAULT 86400.0    -- λ в секундах (24 часа)
);

-- Граф связей: от родителя к ребенку с типом связи
CREATE TABLE IF NOT EXISTS memory_edges (
    source_id INTEGER NOT NULL,
    target_id INTEGER NOT NULL,
    relationship TEXT NOT NULL,          -- 'context', 'contradicts', 'supports', 'references'
    strength REAL DEFAULT 1.0,
    created DATETIME DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (source_id, target_id, relationship),
    FOREIGN KEY (source_id) REFERENCES memories(id) ON DELETE CASCADE,
    FOREIGN KEY (target_id) REFERENCES memories(id) ON DELETE CASCADE
);

-- Виртуальная таблица FTS5 для быстрого текстового поиска
CREATE VIRTUAL TABLE IF NOT EXISTS memories_fts USING fts5(
    content,
    content='memories',                  -- внешняя таблица
    content_rowid='id'                   -- связь по rowid
);

-- Триггеры для синхронизации FTS
CREATE TRIGGER IF NOT EXISTS memories_ai AFTER INSERT ON memories BEGIN
    INSERT INTO memories_fts(rowid, content) VALUES (new.id, new.content);
END;

CREATE TRIGGER IF NOT EXISTS memories_ad AFTER DELETE ON memories BEGIN
    INSERT INTO memories_fts(memories_fts, rowid, content) VALUES('delete', old.id, old.content);
END;

-- Индексы для скорости
CREATE INDEX IF NOT EXISTS idx_memories_time ON memories(timestamp);
CREATE INDEX IF NOT EXISTS idx_memories_accessed ON memories(last_accessed);
CREATE INDEX IF NOT EXISTS idx_edges_source ON memory_edges(source_id);
CREATE INDEX IF NOT EXISTS idx_edges_target ON memory_edges(target_id);

Видите этот decay_lambda? Это наша реализация кривой забывания. По умолчанию - 24 часа. Если воспоминание не трогали сутки, его вес упадет в e раз (~2.718). Настраивайте под задачу: для технических спецификаций ставьте 604800 (неделя), для разговоров - 3600 (час).

2 Класс памяти: Python, который не стыдно показать

# memory_graph.py
import sqlite3
import json
import numpy as np
from datetime import datetime, timedelta
from sentence_transformers import SentenceTransformer
import hashlib

class MemoryGraph:
    """Графовая когнитивная память для ИИ-агентов"""
    
    def __init__(self, db_path=':memory:', model_name='sentence-transformers/all-MiniLM-L12-v2'):
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self.conn.row_factory = sqlite3.Row
        self.model = SentenceTransformer(model_name)
        self.dim = 384  # для MiniLM-L12, актуально на 2026
        self._init_db()
    
    def _init_db(self):
        with open('memories.sql', 'r') as f:
            schema = f.read()
        self.conn.executescript(schema)
        
        # Проверяем, есть ли столбец embedding и его размер
        cursor = self.conn.execute("PRAGMA table_info(memories)")
        columns = [col[1] for col in cursor.fetchall()]
        if 'embedding' not in columns:
            self.conn.execute('ALTER TABLE memories ADD COLUMN embedding BLOB')
        
        # Для векторного поиска нужен индекс (используем простой L2, для продакшена - FAISS)
        # SQLite не умеет в векторные индексы, поэтому ищем по полному сканированию для малых объемов
        # Для больших объемов - см. расширение vectorlite или вынос в специализированную БД
    
    def add_memory(self, content, importance=1.0, decay_lambda=86400.0, parent_id=None, relationship='context'):
        """Добавить воспоминание, возможно, с связью с существующим"""
        embedding = self.model.encode(content).astype(np.float32).tobytes()
        
        cursor = self.conn.execute("""
            INSERT INTO memories (content, embedding, importance, decay_lambda)
            VALUES (?, ?, ?, ?)
        """, (content, embedding, importance, decay_lambda))
        memory_id = cursor.lastrowid
        
        if parent_id:
            self.conn.execute("""
                INSERT OR IGNORE INTO memory_edges (source_id, target_id, relationship)
                VALUES (?, ?, ?)
            """, (parent_id, memory_id, relationship))
        
        self.conn.commit()
        return memory_id
    
    def _calculate_recency_weight(self, memory_row):
        """Вес по кривой Эббингауза"""
        last_accessed = datetime.fromisoformat(memory_row['last_accessed'])
        delta = (datetime.now() - last_accessed).total_seconds()
        lambda_val = memory_row['decay_lambda']
        return np.exp(-delta / lambda_val) * memory_row['importance']
    
    def hybrid_search(self, query, limit=10, fts_weight=0.4, vector_weight=0.6):
        """Гибридный поиск: FTS5 + векторная схожесть"""
        # 1. Полнотекстовый поиск
        fts_results = {}
        cursor = self.conn.execute("""
            SELECT rowid, bm25(memories_fts) as score
            FROM memories_fts
            WHERE memories_fts MATCH ?
            ORDER BY score
            LIMIT ?
        """, (query, limit*2))  # Берем в 2 раза больше для последующего слияния
        
        for idx, row in enumerate(cursor):
            fts_results[row['rowid']] = 1.0 / (60 + idx)  # Reciprocal Rank Fusion, k=60
        
        # 2. Векторный поиск
        query_embedding = self.model.encode(query)
        vector_results = {}
        
        cursor = self.conn.execute("SELECT id, embedding FROM memories")
        # ВНИМАНИЕ: полное сканирование! Для >10К воспоминаний нужно кэшировать или использовать FAISS
        for row in cursor:
            if row['embedding']:
                emb = np.frombuffer(row['embedding'], dtype=np.float32)
                similarity = np.dot(query_embedding, emb) / (np.linalg.norm(query_embedding) * np.linalg.norm(emb))
                vector_results[row['id']] = max(similarity, 0)  # косинусная схожесть
        
        # Сортируем векторные результаты
        sorted_vector = sorted(vector_results.items(), key=lambda x: x[1], reverse=True)[:limit*2]
        for idx, (mem_id, score) in enumerate(sorted_vector):
            vector_results[mem_id] = 1.0 / (60 + idx)  # RRF для векторных
        
        # 3. Слияние результатов
        all_ids = set(fts_results.keys()) | set(vector_results.keys())
        combined_scores = {}
        
        for mem_id in all_ids:
            fts_score = fts_results.get(mem_id, 0)
            vec_score = vector_results.get(mem_id, 0)
            combined = fts_weight * fts_score + vector_weight * vec_score
            
            # Получаем строку для расчета веса забывания
            cursor = self.conn.execute("SELECT * FROM memories WHERE id = ?", (mem_id,))
            mem = cursor.fetchone()
            if mem:
                recency_weight = self._calculate_recency_weight(mem)
                combined *= recency_weight
                combined_scores[mem_id] = combined
        
        # 4. Обновляем last_accessed для найденных воспоминаний
        for mem_id in list(combined_scores.keys())[:limit]:
            self.conn.execute("""
                UPDATE memories 
                SET last_accessed = CURRENT_TIMESTAMP, access_count = access_count + 1
                WHERE id = ?
            """, (mem_id,))
        self.conn.commit()
        
        # 5. Возвращаем топ-N
        sorted_memories = sorted(combined_scores.items(), key=lambda x: x[1], reverse=True)[:limit]
        result = []
        for mem_id, score in sorted_memories:
            cursor = self.conn.execute("SELECT * FROM memories WHERE id = ?", (mem_id,))
            mem = cursor.fetchone()
            result.append(dict(mem))
        
        return result
    
    def get_context_for_agent(self, query, token_budget=4000):
        """Получить контекст для агента, укладываясь в бюджет токенов"""
        memories = self.hybrid_search(query, limit=20)
        context = []
        total_tokens = 0
        
        for mem in memories:
            # Простая оценка токенов (~4 токена на слово для русского)
            mem_tokens = len(mem['content'].split()) * 4
            if total_tokens + mem_tokens > token_budget:
                break
            context.append(mem['content'])
            total_tokens += mem_tokens
            
            # Добавляем связанные воспоминания (графовый контекст)
            cursor = self.conn.execute("""
                SELECT m.content 
                FROM memory_edges e
                JOIN memories m ON e.target_id = m.id
                WHERE e.source_id = ? AND e.relationship = 'context'
                LIMIT 2
            """, (mem['id'],))
            for related in cursor:
                rel_tokens = len(related[0].split()) * 4
                if total_tokens + rel_tokens > token_budget:
                    break
                context.append(f"[Связано] {related[0]}")
                total_tokens += rel_tokens
        
        return "\n\n".join(context), total_tokens
    
    def forget_old_memories(self, threshold=0.01):
        """Удалить воспоминания с весом ниже порога (автоочистка)"""
        cursor = self.conn.execute("SELECT * FROM memories")
        to_delete = []
        
        for row in cursor:
            weight = self._calculate_recency_weight(row)
            if weight < threshold and row['access_count'] < 3:  # Мало использовались
                to_delete.append(row['id'])
        
        if to_delete:
            placeholders = ','.join('?' * len(to_delete))
            self.conn.execute(f"DELETE FROM memories WHERE id IN ({placeholders})", to_delete)
            self.conn.commit()
            return len(to_delete)
        return 0
    
    def close(self):
        self.conn.close()

Это рабочий код. Скопируйте, запустите. Только не забудьте pip install sentence-transformers numpy на актуальных версиях 2026 года (проверьте, что у вас 3.x, там API мог поменяться).

Интеграция: как встроить память в агента

Теперь самая интересная часть. Ваш агент на OpenAI API, Claude или локальной Llama 3.3 (актуально на 2026) должен общаться с этой памятью.

# agent_with_memory.py
from memory_graph import MemoryGraph
import openai  # или anthropic, или llama_cpp

class AgentWithCognitiveMemory:
    def __init__(self):
        self.memory = MemoryGraph('agent_memory.db')
        self.conversation_id = None  # для группировки диалогов
        
    def chat(self, user_input):
        # 1. Ищем релевантные воспоминания
        context, tokens = self.memory.get_context_for_agent(user_input)
        
        # 2. Формируем промпт с контекстом
        prompt = f"""Ты ИИ-агент с долговременной памятью.
        
        Релевантный контекст из прошлых разговоров:
        {context}
        
        Текущий запрос: {user_input}
        
        Ответь, учитывая контекст выше. Если контекст нерелевантен, игнорируй его."""
        
        # 3. Вызов модели (пример для OpenAI GPT-4.5 Mini, актуально на 2026)
        response = openai.chat.completions.create(
            model="gpt-4.5-mini",
            messages=[{"role": "user", "content": prompt}],
            max_tokens=1000
        )
        
        answer = response.choices[0].message.content
        
        # 4. Сохраняем оба высказывания в память со связью
        user_mem_id = self.memory.add_memory(
            content=user_input, 
            importance=0.8, 
            decay_lambda=7200  # 2 часа для разговоров
        )
        
        answer_mem_id = self.memory.add_memory(
            content=answer,
            importance=0.9,
            decay_lambda=7200,
            parent_id=user_mem_id,
            relationship='response_to'
        )
        
        # 5. Периодическая очистка старых воспоминаний
        if self.memory.conn.total_changes % 100 == 0:  # каждые 100 изменений
            forgotten = self.memory.forget_old_memories()
            print(f"[Память] Забыто {forgotten} воспоминаний")
        
        return answer

Теперь ваш агент помнит. Не все подряд, а важное. И забывает мусор. Как человек.

Ошибки, которые сломают вашу память

Видел такое сто раз. Не повторяйте.

Ошибка 1: Векторный поиск по всей таблице

Мой код делает SELECT id, embedding FROM memories и сканирует всё. Для 1000 воспоминаний - нормально. Для 100000 - ад. Решение:

• Кэшируйте эмбеддинги в numpy массив при запуске (если память read-heavy).
• Используйте FAISS индекс поверх эмбеддингов. Сохраняйте его на диск, обновляйте при добавлении.
• Или переходите на специализированную векторную БД, когда превысите 50К записей. Но тогда теряется простота SQLite.

Ошибка 2: Жесткие пороги забывания

Удалять всё с весом < 0.01 - опасно. Вдруг это редкое, но важное воспоминание? Добавьте страховку:

# Вместо простого порога
def should_forget(memory_row, threshold=0.01):
    weight = calculate_weight(memory_row)
    if weight < threshold:
        # Никогда не забывать то, что помечено как важное
        if memory_row['importance'] > 2.0:
            return False
        # Никогда не забывать то, что связано с многими другими узлами
        cursor = conn.execute("SELECT COUNT(*) FROM memory_edges WHERE source_id=? OR target_id=?", 
                             (memory_row['id'], memory_row['id']))
        connections = cursor.fetchone()[0]
        if connections > 5:
            return False
    return True

Ошибка 3: Игнорирование транзакций

SQLite без proper transaction handling ломается при конкурентном доступе. Если ваш агент многопоточный:

# Вместо прямого execute
with self.conn:
    self.conn.execute("INSERT ...")
    # Автоматический commit или rollback

# Или используйте WAL режим
self.conn.execute("PRAGMA journal_mode=WAL;")  # Позволяет читать и писать одновременно

Больше ловушек смотрите в статье про KV-cache - там много пересечений.

Что дальше? Когда всё это взорвется

Эта архитектура работает для агентов с десятками тысяч воспоминаний. Дальше будет больно. Но боль - это рост.

На 50К+ воспоминаний:

1. Векторный поиск переезжает в FAISS или специализированную БД.
2. Графовые запросы (поиск путей, кластеризация) начинают тормозить. Нужны предварительно вычисленные индексы связности.
3. SQLite файл может превысить 10ГБ. Переходите на PostgreSQL с расширением pgvector, но теряете embedded-природу.

Но главное - вы теперь думаете о памяти агента как о когнитивной системе, а не как о логе чата. Это меняет всё.

Хотите пойти дальше? Добавьте эмоциональный вес воспоминаниям (положительные/отрицательные). Или временные метки для сезонных паттернов (агент помнит, что в декабре вы всегда спрашиваете про бюджет). Или полноценные knowledge graphs с онтологиями.

Но начните с этого кода. Запустите. Посмотрите, как ваш агент из золотой рыбки превращается в слона - того, который никогда не забывает.

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