Розробка чотирирівневої пам’яті агента за допомогою 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 для оцінки семантичної схожості.
Таблиця, визначена у 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
Епізодична пам’ять фіксує події; семантична пам’ять — знання. Різниця є тонкою, але важливою. Епізод може виглядати так: «У контракті з постачальником, проаналізованому минулої вівторка, була нестандартна клозула 7.3». Відповідний семантичний факт звучить так: «Клозула 7.3 у контрактах з постачальниками зазвичай стосується лімітів відповідальності; звичайний діапазон — це подвоєна вартість контракту, а будь-що менше — це тривожний сигнал».
Семантичні факти зберігаються довше, ніж епізоди. Вони не закінчують своєї дії через 90 днів; вони удосконалюються з часом та допомагають у багатьох типах завдань, а не лише в тих, що схожі на попередні виконання.
Bedrock Knowledge Bases є ідеальним середовищем для цього рівня, оскільки вона обробляє чанкування, ембеддинги, зберігання та пошук інформації. Однак більшість команд ще не створили механізм запису, адже бази знань зазвичай використовуються лише для читання. Наведений нижче механізм покриває обидві напрямки:
retrieveвиконує семантичний векторний пошук, за бажанням фільтруючи його за ключем метаданихdomain, та відкидає результати з оцінкою нижче 0.6.store_factзберігає факт у вигляді JSON-документа у S3 та запускає завдання імпорту, щоб База знань могла його отримати.extract_fact_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 її поверне, причому ця функція також вимагає щонайменше трьох попередніх запитів. Функція 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, про яку йшлося раніше. Нарешті, поєднання критеріїв „перевіреного“ та „застосованого принаймні три рази“ може створити ситуацію „курка чи яйце“: потрібно вирішити, як процедура, щойно перевірена, отримає свої перші застосування — наприклад, через режим тестування чи за допомогою винятку, схваленого людиною.
Навмисне забування: занепад пам’яті
Зберігання кожної спогади назавжди створює дві проблеми. Зберігання сотень тисяч підсумків виконань у DynamoDB та S3 має реальні витрати. Ще більш непомітно, застарілі спогади активно вводять у оману: схема, яка працювала шість місяців тому, може бути неправильною зараз через зміни в сфері діяльності. Добре спроєктоване забування є перевагою, а не недоліком.
Менеджер занепаду пам’яті застосовує різну політику для кожного рівня. Для епізодів, які старші за мінімальний термін зберігання (за замовчуванням сім днів), він видаляє записи, результатом яких був FAILED. Для процедур він позначає їх, а не видаляє: все, що не оновлювалося протягом 60 днів або що використовувалося принаймні три рази зі швидкістю успіху нижче 0,6, отримує позначку needs_review для перевірки людиною. Цей код написаний на Python, хоча позначений як SQL:
# 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"
)
Шлях запису навмисно відокремлений від шляху читання. Наприкінці виконання програми збиральник даних витягує елементи з робочої пам’яті, зберігає епізод, витягує та зберігає семантичні факти лише після завершення виконання та пропонує процедуру очікування підтвердження. Для цього використовується більша модель, ніж під час стиснення під час виконання, оскільки цей крок впливає на майбутню поведінку; переконайтеся, що ідентифікатор моделі все ще доступний у вашому регіоні. У цьому фрагменту також використовується 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 для виявлення персональних даних, яке добре працює з англійським текстом. Наведена нижче функція замінює кожну виявлену entитет на її типову мітку, ітеруючи з кінця рядка, щоб ранні позиції залишалися дійсними:
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 |
+-----------------------------+---------------------+------------------------------+
Основні висновки
- Розглядайте вікно контексту як дошку для записів та створюйте окремий, постійний блокнот для знань, які мають залишитися після завершення виконання.
- Розділіть пам’ять на робочий, епізодичний, семантичний та процедурний рівні, оскільки кожен з них потребує різного способу зберігання, пошуку та обробки помилок.
- Перетворюйте результати виконання на компактні узагальнення та факти; ніколи не зберігайте та не шукайте первинні журнали як частину пам’яті.
- Встановіть найсуворіші контролі на тих рівнях, які найбільше впливають на поведінку: підтвердження важливе для фактів, людська перевірка — для процедур.
- Видаліть персональні дані та забезпечте ізоляцію користувачів на рівні зберігання ще до того, як буде записана перша інформація в пам’ять.
- З самого початку передбачіть кошти на процеси забування, моніторинг пошуку, затримки у прийомі даних та їх подальше вставлення.