Проектирование четырёхуровневой памяти агента с использованием 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 последних эпизодов и вычисляет степень сходства в коде приложения; как отмечено в комментарии, для более крупных систем необходимо использовать пагинацию и векторные базы данных. Также обратите внимание: ресурс DynamoDB в boto3 отклоняет значения типа 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, о которой говорилось ранее. Наконец, сочетание критериев «проверенности» и «использования как минимум три раза» может привести к ситуации «курица и яйцо»: необходимо определить, как процедура, только что проверенная, получает свои первые вызовы — например, через режим тестирования или исключение с одобрением человека.
/p>
Умышленное забывание: ухудшение памяти
Хранение каждой записи в памяти навсегда создаёт две проблемы. Хранение сотен тысяч сводок выполнения задач в 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 для обнаружения персональных данных, хорошо работающее с английским текстом. Приведённая ниже функция заменяет каждую обнаруженную сущность на её метку типа, проходя по строке с конца, чтобы ранее обработанные участки оставались действительными:
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 |
+-----------------------------+---------------------+------------------------------+
Основные выводы
- Рассматривайте окно контекста как доску и создавайте отдельную, постоянную записную книжку для знаний, которые должны сохраняться после завершения выполнения задачи.
- Разделяйте память на рабочий, эпизодический, семантический и процедурный уровни, поскольку каждый из них требует различных способов хранения, поиска и обработки сбоев.
- Преобразуйте результаты работы в компактные краткие изложения и факты; никогда не храните и не ищите в памяти сырые логи.
- Устанавливайте самые строгие механизмы контроля на те уровни, которые наиболее сильно влияют на поведение: для фактов необходимы подтверждения, а для процедур — проверка человеком.
- Удаляйте персональные данные и обеспечивайте изоляцию пользователей на уровне хранения ещё до того, как будет записана первая информация в память.
- С самого начала выделяйте ресурсы на учет возможности забывания, мониторинг поиска, задержки при вводе данных и их последующую перезапись.