Четырехуровневая архитектура памяти для ИИ-агентов с информацией о происхождении данных
Разделение краткосрочной и долгосрочной памяти, четыре уровня (буфер, эпизодический, семантический, процедурный), варианты хранения, асинхронная запись и меры защиты от отравления памяти.
Источники: All You Want For Agentic Memory Design
Представьте агента, который терпит неудачу на четырнадцатом шаге длинного рабочего процесса. Вина не в самой модели — агент просто больше не помнит, что уже сделал. Неудача следует по знакомой схеме: вызовы инструментов работают корректно, подсказки достоверны, но как только диалог превышает определённую длину, агент начинает заходить в замкнутый круг. Он повторяет уже заданные вопросы, запрашивает данные, которые у него уже есть, и противоречит сам себе между шестым и двенадцатым шагами, словно первые пять шагов вообще не происходили.
Замена модели на другую редко меняет ситуацию. Проблема заключается в инфраструктуре, окружающей модель, а точнее — в отсутствии настоящей архитектуры памяти. Вставка последних N сообщений в подсказку работает для чат-бота, но теряет эффективность, когда агенту необходимо выполнить двадцатишаговый рабочий процесс и запомнить на семнадцатом шаге то, что он решил на третьем.
Одним из решений этой проблемы является четырехуровневая стек-структура памяти, при которой за каждым уровнем располагается соответствующее хранилище. Далее рассматривается функция каждого уровня, как команды GitHub, Anthropic и Berkeley описывают свои версии такой структуры, а также примеры кода на Python для каждой её части.
Окно контекста — это не память
Окно контекста представляет собой рабочую память: оперативную память, а не жесткий диск. Каждый токен в нем имеет свою стоимость. Заполнение всей истории разговора окном объемом 128 Кб может казаться безопасным, пока вы не начнете использовать агента, требующего много ресурсов. Тридцатишаговый процесс с вызовами инструментов, наблюдениями и следами логических рассуждений быстро исчерпывает лимит в 100 Кб токенов, и когда окно заполняется, самый старый контент удаляется с самого начала, где обычно находятся первоначальные требования пользователя и описание задачи.
Ещё более серьезной проблемой является то, что она скрытнее. Исследование ученых из Стэнфорда и Калифорнийского университета в Беркли, опубликованное в 2024 году в журнале ACL под названием «Потеряно посредине: как языковые модели используют длинные контексты», показало, что производительность ЯИМ следует кривой в форме буквы U в зависимости от длины контекста. Модели показывают хорошие результаты с информацией, находящейся в начале и в конце контекста, но значительно хуже с информацией, расположенной в середине. Этот эффект наблюдался у различных архитектур, размеров моделей и длин контекста; тестировались окна длиной от 4K до 32K токенов. Последующее исследование, проведенное в 2025 году в MIT, предложило математическое объяснение этого феномена, утверждая, что маскировка каузального внимания и кодирование положения способствуют снижению веса внимания к токенам, находящимся далеко от начала или конца последовательности.
Для агентов последствия очевидны. Даже если всё соответствует требованиям, модель не будет равномерно уделять внимание всем элементам. Критическое ограничение на уровне 30 тысяч элементов в контексте объемом 60 тысяч токенов действительно может сделаться практически незаметным.
Модель забывает, потому что система поступает к ней больше информации, чем она может надежно обработать, и нет никакого контроля за тем, что попадает в её рабочее пространство.
Исходное разделение: краткосрочная и долгосрочная память
До появления четырехуровневой таксономии существует более простое разделение, которое большинство инженеров имеют в голове. Память в ИИ-системах делят на краткосрочную и долгосрочную, и понимание того, чего каждая из них не может делать, становится основой для более детальной архитектуры.
Краткосрочная память ограничена текущей сессией и существует только во время одного взаимодействия. Она хранит то, что необходимо агенту прямо сейчас: состояние текущего задания, этап, на котором он находится, результаты последнего использования инструмента и информацию о процессе рассуждений. На практике это окно контекста, которое крайне динамично, постоянно обновляется, имеет высокую стоимость хранения за токен и очищается по окончании сессии.
В документации LangChain это хорошо описано: краткосрочная память считается местом хранения истории разговора, результатов работы инструментов и временного состояния рассуждений. Однако не упоминается, что именно она первой страдает при масштабировании. В рабочем процессе, включающем двадцать шагов выполнения задачи, накапливаются промежуточные результаты, размышления, найденная информация и ответы инструментов, которые конкурируют за ограниченное пространство памяти. Это похоже на попытку запомнить сложный план в голове, пока кто-то постоянно прерывает вас новыми фактами.
Краткосрочная память имеет два практических компонента:
- Состояние задачи — это рабочий отчет о прогрессе: текущий шаг, незавершенные подзадачи, статус выполнения инструментов и координация между агентами. Агент, отвечающий за планирование поездок, сохраняет информацию о пункте назначения, лимитах бюджета, рассмотренных рейсах и отобранных отелях, чтобы не повторять работу с второго шага при переходе к восьмому.
- Промежуточные результаты — это продукты размышлений в процессе работы: результаты использования инструментов, сгенерированные краткие сводки, выводы SQL, фрагменты поисковых результатов и заметки в временной памяти. Инженерная команда Anthropic выявила закономерность в рабочих системах: агенты, которые сохраняют заметки вне окна контекста и загружают их позже, демонстрируют лучшие результаты при выполнении длительных задач по сравнению с агентами, пытающимися хранить всё в контексте. Сам процесс ведения заметок является стратегией краткосрочной памяти.
Долгосрочная память сохраняется между сессиями. Она позволяет системе вспоминать, что пользователю нравятся краткие ответы, что прошлая неделя развертывания провалилась из-за проблем с миграцией или что схема базы данных изменилась в январе. LangChain определяет её как память, которая сохраняется между потоками и сессиями; на практике это означает, что для хранения таких данных требуется база данных, а не окно контекста.
Существует два основных вида такой памяти:
- Предпочтения пользователя — это сигналы персонализации, которые накапливаются со временем: стиль написания, тон, уровень технической подготовки, выбор инструментов и привычки форматирования. Помощник по написанию, который помнит, что вы всегда хотите использовать markdown и требует высокого уровня технической информации, становится полезным в том смысле, в котором никакая безсостоянийная система не сможет сравниться, каким бы хорошим ни был её модель.
Многие команды допускают те же ошибки. Они создают краткосрочную память, заполняя окно контекста, и долгосрочную память, создавая один хранилище векторов. Оба подхода неверны или, в лучшем случае, неполны. Первый не справляется в масштабе по вышеуказанным причинам. Второй теряет эффективность, поскольку поиск по сходству — это плохой способ получения записей сессий, предпочтений или правил поведения, к тому же каждый тип долгосрочной памяти требует своего бэкенда. Четырехуровневая архитектура представляет собой усовершенствованную версию разделения на краткосрочную и долгосрочную память, при которой каждый тип памяти соотносится с необходимым хранилищем.
Четыре типа памяти, одна таксономия
Команды, разрабатывающие промышленные ИИ-системы, независимо друг от друга пришли к одной и той же таксономии. Названия могут отличаться, но структура остаётся прежней.
Проект MemGPT, представленный Чарльзом Паккером и его коллегами из Калифорнийского университета в Беркли в 2023 году, популяризировал идею рассматривать окно контекста больших языковых моделей как оперативную память, которая ограничена по объёму и ценна. Операционные системы расширяют объём памяти с помощью страницирования и хранения на диске, а фреймворки агентов могут аналогичным образом имитировать долгосрочную память, перемещая более старый контекст в более объёмное хранилище и восстанавливая его по мере необходимости. Эта идея позже стала основой для Letta.
Команда LangMem от LangChain, фреймворк ReMem от Google DeepMind и промышленная архитектура Mem0 описывают все варианты четырёх типов:
- Рабочая память — это всё, что находится в текущем контексте: текущие этапы разговора, загруженные документы и результаты работы инструментов. Она работает быстро, имеет ограничения по объему и требует значительных ресурсов на обработку каждого токена.
- Эпизодическая память хранит записи о прошлых взаимодействиях, например, что агент делал для определенного пользователя в прошлый вторник и каков был результат. Она организована по хронологии и ограничена текущей сессией, обычно хранится в быстром хранилище типа key-value или SQL-базе данных, и данные из нее извлекаются по принципу свежести или семантического сходства.
- Семантическая память содержит общие факты, знания в определенной области и информацию, полученную в процессе работы. Это слой RAG, который задействуется при сравнении с помощью эмбеддингов каждый раз, когда задача требует знаний из определенной области или ранее полученных данных.
Эти типы соответствуют принципам когнитивной науки, что объясняет, почему команды приходят к ним. Они соответствуют системам эпизодической, семантической и процедурной памяти в нейронауке о человеке, а рабочая память в качестве окна контекста — это не просто метафора. Это наиболее близкий функциональный аналог.
Выбор хранилища: SQL, векторное или графовое
В какой базе данных следует хранить память? Прежде чем писать код, честный ответ заключается в том, что это зависит от типа памяти.
Векторные базы данных (Weaviate, Pinecone, Qdrant, Chroma) подходят для реализации семантической памяти, поскольку они обеспечивают быстрый поиск по сходству в эмбеддингах даже при больших объемах данных, причем сейчас большинство из них предлагает гибкий метод поиска, сочетающий алгоритм BM25 с векторами, что значительно превосходит чистый векторный поиск для текста. У них отсутствует возможность временных рассуждений и выполнения реляционных запросов, к тому же два противоречивых факта с практически идентичными эмбеддингами могут ввести в заблуждение. Не храните в них эпизодические записи и структурированные аудит-трейлы.
Базы данных SQL, такие как PostgreSQL и SQLite, подходят для хранения эпизодической памяти. Запросы структурированы по идентификатору пользователя, сессии, типу задачи или временной метке; объем записей невелик, а гарантии ACID имеют важное значение при фиксации действий агента и времени их выполнения. PostgreSQL с расширением pgvector позволяет использовать реляционные запросы и векторный поиск в одной системе. При количестве записей менее десяти миллионов это практичный выбор благодаря знакомым операциям, недорогому хостингу и возможности использования SQL для отладки.
Графовые базы данных, такие как Neo4j, применимы в системах памяти только тогда, когда у эпизодических или семантических данных сложная реляционная структура, требующая многократных запросов — например, определения того, кто доложил кому три месяца назад при создании документа. Простые записи о задачах агентов не подходят для этого. Многие команды обращаются к графовым базам данных, но в итоге оказывается, что им нужен SQL.
В большинстве случаев развертывания используется реляционное хранилище (SQL или PostgreSQL) для хранения эпизодов и профилей пользователей, векторное хранилище для семантических поисков и внутренний состояние процесса для работы с контекстом; комбинация PostgreSQL и pgvector выбирается теми командами, которые хотят снизить операционную нагрузку. Выбирайте уровень хранилища в зависимости от функций, которые должен выполнять тип памяти, и не векторизируйте всё без необходимости.
Уровень 1: буферная память с обобщением
Самая простая форма — это очередь с ограничением по количеству сообщений:
from collections import deque
from typing import List, Dict
class BufferMemory:
def __init__(self, max_turns: int = 10):
# Each turn = one user message + one assistant message
self.history: deque = deque(maxlen=max_turns * 2)
def add_message(self, role: str, content: str):
self.history.append({"role": role, "content": content})
def get_context(self) -> List[Dict]:
return list(self.history)
def token_estimate(self) -> int:
total_chars = sum(len(m["content"]) for m in self.history)
return total_chars // 4 # rough approximation
def clear(self):
self.history.clear()
Этот подход работает при кратких обменах сообщениями. При более длительных разговорах старые сообщения просто исчезают в конце, из-за чего агент теряет нить разговора. Лучший способ, который LangMem называет методом «горячего пути», заключается в сведении старых сообщений к краткому резюме перед их удалением:
import os
from collections import deque
from typing import List, Dict
from openai import OpenAI
class SummarizingBufferMemory:
def __init__(self, max_turns: int = 8, summary_batch: int = 4):
self.client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
self.model = "deepseek/deepseek-v4-flash-0731"
self.recent: deque = deque(maxlen=max_turns * 2)
self.rolling_summary: str = ""
self.summary_batch = summary_batch
def add_message(self, role: str, content: str):
if len(self.recent) == self.recent.maxlen:
self._compress_oldest()
self.recent.append({"role": role, "content": content})
def _compress_oldest(self):
batch = [self.recent.popleft() for _ in range(min(self.summary_batch * 2, len(self.recent)))]
text = "\n".join(f"{m['role']}: {m['content']}" for m in batch)
response = self.client.chat.completions.create(
model=self.model,
max_tokens=300,
messages=[
{
"role": "system",
"content": "You compress conversation history. Reply with the summary only.",
},
{
"role": "user",
"content": (
"Summarize this conversation segment in 2-4 sentences. "
"Preserve any decisions made, constraints stated, and conclusions reached.\n\n"
f"{text}"
),
},
],
)
new_summary = response.choices[0].message.content.strip()
if self.rolling_summary:
self.rolling_summary = f"{self.rolling_summary} | {new_summary}"
else:
self.rolling_summary = new_summary
def get_context(self) -> List[Dict]:
context = []
if self.rolling_summary:
context.append({
"role": "system",
"content": f"[Prior conversation summary: {self.rolling_summary}]",
})
context.extend(list(self.recent))
return context
LangMem разделяет суммаризацию в реальном времени, которая выполняется прямо в процессе работы и увеличивает задержку для пользователя, от фоновой суммаризации, которая происходит после завершения сессии. Для приложений, ориентированных на пользователя, фоновый режим почти всегда предпочтительнее: качество кратких изложений улучшается благодаря возможности проанализировать всю переписку, и никто не вынужден ждать. Для автономных агентов, работающих без присмотра, режим в реальном времени приемлем, поскольку пользователь все равно не ждет.
Уровень 2: эпизодическая память, проверяющая собственные цитаты
Здесь многие команды допускают ошибки: они создают систему поиска, но так и не записывают, что на самом деле сделал агент.
В техническом блоге GitHub Copilot описывается паттерн, который стоит скопировать. Каждая запись в памяти сопровождается цитатами — конкретными адресами вроде путей к файлам, URL-адресов, названий инструментов и номеров строк, которые позволяют при необходимости проверить, сохраняется ли правда, хранящаяся в памяти. GitHub называет это проверкой в реальном времени. Когда агент загружает запись из памяти в начале выполнения задачи, он проверяет, существует ли упомянутый источник и продолжает ли он подтверждать изложенные факты. Если это не так, запись корректируется или удаляется, тем самым пул данных самостоятельно очищается без необходимости оффлайн-обработки.
GitHub зафиксировал ощутимый эффект: у агента для программирования с функцией памяти наблюдалось увеличение скорости объединения пул-запросов на 7%, с 83% до 90%.
import json
import sqlite3
from datetime import datetime, timedelta
from dataclasses import dataclass, field
@dataclass
class Episode:
task_type: str
user_query: str
outcome: str
tools_used: list
citations: list # source references for validation
duration_seconds: float
success: bool
session_id: str = ""
created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())
class EpisodicMemory:
EXPIRY_DAYS = 28 # GitHub Copilot's production default
def __init__(self, db_path: str = "episodes.db"):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self._init_schema()
def _init_schema(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS episodes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_type TEXT NOT NULL,
user_query TEXT,
outcome TEXT,
tools_used TEXT,
citations TEXT,
duration_s REAL,
success INTEGER,
session_id TEXT,
created_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_task_type ON episodes (task_type, success, created_at);
""")
self.conn.commit()
def record(self, ep: Episode):
self.conn.execute(
"""
INSERT INTO episodes
(task_type, user_query, outcome, tools_used, citations,
duration_s, success, session_id, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
ep.task_type, ep.user_query, ep.outcome,
json.dumps(ep.tools_used), json.dumps(ep.citations),
ep.duration_seconds, int(ep.success),
ep.session_id, ep.created_at,
),
)
self.conn.commit()
def recall_similar(self, task_type: str, limit: int = 3) -> list[Episode]:
cutoff = (datetime.utcnow() - timedelta(days=self.EXPIRY_DAYS)).isoformat()
cursor = self.conn.execute(
"""
SELECT task_type, user_query, outcome, tools_used, citations,
duration_s, success, session_id, created_at
FROM episodes
WHERE task_type = ? AND success = 1 AND created_at > ?
ORDER BY created_at DESC
LIMIT ?
""",
(task_type, cutoff, limit),
)
return [
Episode(
task_type=r[0], user_query=r[1], outcome=r[2],
tools_used=json.loads(r[3]), citations=json.loads(r[4]),
duration_seconds=r[5], success=bool(r[6]),
session_id=r[7], created_at=r[8],
)
for r in cursor.fetchall()
]
def validate_citations(self, episode: Episode, validator_fn) -> bool:
"""
validator_fn(citation: str) -> bool
Check if cited sources are still valid (file exists, URL responds, etc.)
Return False if any citation fails; the episode should be discarded.
"""
return all(validator_fn(c) for c in episode.citations)
Срок действия в 28 дней соответствует решению GitHub для производственной среды. Старые записи не удаляются, но теряют возможность извлечения; запрос LIMIT к индексу с фильтрацией по дате выполняется быстро и не требует фоновой обработки.
Уровень 3: семантическая память с использованием гибридного поиска Weaviate
Это уровень RAG. Большинство руководств описывают настройку и простую работу, тогда как настоящие производственные проблемы связаны с качеством извлечения информации и её происхождением.
Частой первой ошибкой является настройка семантической памяти в режиме только для чтения. Агенту, работающему длительное время, необходим режим чтения/записи: когда агент находит факт в ходе выполнения задачи, проверяет его и решает, что он стоит сохранить, его следует записать в семантическую память для будущих сессий. Второй ошибкой — полагаться исключительно на косинусное сходство. Гибридный поиск Weaviate сочетает совпадение ключевых слов BM25 с векторным сходством, что обычно даёт лучшие результаты по сравнению с использованием любого из методов в одиночку при извлечении информации из памяти агента.
В-третьих, и именно это создаёт возможность для отравления данных: записи зачастую не имеют информации об источнике. Каждой записи требуются источник, показатель надёжности и идентификатор сессии. Без этих элементов невозможен поиск с учётом степени доверия, устаревшие или сомнительные записи нельзя понизить в приоритете, а выявление ошибочного результата поиска превращается в угадывание.
import uuid as uuid_lib
from datetime import datetime
import weaviate
from weaviate.classes.config import Configure, Property, DataType, VectorDistances
from weaviate.classes.query import MetadataQuery, HybridFusion, Filter
def embed(text: str) -> list[float]:
"""
Swap in any embedding source: OpenAI, Cohere, sentence-transformers, etc.
Returns a normalized float vector.
"""
from sentence_transformers import SentenceTransformer
_model = SentenceTransformer("all-MiniLM-L6-v2")
return _model.encode(text, normalize_embeddings=True).tolist()
class WeaviateSemanticMemory:
COLLECTION = "AgentMemory"
def __init__(self, host: str = "localhost", port: int = 8080):
self.client = weaviate.connect_to_local(host=host, port=port)
self._ensure_collection()
def _ensure_collection(self):
if self.client.collections.exists(self.COLLECTION):
return
self.client.collections.create(
name=self.COLLECTION,
# We provide our own vectors; no built-in vectorizer needed.
# Swap to Configure.Vectorizer.text2vec_openai() if you prefer managed embedding.
vectorizer_config=Configure.Vectorizer.none(),
vector_index_config=Configure.VectorIndex.hnsw(
distance_metric=VectorDistances.COSINE
),
properties=[
Property(name="content", data_type=DataType.TEXT),
Property(name="task_type", data_type=DataType.TEXT),
Property(name="source", data_type=DataType.TEXT),
Property(name="citations", data_type=DataType.TEXT),
Property(name="session_id", data_type=DataType.TEXT),
Property(name="confidence", data_type=DataType.NUMBER),
Property(name="created_at", data_type=DataType.TEXT),
],
)
def store(
self,
content: str,
task_type: str = "",
source: str = "agent",
citations: str = "",
session_id: str = "",
confidence: float = 1.0,
) -> str:
collection = self.client.collections.get(self.COLLECTION)
doc_id = str(uuid_lib.uuid4())
collection.data.insert(
properties={
"content": content,
"task_type": task_type,
"source": source,
"citations": citations,
"session_id": session_id,
"confidence": confidence,
"created_at": datetime.utcnow().isoformat(),
},
vector=embed(content),
uuid=doc_id,
)
return doc_id
def retrieve_hybrid(
self,
query: str,
n_results: int = 5,
min_confidence: float = 0.6,
task_type: str = None,
) -> list[dict]:
collection = self.client.collections.get(self.COLLECTION)
# Filter by confidence floor and optionally by task type
confidence_filter = Filter.by_property("confidence").greater_or_equal(min_confidence)
if task_type:
active_filter = (
Filter.by_property("task_type").equal(task_type) & confidence_filter
)
else:
active_filter = confidence_filter
results = collection.query.hybrid(
query=query,
vector=embed(query),
limit=n_results,
fusion_type=HybridFusion.RELATIVE_SCORE,
filters=active_filter,
return_metadata=MetadataQuery(score=True),
)
return [
{
"content": obj.properties["content"],
"score": obj.metadata.score,
"confidence": obj.properties.get("confidence", 1.0),
"source": obj.properties.get("source", ""),
"citations": obj.properties.get("citations", ""),
"uuid": str(obj.uuid),
}
for obj in results.objects
]
def update_confidence(self, doc_id: str, new_confidence: float):
collection = self.client.collections.get(self.COLLECTION)
collection.data.update(
uuid=doc_id,
properties={"confidence": new_confidence},
)
def close(self):
self.client.close()
Во многих реализациях в функции retrieve_hybrid отсутствуют два важных элемента. Во-первых, гибридный конечный пункт Weaviate принимает как сырой текст запроса (используемый для алгоритма BM25), так и его эмбеддинг (используемый для векторного поиска). Точные совпадения по ключевым словам обеспечиваются первым элементом, а семантическая близость — вторым, причем их совместное использование дает лучшие результаты, чем любой из них по отдельности. Во-вторых, фильтр min_confidence, который исключает из контекста записи с низким уровнем доверия, независимо от их семантической близости. По сути, это мера безопасности, а не просто настройка релевантности.
Метод update_confidence обеспечивает механизм обратной связи. Если собственные рассуждения агента противоречат полученной записи в ходе выполнения задачи, необходимо снизить её уровень доверия; если проверка цитат подтверждает правильность записи, его следует повысить. Система должна учиться на практике.
Уровень 4: процедурная память и эволюция промптов
Именно этот слой чаще всего остается без внимания. Простая реализация представляет собой JSON-файл с правилами, который агент читает, соблюдает и дополняет после сбоев. Этот подход работает, но имеет свои ограничения: правила в отдельном файле конкурируют за пространство для хранения контекста и требуют явного извлечения.
Инженеры LangMem из LangChain предложили интересную идею, которую стоит использовать: в производственных условиях процедурная память реализуется через сам промпт системы. Она не хранится рядом с промптом и не извлекается из него — она сама является промптом. Каждый раз, когда сбой заставляет агента освоить новое правило поведения, сам промпт переписывается; это явление они называют оптимизацией промпта как процедурной памяти. Изученные моделями поведения превращаются в инструкции на естественном языке, которые модель читает в начале каждого взаимодействия.
Прощая версия с JSON-файлом — это то, с чего должны начать большинство команд:
import json
import os
import re
from datetime import datetime, timedelta, timezone
from openai import OpenAI
MODEL = "deepseek/deepseek-v4-flash-0731"
client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
class ProceduralMemory:
def __init__(self, rules_path: str = "procedural_rules.json"):
self.rules_path = rules_path
self.rules: list[dict] = self._load()
def _load(self) -> list[dict]:
try:
with open(self.rules_path) as f:
return json.load(f)
except FileNotFoundError:
return []
def add_rule(self, situation: str, action: str, reason: str, confidence: float = 1.0):
self.rules.append({
"situation": situation,
"action": action,
"reason": reason,
"confidence": float(confidence),
"added_at": datetime.now(timezone.utc).isoformat(),
"trigger_count": 0,
})
self._save()
def get_applicable_rules(self, context: str, min_confidence: float = 0.7) -> list[dict]:
relevant = []
ctx_lower = context.lower()
dirty = False
for rule in self.rules:
if rule.get("confidence", 1.0) < min_confidence:
continue
keywords = rule["situation"].lower().split()
if not keywords:
continue
hits = sum(1 for kw in keywords if kw in ctx_lower)
if hits >= max(1, len(keywords) // 3):
rule["trigger_count"] = rule.get("trigger_count", 0) + 1
relevant.append(rule)
dirty = True
if dirty:
self._save()
relevant.sort(key=lambda r: r.get("confidence", 1.0), reverse=True)
return relevant[:5]
def prune_stale(self, max_age_days: int = 60, min_triggers: int = 2):
cutoff = (datetime.now(timezone.utc) - timedelta(days=max_age_days)).isoformat()
self.rules = [
r for r in self.rules
if r.get("added_at", "") > cutoff or r.get("trigger_count", 0) >= min_triggers
]
self._save()
def _save(self):
tmp = f"{self.rules_path}.tmp"
with open(tmp, "w") as f:
json.dump(self.rules, f, indent=2)
os.replace(tmp, self.rules_path)
def _parse_json(text: str) -> dict:
text = text.strip()
fenced = re.search(r"```(?:json)?\s*(.*?)```", text, re.S)
if fenced:
text = fenced.group(1).strip()
return json.loads(text)
def extract_rule_from_failure(failure_trace: str, memory: ProceduralMemory):
response = client.chat.completions.create(
model=MODEL,
max_tokens=250,
response_format={"type": "json_object"},
messages=[
{
"role": "system",
"content": "You return only a JSON object. No prose, no code fences.",
},
{
"role": "user",
"content": (
"A task failed. Extract one behavioral rule to prevent this failure.\n"
"Return ONLY valid JSON with keys: situation, action, reason, "
"confidence (0.0-1.0)\n"
f"Failure trace:\n{failure_trace}"
),
},
],
)
try:
rule = _parse_json(response.choices[0].message.content)
except (json.JSONDecodeError, AttributeError, TypeError):
return None
if not all(k in rule for k in ("situation", "action", "reason")):
return None
memory.add_rule(
situation=str(rule["situation"]),
action=str(rule["action"]),
reason=str(rule["reason"]),
confidence=float(rule.get("confidence", 1.0)),
)
return rule
prune_stale борется с проблемой забывания на процедурном уровне. Правило, созданное после одной необычной ошибки и больше никогда не активированное, должно постепенно исчезать, в то время как правило, которое активируется регулярно, остается в силе независимо от своего возраста. Это грубый инструмент, но именно на этом уровне проще всего решить проблему забывания.
Как только агент начинает работать стабильно, следующим логичным шагом является включение правил, которым доверяют и которые используются часто, непосредственно в сам промпт системы. LangMem автоматизирует этот процесс с помощью цикла оптимизации промптов, а для более мелких систем вполне подходит ручная настройка.
Связывание слоев между собой
В следующем описании показано, как четыре слоя взаимодействуют в реальных условиях. Одно ограничение является неоспоримым: запись в память ни в коем случае не должна замедлять формирование ответа.
import json
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
@dataclass
class TaskResult:
task_type: str
query: str
outcome: str
tools_used: list
citations: list
duration_seconds: float
success: bool
new_facts: list[dict] = field(default_factory=list)
session_id: str = ""
class MemoryManager:
def __init__(
self,
weaviate_host: str = "localhost",
episodic_db: str = "episodes.db",
rules_path: str = "procedural_rules.json",
):
self.buffer = SummarizingBufferMemory(max_turns=8)
self.episodic = EpisodicMemory(db_path=episodic_db)
self.semantic = WeaviateSemanticMemory(host=weaviate_host)
self.procedural = ProceduralMemory(rules_path=rules_path)
self._pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="memory_write")
def build_context(self, query: str, task_type: str) -> list[dict]:
context: list[dict] = []
# Procedural rules first: they constrain behavior throughout the task
rules = self.procedural.get_applicable_rules(query)
if rules:
rules_text = "\n".join(
f"- When '{r['situation']}': {r['action']} (reason: {r['reason']})"
for r in rules
)
context.append({"role": "user", "content": f"[Behavioral rules:\n{rules_text}]"})
# Semantic facts: domain knowledge and past discoveries
facts = self.semantic.retrieve_hybrid(query, n_results=5, min_confidence=0.6)
if facts:
facts_text = "\n".join(f"- {f['content']}" for f in facts)
context.append({"role": "user", "content": f"[Relevant knowledge:\n{facts_text}]"})
# Past episodes: outcome templates for similar tasks
episodes = self.episodic.recall_similar(task_type, limit=3)
if episodes:
ep_text = "\n".join(
f"- Outcome: {e.outcome} (tools: {', '.join(e.tools_used)})"
for e in episodes
)
context.append({"role": "user", "content": f"[Past similar tasks:\n{ep_text}]"})
# Current conversation last: the model reads this most carefully
context.extend(self.buffer.get_context())
return context
def record_turn(self, role: str, content: str):
self.buffer.add_message(role, content)
def persist(self, result: TaskResult):
# Submit to thread pool and return immediately; never block the caller
self._pool.submit(self._persist_worker, result)
def _persist_worker(self, result: TaskResult):
ep = Episode(
task_type=result.task_type,
user_query=result.query,
outcome=result.outcome,
tools_used=result.tools_used,
citations=result.citations,
duration_seconds=result.duration_seconds,
success=result.success,
session_id=result.session_id,
)
self.episodic.record(ep)
for fact in result.new_facts:
self.semantic.store(**fact)
if not result.success:
extract_rule_from_failure(result.outcome, self.procedural)
def shutdown(self):
self._pool.shutdown(wait=True)
self.semantic.close()
Порядок элементов контекста в функции build_context выбран намеренно. Исследования, посвящённые смещению внимания к определённым позициям, подтверждают идею о том, что модели уделяют больше внимания началу и концу окна. Правила поведения размещаются вверху, поскольку они представляют собой ограничения, которые должны определять всё, что идёт далее. Текущий диалог находится внизу, так как именно его модели нужно учитывать в первую очередь для ответа. Семантические факты и прошлые сценарии располагаются между ними, обеспечивая ссылки без перегрузки текущей задачи.
Строка self._pool.submit(self._persist_worker, result) является самым важным инженерным решением в этом классе. Каждый вызов для записи возвращается мгновенно, поэтому ответ ЯИ-модели уже передаётся пользователю до того, как начнётся любая запись в базу данных.
Три решения, определяющих архитектуру
При наличии четырех слоев три архитектурных решения влияют на всё остальное, и многие команды допускают хотя бы одну ошибку.
Записи в режиме активной обработки или на фоне. Команда LangMem вложила значительные усилия в реализацию этой функции. Записи в режиме активной обработки происходят во время разговора, когда агент обнаруживает факт, достойный хранения, и выполняет запись непосредственно в процессе общения. Записи на фоне осуществляются после окончания сессии и предполагают анализ всего разговора с целью определения того, что следует сохранить. Режим активной обработки увеличивает задержку, поскольку каждый вызов semantic.store() в цикле ответа соответствует операции сохранения в базе данных, которую должен переждать пользователь. Записи на фоне улучшают качество воспроизведения информации, поскольку позволяют изучить темы и решения, принятые на протяжении всего разговора, и сохранять более подробные краткие изложения. Для прикладных программ, ориентированных на пользователей, режим записи на фоне практически всегда является оптимальным. Пул потоков в функции persist() обеспечивает значительную экономию ресурсов, избегая при этом полностью асинхронной архитектуры.
Поиск или резюмирование. Эти подходы решают разные задачи, и команды часто путают их. Поиск подразумевает нахождение соответствующего отрывка в большом корпусе данных. Запрос является конкретным, а корпус — большим и относительно стабильным, например, в документации, каталогах продуктов или коде. Он терпит неудачу, когда запрос сложен, релевантная информация распределена по множеству отрывков или корпус изменяется быстрее, чем индекс эмбеддингов. Резюмирование подразумевает сжатие разговора или документа для соответствия контексту. Оно работает, когда входные данные имеют последовательную структуру, как в истории разговоров, и допустима определённая потеря информации; оно терпит неудачу, когда отброшенные детали оказываются важными. В 2025 году отрасль начала заменять оба термина на «инжиниринг контекста»: вопрос уже не в том, поиск или резюмирование, а в том, какое сочетание хранимых знаний, найденных отрывков и сжатой истории даст наилучший результат.
Это контекст для данного конкретного запроса на вывод. Методbuild_context менеджера памяти представляет собой процесс формирования контекста, поскольку он определяет, из четырех источников данных следует зачитывать информацию, сколько брать из каждого и в каком порядке собирать полученные данные. Происхождение данных с самого начала. В систему памяти невозможно добавить функцию отслеживания происхождения данных задним числом. Уже с первой записи каждая запись должна содержать информацию о её источнике, коэффициенте достоверности, идентификаторе сессии и времени создания. Без этой информации невозможно осуществлять поиск с учетом степени доверия к данным, фильтровать результаты по времени создания и проверять некорректные результаты поиска. Команды, которые пропускают отслеживание происхождения данных на первом этапе работы, в итоге тратят недели на его добавление в базу данных, уже содержащую тысячи записей без такой информации.
Отравление памяти: схемы атак и способы защиты
Это та часть, которую руководства по проектированию памяти обычно опускают, хотя она должна присутствовать во всех из них.
В статье arXiv 2025 года была представлена модель атаки MINJA, являющаяся одним из самых полных публичных анализов проблемы отравления памяти в агентах больших языковых моделей. Её авторы сообщают, что с помощью обычных взаимодействий с пользователем без повышенных привилегий злоумышленники могут достигать уровня успеха инъекций выше 95% с использованием трёхэтапной схемы. На первом этапе устанавливается связь между невинным, законным объектом и темой, которую преследует злоумышленник. На втором этапе предоставляются тонкие намёки, делающие вредоносную информацию кажущейся надёжной. На третьем этапе происходит постепенное сокращение объёма инъекции до минимального уровня, который сохраняется при извлечении памяти.
Отравление памяти отличается от обычной инъекции промптов тем, что отравлённая память переходит через границы сессий и накапливается. Одна успешная инъекция в первой сессии может влиять на все последующие сессии бесконечно долго.
В отдельной статье 2025 года о защите от неудоверенных операций с использованием байесовских методов была описана пятиуровневая схема, охватывающая всю поверхность возможных атак:
- Регулирование входных данных в момент записи. Перед каждой записью в память вычисляется совокупный показатель доверия на основе нескольких независимых факторов: происхождение данных, семантический анализ шаблонов команд и обнаружение аномалий на основе истории действий пользователя. Один только фактор недостаточен, а инъекции типа MINJA разрабатываются так, чтобы обходить проверки, основанные на одном факторе.
- Удаление командных элементов перед записью. Если данные, предназначенные для сохранения в память, имеют форму команды ("всегда", "никогда", "вы обязаны", "не делайте"), их необходимо очистить или проигнорировать, вместо того чтобы сохранять. Настоящие фактические данные никогда не должны содержать приказов для агента.
min_confidence на стороне Weaviate в третьем уровне. Записи с низким уровнем достоверности исключаются из контекста независимо от степени их сходства. В более полной реализации поле доверия обновляется на основе результатов проверки, так что записи, которые подтверждаются, получают высокий уровень доверия, а противоречащие им — теряют его.Для большинства команд разумным порядком действий является сначала применение меток происхождения при каждой записи, затем обнаружение инструкций в момент записи, и только после этого — поиск с учетом коэффициента доверия. Мониторинг поведения ставится в конце, поскольку для него требуется база данных поведенческих паттернов, а её создание занимает недели работы с реальными данными.
Еще один момент: исследования, проведенные с реальными системами памяти, содержащими уже существующие законные записи, показали, что реальные условия значительно снижают эффективность атак по сравнению с лабораторными условиями, поскольку вредоносные записи конкурируют с законными данными за места в памяти при поиске. Это не повод игнорировать защитные меры. Это повод поддерживать высокое качество и уровень доверия к законным записям, поскольку они образуют первую линию защиты.
Что создавать в первую очередь
Не все четыре слоя необходимы сразу. Для большинства приложений искусственного интеллекта два из них покрывают примерно 80% случаев нехватки памяти: слой 1 (буфер плюс краткое изложение) и слой 3 (Weaviate с гибридным поиском, фильтр уверенности и информация о происхождении данных). Вместе они решают две наиболее частые проблемы — агент, забывающий что-то на полпути выполнения задачи, и агент, не обладающий доменными знаниями. Создайте оба слоя, сравните степень завершения задач до и после их использования, а уже затем определите, чего не хватает.
Добавьте эпизодическую память, когда вы начнете задаваться вопросом, почему агент забыл то, что сделал в прошлой сессии, ведь именно этот вопрос является сигналом к её добавлению. Не добавляйте её раньше времени. Добавьте процедурную память, когда та же категория ошибок повторяется в разных задачах и методы формулировки запросов не помогают их устранить, поскольку для поведенческого шаблона требуется процедурная память.
PostgreSQL с pgvector — разумный стандарт для большинства команд. Одна инстанция базы данных с командой CREATE EXTENSION IF NOT EXISTS vector обеспечивает соблюдение принципов ACID, гибкий SQL, поиск по векторному сходству и гибкую систему оценки; по оценкам, хостинг для 10 тысяч единиц памяти стоит примерно 45 долларов в месяц. Отдельная векторная база данных становится целесообразной только тогда, когда объем семантической памяти растет настолько, что индекс HNSW в PostgreSQL начинает замедлять работу, чего большинство команд никогда не столкнется.
Сделайте каждую запись в память асинхронной. Ответ на запрос пользователя не должен зависеть от фиксации изменений в базе данных, поэтому передайте эту задачу пулу потоков и продолжайте работу.
И избегайте использования внутренних хранилищ векторов в производстве. Развертывание без постоянного хранилища приводит к потере всей накопленной информации при первой перезагрузке сервера — ошибку, которую мало кто совершает дважды.