Проектаванне чатырьхшаровай памяці агента з викорыстанням LangGraph і Amazon Bedrock
Навучыцеся ствараць у агентах LLM працюючую, эпізодычную, семантичную та процедурную память на Bedrock і LangGraph, а таксама захіщаць яе ад отруўвання, вытэкі персональных данных і парадоксальнага вплыву з боку іншых корыстувачаў.
Крыніцы: Agentic Architectures — Article 7: Agent Memory Architectures
Раней чым пазней агент для вырабатчыка пачатвае виконваць тую ж самую роботу два разы. Нічага не зламана, запит таксама годны; агент проста не памятае, што ўжо рашыў майже ідэнтычную проблему ў празлеглы тыдзень, таму ён пераходзіць праз тыя жэны крокі, павтарае тыя ж узлеглыя помылкі, дасягае таго ж адпаведзення, і вам зноў трэба платіць за кожны токен. Шырокія інструкцыі паказваюць, як выправіць гэта за дапамогою справжней інфраструктуры памяці, а не большага вікна контексту: чатыры разныя рэвеню памяці, створаныя з LangGraph на Amazon Bedrock, а таксама меры захавання для вырабатчыка, якія не дазволяюць памяці стаць прычынай проблем.
Уявіце сабе агента для аналізу контрактав, які мае надзейную систему падтрымкі і работаючую канферэнцыю калькулятораў, але які ўсё равно спяшае першыя калькыляцыі з інструментамі на кожны падобны контракт, занова аднаходзячы структурныя шаблоны, якія ўжо былі выявлены раней. Ёму бракуе памяці.
Большае вікно контексту — гэта не памяць
Калі контэкст запоўняецца, звычныя рэакціі — гледзець на большыя вікна або выкарыстоўваць болей агрэсыўнае стасумаванне. І тое, і другое дапамагае падчас адпаведнай роботы; ніякое з гэтых не ўважаецца памяцю. Корыстны аналагія: вікно контэксту — гэта дошка для пісьма. Яна прыдатная падчас зустрэчы і стираецца, калі зустрэча заканчываецца. Памяць — гэта зошыт, які вы носіце з адной зустрэчы на іншую.
Кантроль контэксту мае трохі слабасей:
- Ён не спаявае розныя етапы роботы. Калі агент завершае свою роботу і його процес заканчываецца, контэкст зникае. Усё, што было вывучана, зникае, якщо толькі вы не будзеце явна його зберагаць.
- Ён не пераходзіць за межы агентаў. У багата-агентнай системе кожны працоўнік мае свой сабсконтэкст. Тое, што выявляе працоўнік, який адпавядае за перагляд безпекі, застаецца невидным для працоўніка, який аналізуе код, якщо толькі вы не будзеце спецыяльна перадаць тую інфармацыю туды.
Памэць задовольняе іншую патрэбу: стойкі, можна шуканы фонд знанняў, які перазіходзіць за межы адзінаковых запускаў і можа быть выкорыстаны ў будзь-якай топалогіі агента.
Чатыры рэвеню памяці агента
Кожны рэвеню паўтарае разную інфраструктуру, способы выкарыстоўвання ія і моды збою:
+---------------------+--------------------------------------------------+----------------------------+
| Memory Tier | What It Stores | Survives Run End? |
+---------------------+--------------------------------------------------+----------------------------+
| Working Memory | Active context during current run | No |
| | Tool outputs, reasoning steps, partial results | |
+---------------------+--------------------------------------------------+----------------------------+
| Episodic Memory | Summaries of past runs | Yes |
| | What happened, what worked, what failed | |
+---------------------+--------------------------------------------------+----------------------------+
| Semantic Memory | Structured knowledge about the domain | Yes |
| | Facts, relationships, learned patterns | |
+---------------------+--------------------------------------------------+----------------------------+
| Procedural Memory | Successful approaches to recurring task types | Yes |
| | How to do things, not just what was done | |
+---------------------+--------------------------------------------------+----------------------------+
Рэвенюі взаімаўпрацоўваюць у цыкле. Рабочая памэць зберагае даны пад час запуску. Калі запуск завершаецца, наступны крок прабывае даны у трох стойкіх рэвеню. Калі пачынаецца наступны запуск, крок выкарыстоўвання вытягвае неабходныя даны знову:
Run starts
|
v
+------------------+
| Working Memory | <-- active context, tool outputs, LangGraph state
| (current run) |
+------------------+
|
| Run ends: distillation step
v
+------------------+ +------------------+ +------------------+
| Episodic Memory | | Semantic Memory | | Procedural Memory|
| "what happened" | | "what I know" | | "how I do this" |
| DynamoDB | | Bedrock KB | | DynamoDB + KB |
+------------------+ +------------------+ +------------------+
| | |
+------------------------+------------------------+
|
Next run starts: retrieval step
|
v
Agent bootstrapped with
relevant prior knowledge
Эпізодычная памяць паведамляе наступнай роботы пра тое, што адбылася, семантычная памяць — пра тое, што вядома пра даную галерэю, а процедурная памяць — пра тое, як падыходзіць да задання. Усе тры викарбуваюцца пад запуск і размешчаюцца ў рабочай памяці прытаманнае агенту першае крокаў логічнага мыслення.
Рэгламент 1: кераванне рабочай памяцю ў межах адной роботы
Рабочая памяць — это все, што існуе пры адной роботы, і вы вже керуеце яю, незалежна ад таго, называеце вы яе такім чынам чы не. У LangGraph стан графа ёсць рабочая памяць: кожна змест, выход з інструменту і проміжны рэзультат знаходзяцца там.
Проблема заключаецца ў тым, што гэта явішча продовжвае растаць. Компрэсія контексту между стадіямі працы системы, яка ўважаецца распашчанай тэхнікай для калькуляцыі з мнага агентамі, на самай працэ є кераваннем памяцюючымі ресурсамі пад іншым названням. Клас, прыведзеный нижэй, ператварае гэю ідею на компонент, які можна викорыстоўваць знову. Функцыя maybe_compress выконваецца перад кожным крокам агента; калі лісту паведамленняў досягае певнага порогу (за замовчанням 30), ён захоўвае паведамленняя системы і 15 найсвежэйшых паведамленняў, якія не належаць да системы, без змян, а всё старэе заменяецца адным падсумком, створаным маленькім, дешавым модэлем. Другі метод, extract_run_artifacts, выконваецца напрыканцы роботы і просіць модэль стварыць структураваны JSON-запіс задачы, рэзультатаў, вынікаў, падходаў, інструментаў і шаблонаў, якія можна викорыстоўваць знову; гэты запіс стае вхідным дадзеннем для епізодычной памяці. Цей фрагмент напісаны на Python, а не на JavaScript:
# harness/memory/working_memory.py
from typing import TypedDict, Annotated, List, Optional
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage, SystemMessage, HumanMessage, AIMessage
from langchain_aws import ChatBedrock
import boto3
class WorkingMemoryManager:
"""
Manages the active context within a single agent run.
Applies a sliding window with summarization to prevent context overflow.
"""
def __init__(
self,
max_messages: int = 40,
summarize_after: int = 30,
region: str = "us-east-1",
):
self.max_messages = max_messages
self.summarize_after = summarize_after
bedrock = boto3.client("bedrock-runtime", region_name=region)
# Haiku for summarization: cheap and fast for mechanical compression
self.summarizer = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-haiku-4-5",
model_kwargs={"temperature": 0, "max_tokens": 2000},
)
def maybe_compress(self, messages: List[BaseMessage]) -> List[BaseMessage]:
"""
Called before each agent step. If the message list is getting long,
compress the oldest messages into a summary and keep the recent ones.
"""
if len(messages) < self.summarize_after:
return messages
# Keep the system message and the most recent messages
system_messages = [m for m in messages if isinstance(m, SystemMessage)]
non_system = [m for m in messages if not isinstance(m, SystemMessage)]
# Split: old messages get summarized, recent ones stay verbatim
cutoff = len(non_system) - 15 # always keep last 15 exchanges verbatim
to_summarize = non_system[:cutoff]
to_keep = non_system[cutoff:]
if not to_summarize:
return messages
summary_text = self._summarize(to_summarize)
summary_message = SystemMessage(
content=f"[WORKING MEMORY SUMMARY - earlier in this run]\n{summary_text}"
)
return system_messages + [summary_message] + to_keep
def _summarize(self, messages: List[BaseMessage]) -> str:
messages_text = "\n".join([
f"{type(m).__name__}: {m.content[:500]}"
for m in messages
])
response = self.summarizer.invoke([
SystemMessage(content="""
Summarize these agent messages into a compact working memory record.
Include: decisions made, tools called and their results, key findings,
and anything the agent will need to continue its work.
Discard: failed attempts, repetitive reasoning, tool errors that were resolved.
"""),
HumanMessage(content=messages_text)
])
return response.content
def extract_run_artifacts(self, messages: List[BaseMessage]) -> dict:
"""
At end of run, extracts key artifacts for episodic memory storage.
Separates what happened from how it happened.
"""
response = self.summarizer.invoke([
SystemMessage(content="""
Extract the following from these agent messages and return as JSON:
{
"task_summary": "one sentence description of what was asked",
"outcome": "COMPLETED | PARTIAL | FAILED",
"key_findings": ["list of important discoveries"],
"approach_used": "brief description of the approach that worked",
"tools_used": ["list of tool names"],
"time_taken_steps": 0,
"reusable_patterns": ["any patterns that could help on similar future tasks"]
}
Return only valid JSON.
"""),
HumanMessage(content="\n".join([
f"{type(m).__name__}: {m.content[:300]}"
for m in messages[-30:] # last 30 messages are enough
]))
])
import json
try:
return json.loads(response.content)
except json.JSONDecodeError:
return {"outcome": "UNKNOWN", "key_findings": [], "reusable_patterns": []}
Запрос на стварэнне падытку ігнаруе няудачныя спробы та дыяктуальныя памяткі інструментоў, а парсаванне JSON пераходзіць на результат UNKNOWN з порожнімі спісамі, таму некоректны адпаведзенне ніколі не выклікае збою ў працэсе. Ідентыфікаторы модэляў, такія як ID Haiku, паказваны тут, зменяюцца з часам і можа быць неабходнае наяўнасць профілю інферэнцыі, прызначанага для конкрэтнага регіону, таму пераканайцеся, што вы бачылі актуальны список модэляў Bedrock пры ўсуненні іх.
Для локальнага розвіцця вы можете заменіць стваральнік падытака Bedrock на модэль, якая працюе через Ollama:
# Local dev alternative
# ollama pull llama3.1:8b
from langchain_ollama import ChatOllama
summarizer = ChatOllama(model="llama3.1:8b", temperature=0, format="json")
Рэгіон 2: эпізодычная память пра пакананні
Эпізодычная память адпавядае на запитанне „Што выйшла раней?“. На пачатку паканання агент выкарыстоўвае падыткі пакананняў, якія супадаюць з текущым заведам, і ўжывае іх, каб ухіліцца ад зайвай роботы та вядомых памылак.
У даным дзеянні зберагаюцца структураваныя аптаты, а не первасовыя лог-файлы запускаў. Полныя лог-файлы даскладна зберагаць і медленна шукаць; компактныя аптаты ў дробнай канцэнацыі є дешавымі і швядкімі. Для зберагання викорыстоўваецца DynamoDB, а для выкарыстання інфармацыі — Bedrock embeddings для вярнага адліквідацыі сэмантычнай супараднасці.
Тэбл, заданы ў Terraform, викорыстоўвае agent_id як ключ партыцыі і run_id як ключ сортавання. Глобальны вторынны індэкс на agent_id і created_at дапамагае выплываць запыткі на кшталт „найсвежэйшыя эпізоды для гэтага агента“, а TTL у DynamoDB для expires_at автаматычна адмахвае старыя записы:
# infrastructure/memory_tables.tf
resource "aws_dynamodb_table" "episodic_memory" {
name = "agent-episodic-memory"
billing_mode = "PAY_PER_REQUEST"
hash_key = "agent_id"
range_key = "run_id"
attribute {
name = "agent_id"
type = "S"
}
attribute {
name = "run_id"
type = "S"
}
attribute {
name = "created_at"
type = "N"
}
global_secondary_index {
name = "agent-time-index"
hash_key = "agent_id"
range_key = "created_at"
projection_type = "ALL"
}
ttl {
attribute_name = "expires_at"
enabled = true
}
tags = {
Component = "agent-memory"
}
}
Сам магазін вбудоввае падзеўку задачы разам з ключовымі нарадкамі, запісвае елемент з терміном давальнасці 90 дзён, а пад час выявлення порыхвае вбудованае падзеўкі ныячайшай задачы з недавнімі эпізодамі за дапамою косай схожасці. Зберагаюцься толькі эпізоды з схожасцю вышэй за порог у 0,75; у разе равенства прыорітэт даёцца результатам COMPLETED; а функцыя format_for_injection выканае пераможцав як пазначаны блок для запросу системы:
# harness/memory/episodic.py
import boto3
import time
import json
import uuid
from typing import List, Optional
from dataclasses import dataclass
from langchain_aws import BedrockEmbeddings
import numpy as np
@dataclass
class EpisodeRecord:
run_id: str
agent_id: str
task_summary: str
outcome: str
key_findings: List[str]
approach_used: str
tools_used: List[str]
reusable_patterns: List[str]
embedding: Optional[List[float]]
created_at: float
expires_at: float
class EpisodicMemoryStore:
"""
Stores and retrieves summaries of past agent runs.
Uses embeddings for semantic retrieval so the agent can find
relevant episodes even when the wording is different.
"""
RETENTION_DAYS = 90
def __init__(
self,
agent_id: str,
table_name: str = "agent-episodic-memory",
region: str = "us-east-1",
):
self.agent_id = agent_id
self.region = region
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
bedrock = boto3.client("bedrock-runtime", region_name=region)
self.embeddings = BedrockEmbeddings(
client=bedrock,
model_id="amazon.titan-embed-text-v2:0",
)
def store_episode(self, artifacts: dict) -> str:
"""
Called at the end of each run to store what happened.
Returns the run_id of the stored episode.
"""
run_id = str(uuid.uuid4())
now = time.time()
# Embed the task summary for semantic retrieval later
text_to_embed = artifacts.get("task_summary", "") + " " + \
" ".join(artifacts.get("key_findings", []))
embedding = self.embeddings.embed_query(text_to_embed)
record = {
"run_id": run_id,
"agent_id": self.agent_id,
"task_summary": artifacts.get("task_summary", ""),
"outcome": artifacts.get("outcome", "UNKNOWN"),
"key_findings": artifacts.get("key_findings", []),
"approach_used": artifacts.get("approach_used", ""),
"tools_used": artifacts.get("tools_used", []),
"reusable_patterns": artifacts.get("reusable_patterns", []),
"embedding": embedding,
"created_at": int(now),
"expires_at": int(now + self.RETENTION_DAYS * 86400),
}
self.table.put_item(Item=record)
return run_id
def retrieve_relevant_episodes(
self,
current_task: str,
max_results: int = 5,
min_similarity: float = 0.75,
) -> List[EpisodeRecord]:
"""
Finds past episodes semantically similar to the current task.
Returns them ranked by relevance, filtered by outcome quality.
"""
query_embedding = self.embeddings.embed_query(current_task)
# Fetch recent episodes for this agent
# In production, you'd paginate this and use a vector DB for scale
response = self.table.query(
IndexName="agent-time-index",
KeyConditionExpression="agent_id = :aid",
ExpressionAttributeValues={":aid": self.agent_id},
ScanIndexForward=False, # newest first
Limit=100,
)
episodes = []
for item in response.get("Items", []):
stored_embedding = item.get("embedding")
if not stored_embedding:
continue
similarity = self._cosine_similarity(query_embedding, stored_embedding)
if similarity >= min_similarity:
episodes.append((similarity, item))
# Sort by similarity, take top results, prefer COMPLETED outcomes
episodes.sort(key=lambda x: (x[0], x[1].get("outcome") == "COMPLETED"), reverse=True)
return [
EpisodeRecord(**{k: v for k, v in item.items() if k != "embedding"}, embedding=None)
for _, item in episodes[:max_results]
]
def format_for_injection(self, episodes: List[EpisodeRecord]) -> str:
"""
Formats retrieved episodes into a system prompt block
for injection at the start of a new run.
"""
if not episodes:
return ""
lines = ["[EPISODIC MEMORY: relevant past runs]\n"]
for ep in episodes:
lines.append(f"Task: {ep.task_summary}")
lines.append(f"Outcome: {ep.outcome}")
if ep.key_findings:
lines.append("Key findings:")
for f in ep.key_findings:
lines.append(f" - {f}")
if ep.reusable_patterns:
lines.append("Patterns that worked:")
for p in ep.reusable_patterns:
lines.append(f" - {p}")
lines.append("")
return "\n".join(lines)
def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
a, b = np.array(a), np.array(b)
return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-9))
Запит запошчае не больш чым 100 няўжоўных эпізодаў і вырачае сэмантычную супадзімасць у коде прыкладнага праграмы; як зазначае каментар, у большых рэалізацыях трэба выкарыстоўваць пагінацыю і вектарныя базы дадзенаў. Таксама зважайце: ресурс boto3 DynamoDB адхоўляе значэнні типу float у Python, таму ліст з эмбеддінгамі павінен быть пераканвертаваны у Decimal (або серыязаваны, напрыклад у формате JSON або бінарным), перш чым функцыя put_item зможа яго прыняць. Модэль эмбеддінга, які вжываецца тут, описаны у документацыі Amazon Titan Embeddings.
Рэгія 3: сэмантычная память з базамі дадзенаў Bedrock Knowledge Bases
Эпізодычная памяць фіксуе ўражэння; семантычная памяць — знаёмства. Разлік тонкі, але важны. Уражэнне можа выглядаць так: «У контракце з постачальнікам, аналізаваным у пасляўтню вторник, была нестандартная клаза 7.3». Адпаведны семантычны факт: «Клаза 7.3 у контрактах з постачальнікамы зазвычай стосуецца максімума адпаведальнасці; нормальны дыяпазон — два разы вартасць контракта, а ўсё, што меней за один раз, являецца сігналам аб праблемах».
Семантычныя факты жывуць дыльней, чым эпізоды. Яны не выгораюць за 90 дней; яны удосконаляюцца з часам і падаюць у прынцыпе для розных типоў задач, а не толькі для тых, якія падобныя да практычнае выкарыстанні раней.
Bedrock Knowledge Bases — гэта ідеальны рышток для такага рангу, калі ўжо там адбываецца разбіўка на часткі, встроўванне, зберагчыце і выкарыстанне дакументаў. Адно, чаго большасць команд не стварыла, — это механізм зберагчэння новых дакументаў, адколі Knowledge Bases зазвычай выкарыстоўваюцца толькі для чытання. Наведзены нижэй рышток падтрымлівае оба направленні:
retrieveадмініструе семантычны пошук за дапамою вектараў, за неабяжнасцю фільтруючы яго па ключу метаданыхdomain, і адкладае рэзультаты з рэйтингам меншым за 0.6.store_factзбірае факт у вачынку JSON і кэруе яе ў S3, а таксама запускае задачу інтаграцыі, каб База знаёмых яе прыjęла.extract_facts_from_episodeпрасіць модель выявіць узнавальныя факты з завершанага запуску, пры чым ігнаруе деталі, спецыфічныя для конкрэтнай задачы і належнае якія да эпізодычнай памяці, і залишае толькі тыя факты, якіх надзея на правдопадобнасць становіць прынеймна 0.7.
# harness/memory/semantic.py
import boto3
import json
import uuid
import time
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class SemanticFact:
fact_id: str
content: str
domain: str # "contract_law", "security_patterns", etc.
confidence: float # 0.0 to 1.0
source_run_id: str
created_at: float
last_confirmed_at: float
class SemanticMemoryStore:
"""
Manages domain knowledge that persists across runs.
Reads from Bedrock Knowledge Base.
Writes new facts to S3 and triggers KB sync.
"""
def __init__(
self,
knowledge_base_id: str,
s3_bucket: str,
s3_prefix: str = "semantic-memory/",
region: str = "us-east-1",
):
self.kb_id = knowledge_base_id
self.s3_bucket = s3_bucket
self.s3_prefix = s3_prefix
self.region = region
self.bedrock_agent = boto3.client("bedrock-agent-runtime", region_name=region)
self.bedrock_agent_mgmt = boto3.client("bedrock-agent", region_name=region)
self.s3 = boto3.client("s3", region_name=region)
def retrieve(
self,
query: str,
domain: Optional[str] = None,
max_results: int = 5,
min_score: float = 0.6,
) -> List[str]:
"""
Retrieves relevant semantic facts for the current task.
Called at agent startup to bootstrap domain knowledge.
"""
retrieval_config = {
"vectorSearchConfiguration": {
"numberOfResults": max_results,
"overrideSearchType": "SEMANTIC",
}
}
if domain:
retrieval_config["vectorSearchConfiguration"]["filter"] = {
"equals": {"key": "domain", "value": domain}
}
response = self.bedrock_agent.retrieve(
knowledgeBaseId=self.kb_id,
retrievalQuery={"text": query},
retrievalConfiguration=retrieval_config,
)
return [
r["content"]["text"]
for r in response.get("retrievalResults", [])
if r.get("score", 0) >= min_score
]
def store_fact(self, fact: SemanticFact) -> str:
"""
Writes a new fact to S3 and triggers a KB ingestion job.
The fact will be available for retrieval after the sync completes
(typically 1-5 minutes).
For time-sensitive use cases, consider a separate fast-path
retrieval from DynamoDB alongside the KB retrieval.
"""
document = {
"fact_id": fact.fact_id,
"content": fact.content,
"domain": fact.domain,
"confidence": fact.confidence,
"source_run_id": fact.source_run_id,
"created_at": fact.created_at,
"metadata": {
"domain": fact.domain,
"confidence": str(fact.confidence),
}
}
key = f"{self.s3_prefix}{fact.domain}/{fact.fact_id}.json"
self.s3.put_object(
Bucket=self.s3_bucket,
Key=key,
Body=json.dumps(document),
ContentType="application/json",
)
# Trigger sync so new fact becomes retrievable
self.bedrock_agent_mgmt.start_ingestion_job(
knowledgeBaseId=self.kb_id,
dataSourceId=self._get_data_source_id(),
)
return fact.fact_id
def extract_facts_from_episode(
self,
episode_artifacts: dict,
extractor_model,
domain: str,
) -> List[SemanticFact]:
"""
At end of a successful run, extracts generalizable facts
from the episode and prepares them for semantic memory storage.
Only extracts from COMPLETED runs with confidence.
"""
if episode_artifacts.get("outcome") != "COMPLETED":
return []
from langchain_core.messages import SystemMessage, HumanMessage
response = extractor_model.invoke([
SystemMessage(content=f"""
Extract generalizable domain facts from this completed agent run.
Domain: {domain}
Rules:
- Only extract facts that would be useful across many different tasks
- Do not extract task-specific details (those belong in episodic memory)
- Rate confidence 0.0 to 1.0 based on how certain the agent was
- Skip anything that might be specific to this one instance
Return JSON array:
[
{{"content": "the fact", "confidence": 0.9}},
...
]
Return empty array if no generalizable facts found.
"""),
HumanMessage(content=json.dumps(episode_artifacts))
])
try:
facts_data = json.loads(response.content)
except json.JSONDecodeError:
return []
now = time.time()
return [
SemanticFact(
fact_id=str(uuid.uuid4()),
content=f["content"],
domain=domain,
confidence=f.get("confidence", 0.7),
source_run_id=episode_artifacts.get("run_id", "unknown"),
created_at=now,
last_confirmed_at=now,
)
for f in facts_data
if f.get("confidence", 0) >= 0.7 # only store high-confidence facts
]
def _get_data_source_id(self) -> str:
response = self.bedrock_agent_mgmt.list_data_sources(
knowledgeBaseId=self.kb_id,
maxResults=1,
)
return response["dataSourceSummaries"][0]["dataSourceId"]
Новыя факты не можна атрымоўваць негаразду. Процес абробкі трывае пакіяк, зазвычай калькаматна хвілінкі, як зазначаецца ў коментары да коду, таму задачі, чутлівыя да часу, патрабуюць швайнае атрымоўванне дадзеных (напрыклад, у DynamoDB) паўстаць разам з Базай веды. Ішчыя два моменты трэба пераканацца, супрацаваючы з документацыяй па абробцы дадзеных у Базе веды, прытымкі да цього коду. Першае – фільтрацыя метаданых у Базах веды выконваецца на атрыбутах метаданых, якія знаходзяцца ў окремым файле метаданых праза кожны дакумент S3, а не на полях, ўбудованых у тэла дакументаў, таму фільтр domain можа не падыходзіць нічаму так, як напісана. Другое – запуск задачы абробкі для кожнага факту можа стварыць суперсечнасць з задачай, якая вядома працюе над тымі ж дадзеннямі; зазвычай краща агрупаваць запісы і синхронізаваць іх за планам.
Ураган 4: процедурная памяць, з механізмам адзначэння правядзібнасці
Процедурная памяць ўсё часа являецца як наймоцнейшым, так і найбольш апакойлівым ураганам. Яна зберагае інструкцыі, як чаго-небудзь дасяжаць, а не толькі тое, што вже было зданае: успешны падход да павтараючыхся задач скарбліваецца у спісах крокаў, якія можна выкарыстоўваць знову, і вводзіцца ў будучыя запиты системы.
Апавярэнне заключаецца у тым, што некоректная процедура, якая была збережана, будзе выконвалася без запытанняў у кожнай наступной задачы. Самэй таго ў гэтым рангу патрэбна ворота атверджэння на шляху зберагання. У наведзенай нижэй системе propose_procedure працюе толькі для завершаных запускаў і зберагае новыя процедуры, калі параметр validated задаецца як false, тое ў значэнні «у паўпраце». Іншы агент або чалавек павінен затвердзіць процедуру, прычаму get_procedure зможа яе вернуць, а таксама get_procedure выклікаецца пры наявнасці ўжо хаця б трох прыемаслаў. Функція record_application_result фіксуе, як часта прыемалася процедура і як часта ёй вдалося, а format_for_injection паказвае этапы разам з канцэнтрацыяй успеху і нагадкай, што гэтыя этапы є толькі пачатковай базой, а не скрыптам:
# harness/memory/procedural.py
import boto3
import json
import time
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class Procedure:
procedure_id: str
task_type: str # "contract_review", "security_audit", etc.
title: str
steps: List[str]
success_rate: float # updated as procedure is applied to new runs
times_applied: int
times_succeeded: int
created_at: float
last_updated_at: float
validated: bool # human-validated procedures get priority
class ProceduralMemoryStore:
"""
Stores and retrieves learned procedures for recurring task types.
Write path has a validation gate: procedures are stored as PENDING
and must be validated (manually or by a second agent) before use.
This prevents a single bad run from poisoning future behavior.
"""
def __init__(
self,
table_name: str = "agent-procedural-memory",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
def get_procedure(self, task_type: str) -> Optional[Procedure]:
"""
Returns the best validated procedure for a task type, if one exists.
Prefers human-validated procedures, then sorts by success rate.
"""
response = self.table.query(
KeyConditionExpression="task_type = :tt",
FilterExpression="validated = :v AND times_applied >= :min",
ExpressionAttributeValues={
":tt": task_type,
":v": True,
":min": 3, # must have been used at least 3 times
}
)
items = response.get("Items", [])
if not items:
return None
# Sort by success rate, prefer human-validated
best = sorted(
items,
key=lambda x: (x.get("validated", False), x.get("success_rate", 0)),
reverse=True
)[0]
return Procedure(**best)
def propose_procedure(
self,
task_type: str,
episode_artifacts: dict,
extractor_model,
) -> Optional[str]:
"""
After a successful run, proposes a new procedure based on
what worked. Stored as PENDING until validated.
Returns procedure_id if a procedure was proposed, None otherwise.
"""
if episode_artifacts.get("outcome") != "COMPLETED":
return None
from langchain_core.messages import SystemMessage, HumanMessage
response = extractor_model.invoke([
SystemMessage(content=f"""
Based on this successful agent run, extract a reusable procedure for
task type: {task_type}
The procedure should be:
- A numbered list of clear, actionable steps
- Generalizable to similar future tasks, not specific to this one
- Based only on what actually worked, not what was attempted and failed
Return JSON:
{{
"title": "short title for this procedure",
"steps": ["step 1", "step 2", ...]
}}
If the run does not contain a clearly reusable procedure, return null.
"""),
HumanMessage(content=json.dumps(episode_artifacts))
])
try:
procedure_data = json.loads(response.content)
except (json.JSONDecodeError, TypeError):
return None
if not procedure_data or not procedure_data.get("steps"):
return None
import uuid
now = time.time()
procedure_id = str(uuid.uuid4())
self.table.put_item(Item={
"procedure_id": procedure_id,
"task_type": task_type,
"title": procedure_data["title"],
"steps": procedure_data["steps"],
"success_rate": 1.0,
"times_applied": 0,
"times_succeeded": 0,
"created_at": now,
"last_updated_at": now,
"validated": False, # starts as PENDING
"source_run_id": episode_artifacts.get("run_id", "unknown"),
})
return procedure_id
def record_application_result(
self,
procedure_id: str,
succeeded: bool,
):
"""
Updates the success rate after a procedure is applied.
If success rate drops below threshold, flag for review.
"""
self.table.update_item(
Key={"procedure_id": procedure_id},
UpdateExpression="""
SET times_applied = times_applied + :one,
times_succeeded = times_succeeded + :inc,
last_updated_at = :now,
success_rate = (times_succeeded + :inc) / (times_applied + :one)
""",
ExpressionAttributeValues={
":one": 1,
":inc": 1 if succeeded else 0,
":now": time.time(),
}
)
def format_for_injection(self, procedure: Procedure) -> str:
"""Formats a procedure for injection into agent system prompt."""
lines = [
f"[PROCEDURAL MEMORY: recommended approach for {procedure.task_type}]",
f"{procedure.title}",
f"(Success rate: {procedure.success_rate:.0%} over {procedure.times_applied} uses)",
"",
]
for i, step in enumerate(procedure.steps, 1):
lines.append(f"{i}. {step}")
lines.append(
"\nThis procedure is a starting point. Adapt it if the current task requires it."
)
return "\n".join(lines)
Спрытакцею гэтага фрагмента трэба спрыяць як эскізу дыяграмы, а не як готоваму коду, таму што калькі дэталей патрэбна скорекаваць. Выразы апдэйтаў у DynamoDB падтрымляюць толькі адналежанне і вырачэнне, таму дзеўерсію success_rate трэба вырачваць у коде прыкладнення (або пад час чытання) у заместа ўнутры UpdateExpression. Запыт выкарыстоўвае task_type як ключ, тады калі апдэйты — procedure_id, таму таблица патрэбна ключоўая схема або вторынны індэкс, які падтрымвае оба спосабы доступу. Часавыя меткі у формате float і success_rate патрабуюць той самы пераклад у формат Decimal, пра які гаворылася раней. Нарэшце, сумешчанне „паўнастаіленага“ з „застосаванага ўжо трохі разоў“ можа стварыць ситуацыю „куры і яе яйца“: трэба выявіць, як адносна нова паўнастаіленая процедура здобудзе сваі першыя застосаванні, напрыклад через режым пробнай експлуатацыі або выключэнне, затверджанае чалавекам.
/p>
Цялавольна забывчыца: занікненне памяці
Зберагчыце кожную памяць назаўсёды стварае два проблемы. Зберагчыце сотняў тысяч падсумакаў выканання ў DynamoDB і S3 мае рэальныя косты. Што ўжо болей тонка, застарэлыя памяці актываўна вводзяць у глухы кут: шаблон, які працаваў шэсць месцаў тыму, можа стаць некоректным зараз, калі домэн змяніўся. Хораша спроектаваная забывчыца — гэта функцыя, а не дзефект.
Менеджер занікнення памяці прыменяе разныя правілы для розных катэгарыяў. Для эпізодаў, якія старэйшыя за мінімальны період зберагчэння (за замовачам семь дзён), ён вычысляе записы, якія мелі результат FAILED. Для процедураў ён проста пазначае іх, а не вычысляе: все, што не было апдэйтавана прыбліжна за 60 дзён, або все, што было выкарыстана ўжо трохі разоў з рэшыткай нижэй 0,6, атрымвае пазначку needs_review, каб яе мог адзін чалавек пераглядаць. Хоць гэты код пазначаны як SQL, на самай працэ ён напісаны на Python:
# harness/memory/decay.py
import boto3
import time
from decimal import Decimal
from typing import List
class MemoryDecayManager:
"""
Manages the lifecycle of stored memories.
Applies different decay strategies per memory tier.
"""
def __init__(self, region: str = "us-east-1"):
self.dynamodb = boto3.resource("dynamodb", region_name=region)
self.region = region
def apply_episodic_decay(
self,
agent_id: str,
table_name: str = "agent-episodic-memory",
max_episodes: int = 500,
min_retention_days: int = 7,
):
"""
Prunes old episodic records when the store gets large.
Keeps: recent episodes, COMPLETED outcomes, high-similarity clusters.
Removes: old FAILED runs, duplicates of similar episodes.
"""
table = self.dynamodb.Table(table_name)
now = time.time()
min_age = now - (min_retention_days * 86400)
response = table.query(
IndexName="agent-time-index",
KeyConditionExpression="agent_id = :aid AND created_at < :cutoff",
ExpressionAttributeValues={
":aid": agent_id,
":cutoff": Decimal(str(min_age)),
},
ScanIndexForward=True, # oldest first
Limit=max_episodes,
)
candidates_for_removal = [
item for item in response.get("Items", [])
if item.get("outcome") == "FAILED"
]
with table.batch_writer() as batch:
for item in candidates_for_removal:
batch.delete_item(Key={
"agent_id": item["agent_id"],
"run_id": item["run_id"],
})
def flag_stale_procedures(
self,
table_name: str = "agent-procedural-memory",
stale_after_days: int = 60,
min_success_rate: float = 0.6,
) -> List[str]:
"""
Identifies procedures that should be reviewed:
- Not used in a long time (might be outdated)
- Success rate has dropped below acceptable threshold
Does NOT delete automatically. Flags for human review.
Returns list of procedure_ids that need attention.
"""
table = self.dynamodb.Table(table_name)
stale_cutoff = time.time() - (stale_after_days * 86400)
response = table.scan(
FilterExpression=(
"last_updated_at < :cutoff OR "
"(times_applied >= :min AND success_rate < :threshold)"
),
ExpressionAttributeValues={
":cutoff": Decimal(str(stale_cutoff)),
":min": 3,
":threshold": Decimal(str(min_success_rate)),
}
)
flagged_ids = []
for item in response.get("Items", []):
table.update_item(
Key={"procedure_id": item["procedure_id"]},
UpdateExpression="SET needs_review = :v",
ExpressionAttributeValues={":v": True},
)
flagged_ids.append(item["procedure_id"])
return flagged_ids
Выкрыцеўка адны ўпарацаванні стоіць некалькіх дадзенняў контэксту, тады як бяспечнае выдаленне працэдуры можа пазбавіць нас ап вымагчыранай, перакананай людзьмі методыкі, таму і такая асиметрыя. flag_stale_procedures выкарыстоўвае сканаванне табелі, што падходзіць для запланаваных батч-задаń (такіх як Lambda), але не для актыўных шляхоў выкарыстоўвання.
Аб’еднаванне на пачатку і канцы выконання
Усе рэвенюі збіраюцца на пачатку кожнага выконання. Клас bootstrap запытае хранільню упарацаванняяў пра да тры падобныя упарацаваннія, базу знаёмых пра да пяці фактов домэны, а хранільню працэдураў – пра перакананую працэдуру, якая падходзіць да типу задання. Пасля чаго ён абгортае все, што знайшоў, у чыста адзначаныя маркеры, каб модэль могла розлічваць запрашаную памяць ад самага задання:
# harness/memory/bootstrap.py
from typing import Optional
from langchain_core.messages import SystemMessage
from harness.memory.episodic import EpisodicMemoryStore
from harness.memory.semantic import SemanticMemoryStore
from harness.memory.procedural import ProceduralMemoryStore
class MemoryBootstrap:
"""
Assembles the memory context for a new agent run.
Called once at startup, before the agent sees the task.
"""
def __init__(
self,
agent_id: str,
knowledge_base_id: str,
s3_bucket: str,
region: str = "us-east-1",
):
self.episodic = EpisodicMemoryStore(agent_id, region=region)
self.semantic = SemanticMemoryStore(knowledge_base_id, s3_bucket, region=region)
self.procedural = ProceduralMemoryStore(region=region)
def build_memory_context(
self,
task: str,
task_type: Optional[str] = None,
domain: Optional[str] = None,
) -> str:
"""
Retrieves relevant memories from all tiers and assembles
them into a single context block for system prompt injection.
"""
sections = []
# Episodic: what happened on similar tasks before
episodes = self.episodic.retrieve_relevant_episodes(task, max_results=3)
if episodes:
sections.append(self.episodic.format_for_injection(episodes))
# Semantic: what is known about this domain
if domain:
facts = self.semantic.retrieve(task, domain=domain, max_results=5)
if facts:
sections.append(
"[SEMANTIC MEMORY: domain knowledge]\n" +
"\n".join(f"- {f}" for f in facts)
)
# Procedural: how to approach this task type
if task_type:
procedure = self.procedural.get_procedure(task_type)
if procedure:
sections.append(self.procedural.format_for_injection(procedure))
if not sections:
return ""
return (
"\n\n[MEMORY CONTEXT - retrieved from prior runs and domain knowledge]\n\n" +
"\n\n".join(sections) +
"\n\n[END MEMORY CONTEXT]\n"
)
Шлях для запісу навмысна адмініструецца окрема ад шляху для чытання. Пасля завершэння роботы кансолідатор выкарыстоўвае інформацыю з рабочай памяці, зберагае атрыбуты эпізода, выкарыстоўвае та зберагае семантычныя факты толькі пасля завершэння роботы, а таксама пропануе процедуру, якая будзе чакаць падтверджэння. У яго выкарыстоўваецца большы модэль, чым пад час стиснення падчас роботы, адтолькі ўжо таму, што гэты крок вплывае на будучую працю; пераканайцеся, што ID модэля ўсё яшчэ доступны ў вашай регіяў. У кадре таксама выкарыстоўваецца Optional без його імпорту з typing, таму, якщо вы хочаце яго зноў выкарыстоўваць, дадзіце неабходны імпорт:
# harness/memory/consolidator.py
from harness.memory.working_memory import WorkingMemoryManager
from harness.memory.episodic import EpisodicMemoryStore
from harness.memory.semantic import SemanticMemoryStore
from harness.memory.procedural import ProceduralMemoryStore
from langchain_aws import ChatBedrock
import boto3
class MemoryConsolidator:
"""
Runs at the end of each agent run to distill and store memories.
The write path is separate from the read path by design.
"""
def __init__(
self,
agent_id: str,
knowledge_base_id: str,
s3_bucket: str,
region: str = "us-east-1",
):
bedrock = boto3.client("bedrock-runtime", region_name=region)
self.extractor = ChatBedrock(
client=bedrock,
model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
model_kwargs={"temperature": 0, "max_tokens": 4000},
)
self.working = WorkingMemoryManager(region=region)
self.episodic = EpisodicMemoryStore(agent_id, region=region)
self.semantic = SemanticMemoryStore(knowledge_base_id, s3_bucket, region=region)
self.procedural = ProceduralMemoryStore(region=region)
def consolidate(
self,
final_state: dict,
task_type: Optional[str] = None,
domain: Optional[str] = None,
):
"""
End-of-run consolidation. Extracts and stores memories from all tiers.
"""
messages = final_state.get("messages", [])
# Step 1: Extract run artifacts from working memory
artifacts = self.working.extract_run_artifacts(messages)
# Step 2: Store as episodic memory
run_id = self.episodic.store_episode(artifacts)
artifacts["run_id"] = run_id
# Step 3: Extract and store semantic facts (only from successful runs)
if domain and artifacts.get("outcome") == "COMPLETED":
facts = self.semantic.extract_facts_from_episode(
artifacts, self.extractor, domain
)
for fact in facts:
self.semantic.store_fact(fact)
# Step 4: Propose procedural memory update
if task_type:
self.procedural.propose_procedure(task_type, artifacts, self.extractor)
Забезпечэння надзеі працы
Захопленне ад отруўвання памяці
Адзін няправильны запуск можа заўважыць некоректную інформацыю ў семантычный памяці або запропанаваць неверны прыем, і ў масштабе гэтая памылка вырастае, таму што кожны будучы запуск, який выкарыстоўвае гэтую інформацыю або следуе гэму прыему, успадковвае яе. Захопнікі ўтвараюцца шарамі:
- Запісваць можна толькі рэзультаты запускаў з статусам
COMPLETED, чыя даны маюць высокі рэверанс. - Пры падняць інформацыю ў вышэйшы статус неабходны павтаральныя, незалежныя адзірванні: адзін запуск толькі прыпускае якуюсь можлівасць, тады калі жа згода трохо запускаў ёсць рэальным доказам.
- Перад тым, як прыем павінен стаць апраўданым, ён павінен працаваць пад наглядам людзя.
Просты спосаб прыменення другога шару — это лічыльнік падтверджэнняў, які керуецца хешам контэнту інформацыі. Кожна адзірванне атомарна паднімае лічыльнік за дапамою if_not_exists, і інформацыя паднімаецца у вышэйшы статус толькі тады, калі лічба досягне заданага прага:
# Simple confirmation counter before promoting a fact
def promote_fact_if_confirmed(
fact_content: str,
confirmation_threshold: int = 3,
table_name: str = "fact-confirmation-log",
region: str = "us-east-1",
):
"""
Tracks how many independent runs have observed the same fact.
Only writes to semantic memory after threshold is reached.
"""
table = boto3.resource("dynamodb", region_name=region).Table(table_name)
import hashlib
fact_hash = hashlib.sha256(fact_content.encode()).hexdigest()[:16]
response = table.update_item(
Key={"fact_hash": fact_hash},
UpdateExpression="SET confirmation_count = if_not_exists(confirmation_count, :z) + :inc, content = :c",
ExpressionAttributeValues={":z": 0, ":inc": 1, ":c": fact_content},
ReturnValues="UPDATED_NEW",
)
return int(response["Attributes"]["confirmation_count"]) >= confirmation_threshold
Паколькі текст хэшуецца, версіі аднаго факту з разным формулюванням лічыцца окалечна; якщо акцыі стануць занадта рэдкімі, спачатку трэба нормалізаваць або групаваць факты.
Забэранне персональных данных з памяці
Памяць агента будзе сарабатваць з персональнымі ідэнтыфікацыйнымі данымі, якщо толькі вы актываўна гэтага не запобяжыце. Агент для аналізу контрактов зберагае інфармацыю пра контракты, а ў контрактах згадваюцца людзі, заработкі і іншыя персональныя деталі. Найчыстэйшы спосаб — практычна адчыстыць текст прытаманне, перш чым ён будзе запісаны ў эпізодычны або сэмантычны рэўерсы. Amazon Comprehend праставляе API для выявлення персональных данных, які добра працуе з англійскім текстам. Наведзеныя нижэй функцыі заменяе кожную выявленую ентытэцю на ўпіс её типу, ітэруючы з канца строкі, тады ранейшыя пазыцыі застаюцца чыннымі:
import boto3
def scrub_pii_before_storage(text: str, region: str = "us-east-1") -> str:
comprehend = boto3.client("comprehend", region_name=region)
response = comprehend.detect_pii_entities(Text=text[:5000], LanguageCode="en")
scrubbed = text
# Replace detected PII with type labels, working backwards to preserve offsets
for entity in sorted(response["Entities"], key=lambda x: x["BeginOffset"], reverse=True):
scrubbed = (
scrubbed[:entity["BeginOffset"]] +
f"[{entity['Type']}]" +
scrubbed[entity["EndOffset"]:]
)
return scrubbed
Зважайце, што вызов пераглядае толькі першыя 5 000 симваў, таму все, што знаходзіцца пасля яных, захоўваецца без перакантролювання. Дліжэйшыя тэксты трэба разбіць на часткі, і кожную з яных неабходна перакантролюваць окрема. Тыпы і ліміты падтрымваных ентытаў пераказаны ў дасягіненнях Comprehend PII.
Ізоляцыя тэнантов
Якщо агенты обслужваюць кількох клёянтаў на спакушанай інфраструктуре, их памяці должны быть абсалютна ізольаваныя: запуск для аднаго клёянта ніколі не должен збіраць память іншага клёянта. Найпростейшы спосаб — прызначаць кожнаму ключу памяці іменаванне з ідэнтыфікатарам тэнанта і застосоўваць контроль граніц у шары зберагача, напрыклад, за дапамою выразаў умовы DynamoDB, а не толькі пасляльгаваць фільтрацыі на рэвэрсе прыкладнення:
# Always include tenant_id in all memory operations
def store_episode_isolated(tenant_id: str, agent_id: str, artifacts: dict):
namespaced_agent_id = f"{tenant_id}#{agent_id}"
# All subsequent operations use namespaced_agent_id, never bare agent_id
Паколькі agent_id ўжо є ключам партыяцыі эпізодаў, прыфікс дапамагае захаваць звычныя запиты ў межах аднаго тэнанта. Той самы прыем павінен быть застосованы да Базы знаёмых, табелі процедур і журналу паўтарэннях.
Што можна спакаўвацца, калі памяць будзе у працэсе
Памяць з часам робіць агентаў значна кращымі, але ёй таксама прыносіць моды збоёў, якія важкае дыбагаваць у порৱаненні з будзь-чым, што стварае безстановы система:
- Паслабленая памяць горша, чым ўсёўсюдзе яе няма. Зблукалы безстановы агент паспяшае провал адной роботы. Агент з паспамятанай некоректной інформацыяю паспяшае провал кожной роботы, якая яе выкарыстоўвае, можа, на месцы. Перад расшырэнням дапамогі памяці неабходна інвеставаць у механізмы верыфікацыі.
Архітэктура для апылу
Полны цикл жыцця апылу, ад пачатковага запуску да кансалідацыі ў трох стойкіх сховішчах, выглядае так:
New Run Starts
|
v
+-------------------------------+
| Memory Bootstrap |
| 1. Query Episodic Store |
| 2. Query Bedrock KB |
| 3. Query Procedural Store |
| 4. Assemble context block |
+-------------------------------+
|
v
+-------------------------------+
| Agent Run (LangGraph) |
| Working Memory Manager |
| Compression at threshold |
| Tool calls with auth/retry |
+-------------------------------+
|
v
+-------------------------------+
| Memory Consolidator |
| 1. Extract run artifacts |
| 2. Scrub PII |
| 3. Write episodic record |
| 4. Propose semantic facts |
| (if confirmed: store) |
| 5. Propose procedure |
| (pending validation) |
+-------------------------------+
|
v
+------------------+ +------------------+ +------------------+
| DynamoDB | | Bedrock KB | | DynamoDB |
| Episodic Memory | | Semantic Memory | | Procedural Mem. |
| TTL: 90 days | | S3 backed | | Validated flag |
+------------------+ +------------------+ +------------------+
А гэта ўсё інфраструктурнае скупшча, уклікаючы локальную альтэрнатыву для разработкі на базе Ollama, ChromaDB і Docker Compose, якая дазволяе апрабаваць структуру памяці без витатаў AWS:
+-----------------------------+---------------------+------------------------------+
| Component | Technology | Role |
+-----------------------------+---------------------+------------------------------+
| Working Memory | LangGraph State + | Active context compression, |
| | MemorySaver | sliding window management |
+-----------------------------+---------------------+------------------------------+
| Episodic Memory Store | DynamoDB + TTL | Past run summaries, semantic |
| | | similarity retrieval |
+-----------------------------+---------------------+------------------------------+
| Episodic Embeddings | Titan Embed v2 | Semantic similarity search |
| | (Bedrock) | over episode store |
+-----------------------------+---------------------+------------------------------+
| Semantic Memory | Bedrock Knowledge | Domain facts, structured |
| | Bases + S3 | knowledge, write via ingest |
+-----------------------------+---------------------+------------------------------+
| Procedural Memory | DynamoDB | Learned task procedures, |
| | | validated flag, success rate |
+-----------------------------+---------------------+------------------------------+
| PII Scrubbing | AWS Comprehend | Before any memory write, |
| | | English PII detection |
+-----------------------------+---------------------+------------------------------+
| Fact Confirmation | DynamoDB | Multi-run confirmation gate |
| | | before semantic write |
+-----------------------------+---------------------+------------------------------+
| Memory Decay | DynamoDB TTL + | Episode pruning, stale |
| | scheduled Lambda | procedure flagging |
+-----------------------------+---------------------+------------------------------+
| Context Compression | Claude Haiku 4.5 | Working memory summarization |
+-----------------------------+---------------------+------------------------------+
| Artifact Extraction | Claude 3.5 Sonnet | End-of-run distillation |
+-----------------------------+---------------------+------------------------------+
| Local Dev Alternative | Ollama + ChromaDB | Full memory stack testable |
| | Docker Compose | without AWS costs |
+-----------------------------+---------------------+------------------------------+
| Infrastructure as Code | Terraform | DynamoDB tables, S3 bucket, |
| | | KB data source, IAM roles |
+-----------------------------+---------------------+------------------------------+
Галоўныя выводы
- Спрыймайце вікна контэксту як дошку для заметак і ствараць околачы, стойкі ўказальнік для знаёмасцей, якія павінны застацца пасля апылу.
- Раздзеліце памяць на рабочы, эпізодычны, семантычны і процедурны рэвеню, таму што кожны з іх выклікае разныя патрабаванні ў чысткай, адзыскве і карэспансіі на нештатныя ситуацыі.
- Сконцентруйце результаты выкананняў у компактных апісанні і фактах; ніколі не зберагайце і не шукайце первісныя лог-файлы як частку памяці.
- Заставьце найжорсткія меры безпекі на тых рэвеню, якія найбольш вплываюць на прыемнасць: падтверджэння неабходна для фактаў, а людская перапрацоўка — для процедураў.
- Адфільтруйце персональныя даны і забезпечыце ізоляцію корыстнікаў у слоі зберагача пры першай ж самой практыцы запісу памяці.
- З самага пачатку выдзеліце бюджет на процесы забывання, монітарынг адзысквы, затрымкі падчас прыемлень і можлівую перапрацоўку памяці.