Чатырохузроўневая памяць для AI-агентаў з паходжаннем
Падзел кароткатэрміновай і доўгатэрміновай памяці, чатыры ўзроўні (буфер, эпізадычная, семантычная, працэдурная), выбар сховішча, асінхронны запіс і абарона ад атручвання памяці.
Уявіце агента, якій не вдаёцася завершыць чацвёртнае крыцьцо дзяловага працэсу, які складаецца з многа крыццёў. Цэлыя не мадэль — прычына. Агент проста болей не памятае, што ён ужо зрабіў. Такія неудачы выклікаюцца звычным патэранам: вызовы інструментаў выкананы правільна, запіты ёсць чыстымі, але калі дыялог перасягае певную „невідзімую“ дужыну, агент пачынае ціклізаваць — ён павтарае вопыты, на якія ужо былы даны адказы, запрашае даныя, якія ён ужо мае, і супераспраўляецца сам з сабой між шостым і дванацатым крыцьцам, нібы першыя пяць крыццёў узагалі не адбыліся.
Замена мадэлі на іншую рэдка калі зменяе сітуацыю. Проблема заключаецца ў інфраструктурэ, якая абслужвае мадэль, а точней — у відсутнасці справжняй архітэктуры памяці. Вставка последніх N прыемаў у запіт дапамагае чатботу, але цэе не працюе, калі агенту трэба выканаць двадзяцца крыццёў працэсу і памятаць на сівэннае крыцьцо, што ён вірашыў на трэцяе.
Шаблон, які рашуеюць гэтыя проблемы, — чатырошаровая стакава памяці, у якой кожны шар располагаецца за адпаведным сховішчам. У наступнай часты расказваецца, калі служыць кожны шар, як команды ў GitHub, Anthropic і Berkeley опісваюць сваія версіі, а таксама даўаюцца прыклады коду на Python для кожнага элемента.
Вікна кантэксту — гэта не памяць
Вікна кантэксту — гэта рабочая памяць: RAM, а не жорсткі дыск. Кожны токен у яму мае свой расход. Запіс весь історыя размовы ў вікно паўнага 128K здаецца безпечным, пакуль вы не запускаеце агента, які актыўна выкарыстоўвае інструменты. Трэццаць крокаў рабочага працэсу, уключаючы вызовы інструментаў, аналіз даных і логічныя выводы, шыбка спрацьоўваюць 100K токенаў, і калі вікно запоўняецца, старэйшы контэнт выкасываецца з першых месцаў, дзе зазвычай знаходзяцца первісныя умовы і задачы корыстніка.
Іншая, болей тонкая проблема ўсё ж горшая. Даследжэнне, адкрытая дзеячымі з Станфорда і Каліфорнійскага універсітэта Берклі і опублікованая ў журнале ACL у 2024 году пад заглавам "Lost in the Middle: How Language Models Use Long Contexts", паказала, што карыстоце ЛММ залежыць ад формы U працэўнае вікно. Мадэлі добра функцыонуюць з інформацыяй, якая знаходзіцца на пачатку і ў канцы кантэксту, а явна гірша — з інформацыяй, захаванай у серэдзіне. Штось такое спостаралося ў розныях архітэктурах, разных размерах модэляў і дужынях кантэксту, пры чым вікна тэставаліся ад 4K да 32K токенав. Даследжэнне, адкрытая MIT у 2025 году, прадстаўіла матэматычнае адказванне, стверджваючы, што каузальная маскаванне увагі і кодаванне роцыянтных пазіцый якія зменшуюць важар увагі да токенав, якія знаходзяцца далека ад канцаў серыі.
Для агента наследкі зрозумелы. Нават калі все падае під крэтыні, мадэль не будзе роўнасупрацаваць з усім. Крытычныя абмежэння на роўні 30 К токенав кантэксту з 60 К токенамі маюць рэальную можлівасць стаць практычна невыразнымі.
Мадэль забывае, таму што сістэма падае ёй больш мэтаяў, ніж яе можна надзейвана обдумваць, і нічога не контролюе тое, што прабывае ў «вікні» пераказу.
Пачатковыя раздэленні: кашточнае і давгастроковае памяць
До чатырхшаровай таксанаміі існуе простэйшая раздзеленне, якая большасць інжынераў трывожыць у свайму разуме. Памяць у сістэмах AI дзеліцца на кашточную і давгастрокову, і розумеў, чаго кожна з яёх не можа зрабіць, спрыяе стварэнню болей тонкай архітэктуры.
Кораткакалегкі памяць прызначаная для адной сесіі і існуе толькі пры адной взаімадзейнасці. Яна зберагае тое, што агенту трэба ў цёперашні момант: стан тэкушчага задання, на якім ён знаходзіцца, рэзультаты выкарыстоўвання інструментаў і працэс выканання логічных раследавань. У практыцы гэта ўсё ў складзе з вікна кантэксту, якое ёсць дужа дынамічным, постоянна апдэйтаваным, коштае даскладна за кожны токен і стыраецца, калі заканчываецца сесія.
У дасведчэннях LangChain гэта чыста апісана: кораткакалегкія памяты выступаюць для зберагання історыі размовы, рэзультатаў выкарыстоўвання інструментаў і тымчасовага стану логічных раследавань. Але не апісваецца тое, што гэта таксама першы элемент, які знікае пад час масовага выкарыстоўвання. У рабочым працэсе, які складаецца з двадзяці крокаў, проміжныя рэзультаты, размышленні, даныя з пошуку і адпаведзі інструментаў накапліваюцца і конкуруюць за той самы обмежаны простор. Гэта супамінае спробу зберагчы складны план у галаве, калі хтось постаўляе новыя факты і перарывае.
Кораткатэчасовая памяць складаецца з двух практычных элементаў:
- Стан задачы — гэта рабочыя запісы пра прыбылы працы: тэкушчы крок, застаўленыя падзадачы, статус выканання інструментаў і коордынацыя між агентамі. Агент, які плануе паўтарную поезію, фіксуе пункт прыбылы, ліміты бюджэту, розглянутыя рэйсы і выбраныя гатэлі, каб не перадзялваць работу з другога кроку, калі доходзіць да восьмага.
- Проміжныя выходы — гэта рэзультаты працы інструментаў, створаныя апісанні, выходы SQL, фрагменты пошуку і нотаткі. Інжынерныя каманды Anthropic апісалі законамасць у сістэмах, дзе агенты, якія фіксуюць нотаткі параду ў окне кантэкста і выкарыстоўваюць іх пазней, прагледваюцься лепш у дзейнасцях з дужа дзялёгімі задачамі, чым тые, якія намагаюцца зберагчы усё ў кантэксце. Сама практыка запісування нотатак ёсць стратэгіяю кораткатэчасовай памяці.
Довгострокова памяць зберыгаецца поза межамі сеансів. Яна дозволяе агенту згадаты, што корыстувач віддае перэвагу стыслім відповідзям, што мінулого тыжня розгортання зазнало невдачы праз проблемі з міграціею, або што схема базі даных змінілася у січні. LangChain візначае іі як памяць, якая зберыгаецца між потокамі і сеансамі, і на практыці ці даныя потрэбуюць базі даных, а не кантэкстного вікна.
Існуе два основні віды:
- Персональні уподобання корыстувача — гэта сігналі індывідуалізаціі, які формуюцца з часом, включаючы стыль напісання, тон, тэхнічну глібіну, вібір інструментаў і звічкі фарматування. Асістэнт з напісання, які пам’ятае, што ві завжды хочэце фармат Markdown і вісокій рывень тэхнічноі глібіні, стае корыснім у спосіб, які не можа зрывнятыся жодна безстанова сістэма, незалежно ад якосты іі мадэлі.
Багатыя команды падчыняюцца той самай памылцы. Яны ствараюць короткатэрмінную памяць, заполняючы вікно кантэксту, і дзейнаўскую памяць, ствараючы адзін сховішча вектораў. Абодва падходы ўскладненыя, а ў найкрашчым случае — неўжывучыя. Першы не функціонуе на велікых масштабах з вышэзгаданыя прычыны. Другі перастае працаваць, таму што пошук сэмабільнасці — гэта неефектыўны спосаб адзысквання логаў сесій, прывярк чы правілаў паведання, а кожны тып дзейнаўскай памяці выкарыстоўвае іншы бэкенд. Чатырох-слойны дызайн — гэта болей працёзная версія раздзелення на короткатэрмінную і дзейнаўскую памяць, якая прысвойвае кожнаму тыпу памяці неабходны спосаб зберагання.
Чатыры тыпы памяці, адна таксаномія
Команды, які ствараюць прыметныя сістэмы, незалежна адзін ад другога дасяглі той самы класыфікатор. Назвы разныя, але структура — ні.
Праект MemGPT, прэдставлены Чарльзам Пакерам і яго колегамі ў UC Berkeley у 2023 годзе, папулярызав ідэю спрыявання вікна кантэксту LLM як RAM — лімітаванага і цэнныя. Аператывная сістэма расшырвае памяць за дапомогою пейджавання і зберагчыка на дыску, а фрэймворкі агентаў могу аналогічна сімуляваць дзейнавучую памяць, перакладаючы старэйшы кантэкст у большы зберагчык і выкарыстоўваючы яго, калі гэта неабходна. Гэтая ідэя пазней стала асновай для Letta.
Команда LangMem з LangChain, фрэймворк ReMem ад Google DeepMind і прыметная архітэктура Mem0 — усе яны опісваюць варыянты чатырох тыпаў:
- Рабочая памэць — гэта тое, што ўсё час знаходзіцца ў кантэксце: актуальныя крокі размовы, запрашаныя дакументы і рэзультаты работы інструментаў. Яна працюе быстра, мае паказаныя межы і высокія вітраты за кожны токен.
- Эпізодычная памэць храніць даны пра пакульшыя взаімадзеі, напрыклад, што агент зрабіў для пэўнага корыстувальніка ў панедзелак і які быў рэзультат. Яна арганізавана хроналагічна і прыкладзена да конкрэтнай сесіі, зазвычай зберагаецца ў швядкім хранільніку тыпу key-value або базе дадзенаў SQL і запрашаецца за прынцыпам няўздрэвенасці або семантычнай супадзімасці.
- Семантычная памэць зберагае загальныя факты, спецыялізаваныя знаньні і тое, што было вывучана ў ходзе працы. Цэя частка являе сабою шар RAG, які запрашаецца за дапамогою семантычнай супадзімасці кожны раз, калі задача выкарыстоўвае спецыялізаваныя знаньні або ранейшыя даны.
Этыя тыпы памяці адпавядаюць прынцыпам кагнітывныя наукі, якая поясняе, чаму команды прыходзяць да іх. Яны адпавядаюць эпізодычной, семантычнай і пратымальной памяці ў нейронауцы, а рабочая памяць як „вікна кантэксту“ — гэта не проста метафора. Гэта найбліжэйшы функцыйнальны аналаг.
Выбір сховыша: SQL, вектарны чы граф
Якая база дадзеных павінна выконваліваць функцыі памяці? Яшчэ перад напісаннем коду чыстая адпаведзь — гэта залежыць ад тыпу памяці.
Базы дадзэнайў вектарных даных (Weaviate, Pinecone, Qdrant, Chroma) падходзяць для семантычнага запамятовання, таму што яны дазволяюць швырага адшукацыю падобнасці сярод эмбеддзінгаў у вялікіх масштабах, а большасць з іх сён працуюць у гібрыдным рэжыме, які спалюе BM25 з вектарамі, і якраз перавышае чыстую адшукацыю на аднойчынных тэкстах. У іх няма можлівасці тэмпаральнага расчылення і запытаў на адносы, а два суперэктныя факты з майжэ ідэнтычнымі эмбеддзінгамі могу падвесты. Не калічыце ў іх эпізодычныя запісы і структураваныя лёгкія аўдыту.
Базы дадзэных SQL, такія як PostgreSQL і SQLite, падходзяць для эпізодычнай памяці. Запіты структураваны за ID карыстальніка, сесіяй, тыпам задач або часовым меткам; об’ём запісей невялік, а гарантыі ACID маюць значэнне, калі агент фіксуе, што ён зрабіў і калі. PostgreSQL з расшырэнням pgvector аддае можлівасць выканання рэляцыйных запітоў і пошуку за вектарамі ў адной сістэме. Для колькасці менш за дзесяткі мільйонаў запісаў гэта прагматычны выбар — з знакомымі операцыямі, дашчавай хостынгам і можлівасцю адладжвання за дапамою SQL.
Графічныя базы дадзэных, такія як Neo4j, паслужаюць у сістэмах памяці толькі тады, калі эпізодычныя або семантычныя даны маюць складную рэляцыйную структуру, якая вымагае запітоў з кальколькіма перэходамі, напрыклад, выявлення таго, хто і каму дапісаў тры месяцы назад, калі быў створаны дакумент. Простыя запісы задач агента не падходзяць для графічных баз. Багато команд прыбегаюць да графічных баз, але пазнаходзяць, што ў іх все-такі патрэбен SQL.
У большасці сцэнарыяў развёртвання продукта выкорыстоўваецца рэляцыйныя сховішчы (SQL або PostgreSQL) для зберагання эпізодаў і прафіляў корыстуначаў, вектарныя сховішчы для семантычных пошукоў, а таксама стан у межах працэсу для працы з кантэкстам. Команды, якія хочуць меньшага колькасці операцыйнага навантажэння, выбіраюць гібрыдную схему на базе PostgreSQL і pgvector. Выберыце ранг сховішча залежна ад таго, калькі функцыйяў павінна выканаць памяць, і не вектарызавайце все.
Слой 1: буферная памяць з стварэнням апісаў
Найбольш простая форма — гэта deque з максымальным колькасцю паведамленняў, заданым як фіксованая цыфра:
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% адзначны адлік у шчыльнасці злучэння pull-request-аў, 90% проты 83%.
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) ёсць найважлівейшым інжынерным выборам у класе. Кожны вызов для запісу вяртаеся мгновэна, таму адпаведзь LLM ужо транслюецца да корыстніка перад тым, як пачнёцца будзь-які запіс у базе дадзеных.
Тры рашэнні, якія формуюць архітэктуру
Калі ёсць чатыры шары, тры архітэктурныя рашэнні вплываюць на усё іншэ, і багато команд робяць прынеймна адно неправільна.
Запісі па гарачай сцэнарыі або на фоне. Команда LangMem вложыла значны час у рашэнне гэтага праблемы. Запісі па гарачай сцэнарыі выконваюцца пад час размовы, калі агент знаходзіць факт, які варта зберагчы, і негараздуе запіс прамо ў ходзе размовы. Запісі на фоне выконваюцца пасля завершэння сесіі, калі аналізуецца весь ход размовы, ўбачаючы, што варта захаваць. Гарачая сцэнарыя спрычынае затрывкі, таму што кожны semantic.store(), які выконваецца ў цыкле адпаведзей, являе сабою запіс у базе дадзеных, які паўтараецца для корыстніка. Запіс на фоне павышае якасць запам’ятовання, таму што можна аналізаваць тэмы і рашэнні за весь перыяд размовы, а таксама захавваць болей дэталізаваныя падсумкі. Для прыладоў, якія відпаведзаюць корыстніку, запіс на фоне практычна завжды ёсць правільным выборам. Пул задач у persist() дапамагае досягнуць большай часткі выгоды, утрымваючыся ад абсалютна асінхронной схемы обработкі.
Адаптаванне або стасумарызацыя. Што ўключае адпаведныя рашэнні для разных проблем, але команды часта іх памешчаюць. Адаптаванне — гэта пошук значымых фрагментаў у вялікай канторцы. Запіт ёсць конкрэтным, а канторца вялікая і дастаткова стабільная, напрыклад, у вачынку дакументацыі, каталагаў продуктав або коду. Ён не ўспэльваецца, калі запіт ёсць складным, значымая інфармацыя распадзелена між багатымі фрагментамі, або канторца зменяецца быстрэй, чым індэкс укладання. Стасумарызацыя — гэта стысненне размовы або дакумента, каб ён пасляўдаваў кантэксту. Ёй ўдаёцца, калі вхідныя даны ёсць последовнымі, як у історыі размовы, і дазволена певная втрата дэталей; ёна не ўспэльваецца, калі адмовленыя дэталі віявляюцца важнымі. У 2025 годзе промысел пачаў заменяць оба тэрміны на «інжынерыя кантэксту»: пытанне больш не ў тым, адаптаваць чы стасумарызаваць, а ў тым, якая сумешчанне зберагнутых ведамасцей, адналежных фрагментаў і стыснутай історыі дае найлепшыя рэзультаты.
Гэты элемент являецца кантэкстам для гэтаго конкрэтнага запытку вывярочкі. Методbuild_context калектара памяці ёсць часткай праектавання кантэксту, адколі ён вяршыць выбор, з каторых чатырох сховішча чытаць даныя, сколькі ўзяць з кожнага і якім порадкам асабліваць ўсе элементы разам. Праўіненасць з самага пачатку. Нельга дадаць праўіненасць пазней у сістэме памяці. Ўжо з першага запісу кожны элемент павінен мець свой істочнік, значэнне доверы, ID сесіі і час стварэння. Без гэтага немагчыма адбываць вывярочку з урахоўваннем доверы, фільтравацыю па свежасці і перагляд некорэктных вывярочак. Команды, якія прахоўваюць праўіненасць у першым цыкле працы, востатком пасвячаюць недзелі на ўведэнне яе ў базу дадзеных, якая ужо мае тыясячы элементоў без гэтага.
Отруенне памяці: шаблоны атак і захопнення
Гэты раздзіл часта прахоўваюць у кансэлькт-документах па дызайну памяці, але ён павінен быць у кожным з іх.
У статты arXiv 2025 року була прэдставлена мадэль атакі MINJA – одын із найрэтэльнішых публічніх аналізів проблемі отруення памяці в агентах LLM. Іі авторы повідомляюць, што за допомогою звічайніх взаемодый корыстувача без підвішчэніх прывілеів зловміснікі можуць досягты рывня успішносты ін’екціі понад 95% за допомогою трьохетапноі схемі. На першому етапі встановлюецца зв’язок між невіннім, легітымнім об’ектом і тэмою, яку прагне досягты зловміснік. На другому етапі надаюцца тонкі підказкі, які робляць шкідліву інформацыю здаецца надыйною. На трэцьому етапі ін’екція скорочуецца до мінімальніх розмірыв, шчоб яна залішалася прыдатною пасля вілучэння з памяці.
Отруення памяці відрызняецца ад звічайноі ін’екціі запіту тым, што пошкоджэна інформацыя перэтынае межы сеансів і накопічуецца. Одна успішна ін’екція у першому сеансі можа впліваты на всі подальшы сеансі без обмежэнь у часі.
У окласнай студзіі 2025 года працы па тэме захоўнення доверлівасці на аднойчынных прыемах была раскрыта пяцьшаровая структура, якая пакрывае всю поверхню атак:
- Модэравацыя дадзеных пад час збіркі. Першыя кожнай збірцы дадзеных у памяць вычысляецца сумарны бал доверлівасці на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчынных прыемах на аднойчы
min_confidence у Weaviate, які знаходзіцца ў трэцьям слое. Аднаковыя па рэйтынгу доверчаласці элементы не ўрахоўваныя ў кантэксце, незалежна ад ступеня іх сэмабільнасці. У болей распрацаванай версіі поле доверчаласці апдэйтаванае на аднойчынку з рэзультатамі пераканання, тады элементы, якія паспелі пераканаць, атрымоўваюць доверчаласць, а тые, якія суперсечуцца з іншымі данымі, яе втрачаюць.Для большасця команд разумным порадкам ўраджэння ёсць спачатку атрыбутаванне паводзяў кожнай змены, пасля чаго выкананне пераканання ў час змены, а потым адзысканне з узятым у расчытанне коефіцыентам довер'я. Назіранне за паведаннем стоіць апошнім, таму што для яго трэбуецца базовы рывень паведання, а стварэнне такога рывня заўсёды захопляе некалькі тыдзёнав працоўных дадзеных.
Ішчо адна пункт: даследжэнні, якія выканаліся на рэальных сістэмах памяці з ужо існуючымі легітымнымі елементамі, паказалі, што рэальныя умовы значна зменшаюць эфектыўнасць атак у пораўнанні з лабораторнымі установкамі, адтолькі што заражаныя елементы конкуруюць з легітымнымі дадзеннямі за месцы ў памяці для адзыскання. Гэта не прычына прыхіліцца да захоўнікаў. Гэта прычына падтрымліваць высокую якасць і довер'е да легітымных елементаў, таму што яны ўтвараюць першую лінію захоўніка.
Што стварыць першым
Не всі чатыры шара патрэбны негайна. Для большайшай часткі застосоўанняў ШЫ два з іх пакрываюць апошнія 80% проблем з памэцю: Шар 1 (буфер плюс стыктурацыя) і Шар 3 (Weaviate з гібрыдным пошукам, фільтр доверы і інформацыя пра паводзеў). Яны разам рашаюць два найчастэйшыя проблемы — агента, які забывае частку задання, і агента, які не мае доменных ведамасцей. Створыце іх оба, парабялейце выкананне задання да і пасля, і толькі тады ацэніце, чаго не хапяе.
Дадзіце эпізодычную памэць, калі з’являецца запытанне, чаму агент забывае тое, што досяг у пярэднім сесіі, адколі гэта запытанне ёсць сігналам. Не дадзіце яе раней. Дадзіце процедурную памэць, калі тая ж класа памылак павтараецца ў розных заданнях, а інжынерыя запытанняў не можа яе вылечыць, адколі патэрн поведзення вымагае процедурной памэці.
PostgreSQL з падтрымкам pgvector ёсць разумны стандарт для большынства команд. Адна экзантэрасць базы дадзеных з CREATE EXTENSION IF NOT EXISTS vector абяроць сабе санкцыі ACID, гнучкі SQL, пошук на адпоўнасць вектораў і гібрыдную сістэму ацэнкі; па адныях оценках, хостыранне для 10K элементаў коштае або-альбо $45 на месцы. Стандартная база дадзеных для вектораў становіць сэнс толькі тады, калі об’ём семантычных дадзеных вырастае настолькі, што індэкс HNSW у PostgreSQL становіць прычыну спамтлень, чаго большасць команд ніколі не перажывае.
Зробіце кожную запіс дадзення асінхронной. Адпаведзь на запыт корыстніка не трэба чакаць пасля фіксацыі дадзенняў у базе, таму перадайце цю працу пулу задач і працуйце далей.
А таксама адмовіцеся ад внутрашняях сховішчаў вектораў у прымэтных сценарыях. Развертанне без стойкага сховішча ведае пра пазбавленне ўсіх накопаных дадзенняў пасля першага перыяджвання сервера — такую памятку робяць толькі некалькі людзей два разы.