Чотиришарова архітектура пам’яті для штучних інтелектуальних агентів з інформацією про походження даних
Розділення короткострокової та довгострокової пам’яті, чотири шари (буфер, епізодичний, семантичний, процедурний), варіанти зберігання, асинхронне записування та заходи захисту від отруєння пам’яті.
Уявіть агента, який зазнає невдачі на чотирнадцятому кроці довгого робочого процесу. Це не вина моделі – агент просто більше не пам’ятає, що він вже зробив. Невдача відбувається за знайомою схемою: виклики інструментів виконуються коректно, а запити є чіткими, але як тільки розмова досягає певної „невидимої“ довжини, агент починає повторюватися. Він повторює запитання, на які вже були дані відповіді, отримує дані, які вже має, та суперечить сам собі між шостим та дванадцятим кроками, ніби перші п’ять кроків взагалі не відбувалися.
Заміна моделі на іншу рідко щось змінює. Проблемою є інфраструктура, яка оточує модель, а точніше – відсутність справжньої архітектури пам’яті. Вставка останніх N повідомлень у запит працює для чат-бота, але не функціонує, коли агенту потрібно виконати двадцятикроковий процес та пам’ятати на сімнадцятому кроці те, що він вирішив на третьому.
Шаблон, який вирішує цю проблему, — це чотиришарова стек-пам’ять, де кожен шар має відповідне сховище даних. У наступному описано функції кожного шару, як команди в GitHub, Anthropic та Berkeley описують свої версії цього підходу, а також приклади коду на Python для кожної частини.
Вікно контексту — це не пам’ять
Вікно контексту — це робоча пам’ять: RAM, а не жорсткий диск. Кожен токен у ньому має певну вартість. Заповнення всієї історії розмови у вікно об’ємом 128K здається безпечним, поки ви не почнете використовувати агента, який активно використовує інструменти. Тридцятикроковий процес роботи з викликами інструментів, спостереженнями та послідовностями міркувань швидко споживає 100K токенів, і коли вікно заповнюється, найстаріший контент видаляється з початку, де зазвичай знаходяться початкові вимоги користувача та формулювання завдання.
Ще гіршою є більш тонка проблема. Дослідження дослідників з Стенфорда та Каліфорнійського університету в Берклі, опубліковане у 2024 році в журналі ACL під назвою „Lost in the Middle: How Language Models Use Long Contexts“, показало, що продуктивність ШІ-моделей має форму кривої у вигляді літери U протягом усього діапазону контексту. Моделі добре працюють із інформацією на початку та в кінці контексту, але значно гірше — із інформацією, розташованою посередині. Цей ефект спостерігався у різних архітектурах, розмірах моделей та довжинах контексту, причому тестувалися діапазони від 4K до 32K токенів. У наступному дослідженні 2025 року від MIT було надано математичне пояснення, згідно з яким маскування причинно-наслідкової уваги та кодування положення зменшують вагу уваги до токенів, розташованих далеко від кінців послідовності.
Для агентів наслідки є очевидними. Навіть якщо все підходить, модель не буде рівномірно звертати на це увагу. Критична обмеженість на рівні 30 тисяч у контексті з 60 тисячи токенів справді може призвести до того, що це залишиться практично непомітним.
Модель забуває, тому що система надає їй більше інформації, ніж вона може надійно обробити, і немає механізму контролю за тим, що потрапляє до її „вікна“.
Початковий поділ: короткострокова та довгострокова пам’ять
До появи чотирьохрівневої таксономії існує простіший поділ, який більшість інженерів мають у своїй голові. Пам’ять у системах ШІ поділяється на короткострокову та довгострокову, і розуміння того, чого кожна з них не може робити, спонукає до створення більш деталізованої архітектури.
Короткострокова пам’ять обмежена однією сесією та існує лише протягом однієї взаємодії. Вона зберігає те, що агент потребує прямо зараз: стан поточного завдання, крок, на якому він знаходиться, результати останнього використання інструменту та процес формування міркувань. На практиці це вікно контексту, яке є дуже динамічним, постійно оновлюється, коштує багато за кожен токен та очищується при закінченні сесії.
У документації LangChain це добре описано — короткострокову пам’ять представлено як місце зберігання історії розмови, результатів використання інструментів та тимчасового стану міркувань. Однак не зазначається, що саме цей елемент першим постраждає при масштабуванні. У робочому процесі з двадцятьма кроками агента проміжні результати, роздуми, дані пошуку та відповіді інструментів накопичуються та конкурують за обмежений простір. Це схоже на спробу зберегти складний план у голові, поки хтось постійно перериває новими фактами.
Короткострокова пам’ять має два практичні компоненти:
- Стан завдання — це робочий запис прогресу: поточний крок, незавершені підзавдання, статус виконання інструментів та координація між агентами. Агент, який планує подорож, зберігає напрямок призначення, бюджетні обмеження, переглянуті рейси та список готелів, щоб не повторювати роботу з другого кроку при досягненні восьмого.
- Проміжні результати — це продукти міркувань під час виконання завдання: результати інструментів, створені резюме, результати SQL-запитів, уривки з пошуку та нотатки. Інженерна команда Anthropic описала закономірність у продакшн-системах, згідно з якою агенти, які зберігають нотатки поза вікном контексту та отримують їх пізніше, досягають кращих результатів у довготривалих завданнях, ніж агенти, які намагаються тримати все в контексті. Сам процес записування нотаток є стратегією короткострокової пам’яті.
Довгострокова пам’ять існує довше, ніж сеанси. Вона дозволяє агенту згадувати, що користувачу подобаються стислі відповіді, що минулого тижня розгортання зазнало невдачі через проблему з міграцією, або що схема бази даних змінилася у січні. LangChain визначає її як пам’ять, яка зберігається між потоками та сеансами, а на практиці ці дані потребують бази даних, а не вікна контексту.
Існує два основні види:
- Переваги користувача — це сигнали персоналізації, які накопичуються з часом, включаючи стиль написання, тон, технічну глибину, вибір інструментів та звички форматування. Асистент з написання, який пам’ятає, що ви завжди хочете формат Markdown та високий рівень технічних знань, стає корисним у спосіб, який неможливо досягти безстановим системам, якими б хорошими не були їхні моделі.
Багато команд роблять ті самі помилки. Вони створюють короткострокову пам’ять, заповнюючи вікно контексту, та довгострокову пам’ять, створюючи один сховище векторів. Обидва підходи є неправильними або, у кращому випадку, неповними. Перший не функціонує ефективно у масштабах з причин, зазначених вище. Другий припиняє працювати, оскільки пошук схожостей — це поганий спосіб отримання записів сесій, переваг чи правил поведінки, а кожен тип довгострокової пам’яті потребує різного бекенду. Чотиришарова архітектура є більш досконалою версією розділення на короткострокову та довгострокову пам’ять, яка прив’язує кожен тип пам’яті до необхідного сховища.
Чотири типи пам’яті, одна таксономія
Команди, які розробляють промислові системи ШІ, незалежно одна від одної дійшли до однакової таксономії. Назви можуть відрізнятися, але структура залишається незмінною.
Проект MemGPT, представлений Чарльзом Пакером та його колегами з UC Berkeley у 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 повідомив про помітний ефект: агент для програмування з пам’яттю продемонстрував зростання швидкості об’єднання pull-requests на 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, яка належить до найбільш ретельних публічних аналізів проблеми отруєння пам’яті в агентах LLM. Її автори повідомляють, що через звичайну взаємодію користувача без підвищених привілеїв зловмисники можуть досягати рівня успішності введення даних понад 95% за допомогою трьохетапної схеми. На першому етапі встановлюється зв’язок між невинним, легітимним контентом та темою, яку прагне досягти зловмисник. На другому етапі надаються тонкі підказки, які роблять шкідливу інформацію здавалося б надійною. На третьому етапі відбувається поступове скорочення об’єму інформації, щоб вона залишалася придатною після вилучення з пам’яті.
Те, що відрізняє отруєння пам’яті від звичайного введення даних у запит, полягає у тому, що отруєні дані можуть перетинати межі сеансів та посилювати свій вплив. Одна успішна спроба введення даних у першому сеансі може впливати на всі наступні сеанси протягом необмеженого часу.
Окрема стаття 2025 року про захист від недовіри за допомогою бейзіанських методів описала п’ятишарову структуру, яка охоплює всі можливі точки атаки:
- Модерація вхідних даних під час запису. Перед кожним записом у пам’ять обчислюється комплексний показник довіри на основі кількох незалежних сигналів: походження даних, семантичний аналіз шаблонів команд та виявлення аномалій у історії користувача. Один сигнал сам по собі недостатній, а механізми типу MINJA розроблені так, щоб пройти перевірку на основі одного сигналу.
- Видалення команд перед записом. Якщо потенційний запис схожий на команду („завжди“, „ніколи“, „ви повинні“, „не робіть“), його потрібно очистити або видалити, замість того щоб зберігати. Справжні фактичні дані ніколи не потребують віддавати команди агенту.
min_confidence на рівні Weaviate у третьому рівні. Записи з низьким рівнем достовірності залишаються поза контекстом, незалежно від їхньої схожості. У більш повній реалізації поле довіри оновлюється на основі результатів перевірки, тож записи, які підтверджуються, набувають довіри, а суперечливі — її втрачають.Для більшості команд розумним порядком є спочатку позначення походження даних під час кожного запису, потім виявлення інструкцій у момент запису, а нарешті пошук з урахуванням рівня довіри. Моніторинг поведінки йде останнім, оскільки він потребує базового профілю поведінки, а створення такого профілю вимагає тижнів роботи з реальними даними.
Ще один момент: дослідження, проведені з реальними системами пам’яті з вже наявними легітимними записами, показали, що реальні умови значно знижують ефективність атак порівняно з лабораторними умовами, адже отруєні записи конкурують з легітимними даними за місця у пам’яті під час пошуку. Це не причина ігнорувати заходи захисту. Це причина підтримувати високу якість та надійність легітимних записів, адже саме вони є першою лінією захисту.
Що створювати спочатку
Не всі чотири шари потрібні відразу. Для більшості застосунків ШІ два з них вирішують приблизно 80% проблем з пам’яттю: Шар 1 (буфер та узагальнення) та Шар 3 (Weaviate з гібридним пошуком, фільтр впевненості та інформація про походження даних). Разом вони вирішують дві найпоширеніші проблеми – агент, який забуває частину завдання, та агент, який не має знань у певній галузі. Створіть обидва шари, порівняйте ступінь виконання завдань до та після їх впровадження, і лише тоді визначте, чого бракує.
Додайте епізодичну пам’ять, коли усвідомлюєте, що запитуєте себе, чому агент забув те, що досяг у попередній сесії, адже саме це запитання є сигналом. Не додавайте її раніше. Додайте процедурну пам’ять, коли та сама категорія помилок повторюється у різних завданнях, і оптимізація запитів не допомагає їх усунути, оскільки для поведінкових шаблонів потрібна процедурна пам’ять.
PostgreSQL з pgvector є розумним стандартом для більшості команд. Одна інстанція бази даних з командою CREATE EXTENSION IF NOT EXISTS vector забезпечує відповідність принципам ACID, гнучкий SQL, пошук за схожістю векторів та гібридну систему оцінювання; за оцінками, хостинг коштує близько 45 доларів на місяць для 10K елементів пам’яті. Самостійна база даних векторів стає корисною лише тоді, коли обсяг семантичної пам’яті досягає такого рівня, що індекс HNSW у PostgreSQL починає сповільнювати роботу, що зазвичай не трапляється у більшості команд.
Робіть кожен запис у пам’ять асинхронним. Відповідь користувачеві не повинна залежати від підтвердження операції в базі даних, тому передайте цю роботу пулу потоків та продовжуйте свої справи.
У продакшені краще уникати внутрішньопам’ятних сховищ векторів. Розгортання без постійного зберігання призводить до втрати всіх накопичених даних після перезавантаження сервера — помилку, яку мало хто робить двічі.