使用LangGraph与Amazon Bedrock设计四层代理内存结构
学习如何在 Bedrock 和 LangGraph 上为大型语言模型代理赋予工作记忆、情景记忆、语义记忆和程序记忆，并保护其免受数据污染、个人身份信息泄露以及租户间数据交叉影响。
来源: Agentic Architectures — Article 7: Agent Memory Architectures
迟早，某个生产级代理会重复做同样的工作。系统本身没有问题，提示也正常；只是该代理不记得上周已经解决过几乎相同的问题，因此会再次按照相同的步骤进行推理，重复同样的中间错误，得出相同的答案，而你则不得不为每一次计算再次付费。本指南介绍了如何通过真正的记忆基础设施而非更大的上下文窗口来解决这一问题：利用 Amazon Bedrock 上的 LangGraph 构建四个独立的记忆层级，并设置相应的生产级防护机制，防止记忆成为负担。
想象一下，一个拥有完善支撑系统的合同分析代理，以及功能正常的多代理架构，却仍在处理每份类似合同时，将最初的几次工具调用用于重新发现之前已经分析过的结构模式。它所缺乏的正是记忆功能。
更大的上下文窗口并非记忆
当上下文空间被填满时，人们通常会选择扩大窗口或采用更激进的摘要方式。这两种方法在单次处理过程中都有帮助，但都无法替代记忆功能。一个恰当的类比是：上下文窗口就像一块白板，开会时很方便使用，会议结束后就会被擦掉；而记忆则像是你从一次会议带到下一次会议的笔记本。
上下文管理在三个方面存在不足：
- 它无法连接不同的处理流程。当智能体完成任务并退出进程时，其上下文就会消失。除非你主动将其保存下来，否则所有学到的信息都会丢失。
- 它无法跨越智能体之间的界限。在多智能体系统中，每个智能体都有独立的上下文。安全审查智能体发现的信息，除非你刻意将其传递给代码分析智能体，否则后者是看不到的。
内存则解决了不同的需求：它是一种持久的、可查询的知识存储方式，能够超越单次运行而存在，并且可以在任何代理拓扑结构中共享。
代理内存的四个层级
每个层级对应不同的基础设施、检索方式以及故障模式：
+---------------------+--------------------------------------------------+----------------------------+
| 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
情景记忆用于告知下一次运行之前发生了什么，语义记忆记录关于该领域的已知信息，而程序性记忆则说明如何完成任务。在智能体开始进行首次推理之前，这三类信息都会被调取并存入工作记忆中。
第一层级：在单次运行过程中管理工作记忆
工作记忆指的是在单次运行期间存在的所有内容，无论你是否将其称为工作记忆，其实都在对其进行管理。在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结果，因此格式错误的响应不会导致整个流程崩溃。此处所示的Haiku ID等模型标识符会随时间变化，可能需要针对特定地区的推理配置，因此在复制这些标识符之前请先查看当前的Bedrock模型列表。
在本地开发时，可以用Ollama提供的模型替换Bedrock摘要生成器：
# Local dev alternative
# ollama pull llama3.1:8b
from langchain_ollama import ChatOllama
summarizer = ChatOllama(model="llama3.1:8b", temperature=0, format="json")
第二层：过往运行记录的情景记忆
情景记忆可以回答“之前发生了什么？”的问题。在每次运行开始时，智能体会检索与当前任务相似的以往运行记录的摘要，并利用这些摘要来跳过重复工作，避免重复犯错。
该设计存储的是提炼后的摘要，而非原始运行日志。完整日志的存储成本高昂且检索速度缓慢；而紧凑的摘要则成本低廉且查询迅速。存储介质为DynamoDB，检索时通过Bedrock嵌入模型来实现语义相似度匹配。
在Terraform中定义的表以agent_id作为分区键，以run_id作为排序键。基于agent_id和created_at构建的全局二级索引可支持“该智能体的最新记录”查询，同时DynamoDB针对expires_at设置的TTL功能可自动删除旧记录：
# 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资源不接受Python的float类型值，因此在进行put_item操作之前，需将嵌入向量列表转换为Decimal类型（或序列化为JSON、二进制等形式）才能被接受。此处使用的嵌入模型在Amazon Titan Embeddings文档中有详细说明。
第三级：基于Bedrock知识库的语义记忆
情景记忆记录的是具体事件，而语义记忆记录的是知识。二者差异细微但意义重大。某段情景描述可能是：“上周二分析的一份供应商合同中有非标准的第7.3条。”相应的语义事实则是：“供应商合同中的第7.3条通常涉及责任限额规定；正常范围是合同金额的两倍，低于这一数值则属于危险信号。”
语义事实的存续时间比情景记忆更长。它们不会在90天后失效，还会随着时间不断优化，并且能辅助多种类型的任务，而不仅限于与过去类似的任务。
Bedrock知识库非常适合存储这类信息，因为它具备分块处理、嵌入技术以及存储与检索功能。不过大多数团队尚未构建写入机制，因为知识库通常被视为只读系统。下方的方案涵盖了双向操作：
retrieve会执行语义向量搜索，可选择通过domain元数据键进行过滤，并剔除得分低于0.6的结果。store_fact会将相关事实以JSON文档的形式写入S3，同时启动数据摄取任务，以便知识库能够获取这些数据。extract_facts_from_episode会让模型从已完成的任务中提取可通用的事实，忽略那些属于阶段性记忆的特定任务细节，仅保留置信度至少为0.7的事实。
# harness/memory/semantic.py
import boto3
import json
import uuid
import time
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class SemanticFact:
fact_id: str
content: str
domain: str # "contract_law", "security_patterns", etc.
confidence: float # 0.0 to 1.0
source_run_id: str
created_at: float
last_confirmed_at: float
class SemanticMemoryStore:
"""
Manages domain knowledge that persists across runs.
Reads from Bedrock Knowledge Base.
Writes new facts to S3 and triggers KB sync.
"""
def __init__(
self,
knowledge_base_id: str,
s3_bucket: str,
s3_prefix: str = "semantic-memory/",
region: str = "us-east-1",
):
self.kb_id = knowledge_base_id
self.s3_bucket = s3_bucket
self.s3_prefix = s3_prefix
self.region = region
self.bedrock_agent = boto3.client("bedrock-agent-runtime", region_name=region)
self.bedrock_agent_mgmt = boto3.client("bedrock-agent", region_name=region)
self.s3 = boto3.client("s3", region_name=region)
def retrieve(
self,
query: str,
domain: Optional[str] = None,
max_results: int = 5,
min_score: float = 0.6,
) -> List[str]:
"""
Retrieves relevant semantic facts for the current task.
Called at agent startup to bootstrap domain knowledge.
"""
retrieval_config = {
"vectorSearchConfiguration": {
"numberOfResults": max_results,
"overrideSearchType": "SEMANTIC",
}
}
if domain:
retrieval_config["vectorSearchConfiguration"]["filter"] = {
"equals": {"key": "domain", "value": domain}
}
response = self.bedrock_agent.retrieve(
knowledgeBaseId=self.kb_id,
retrievalQuery={"text": query},
retrievalConfiguration=retrieval_config,
)
return [
r["content"]["text"]
for r in response.get("retrievalResults", [])
if r.get("score", 0) >= min_score
]
def store_fact(self, fact: SemanticFact) -> str:
"""
Writes a new fact to S3 and triggers a KB ingestion job.
The fact will be available for retrieval after the sync completes
(typically 1-5 minutes).
For time-sensitive use cases, consider a separate fast-path
retrieval from DynamoDB alongside the KB retrieval.
"""
document = {
"fact_id": fact.fact_id,
"content": fact.content,
"domain": fact.domain,
"confidence": fact.confidence,
"source_run_id": fact.source_run_id,
"created_at": fact.created_at,
"metadata": {
"domain": fact.domain,
"confidence": str(fact.confidence),
}
}
key = f"{self.s3_prefix}{fact.domain}/{fact.fact_id}.json"
self.s3.put_object(
Bucket=self.s3_bucket,
Key=key,
Body=json.dumps(document),
ContentType="application/json",
)
# Trigger sync so new fact becomes retrievable
self.bedrock_agent_mgmt.start_ingestion_job(
knowledgeBaseId=self.kb_id,
dataSourceId=self._get_data_source_id(),
)
return fact.fact_id
def extract_facts_from_episode(
self,
episode_artifacts: dict,
extractor_model,
domain: str,
) -> List[SemanticFact]:
"""
At end of a successful run, extracts generalizable facts
from the episode and prepares them for semantic memory storage.
Only extracts from COMPLETED runs with confidence.
"""
if episode_artifacts.get("outcome") != "COMPLETED":
return []
from langchain_core.messages import SystemMessage, HumanMessage
response = extractor_model.invoke([
SystemMessage(content=f"""
Extract generalizable domain facts from this completed agent run.
Domain: {domain}
Rules:
- Only extract facts that would be useful across many different tasks
- Do not extract task-specific details (those belong in episodic memory)
- Rate confidence 0.0 to 1.0 based on how certain the agent was
- Skip anything that might be specific to this one instance
Return JSON array:
[
{{"content": "the fact", "confidence": 0.9}},
...
]
Return empty array if no generalizable facts found.
"""),
HumanMessage(content=json.dumps(episode_artifacts))
])
try:
facts_data = json.loads(response.content)
except json.JSONDecodeError:
return []
now = time.time()
return [
SemanticFact(
fact_id=str(uuid.uuid4()),
content=f["content"],
domain=domain,
confidence=f.get("confidence", 0.7),
source_run_id=episode_artifacts.get("run_id", "unknown"),
created_at=now,
last_confirmed_at=now,
)
for f in facts_data
if f.get("confidence", 0) >= 0.7 # only store high-confidence facts
]
def _get_data_source_id(self) -> str:
response = self.bedrock_agent_mgmt.list_data_sources(
knowledgeBaseId=self.kb_id,
maxResults=1,
)
return response["dataSourceSummaries"][0]["dataSourceId"]
新事实无法立即检索到。根据代码注释，数据摄取需要一定时间，通常为几分钟，因此对时间敏感的工作负载需要在知识库之外使用快速查询方式（例如在DynamoDB中）。在依赖此代码之前，还有两点需要根据知识库摄取文档进行验证。首先，知识库中的元数据过滤是基于每个S3文档旁边的单独元数据文件中提供的元数据属性来实现的，而非基于文档正文中嵌入的字段，因此按当前写法domain过滤器可能无法匹配到任何内容。其次，为每个事实都启动一次摄取任务可能会与同一数据源上正在运行的任务发生冲突；通常采用批量写入并按计划同步的方式更为合适。
第四级：程序性记忆，带有验证机制
程序性记忆既是强度最高的层级，也最为危险。它存储的是如何完成某事，而不仅仅是已经完成的内容：针对某种重复性任务的可行方法会被提炼成可复用的步骤列表，并注入到未来的系统提示中。
危险在于，一旦有缺陷的流程被存储下来，后续的所有任务都会不加质疑地遵循它。正因如此，这一层级需要在写入路径上设置验证机制。在下面的存储结构中，propose_procedure仅会在任务完成时运行，并将新流程的validated字段设置为false，表示该流程仍处于待处理状态。在get_procedure返回该流程之前，必须由其他智能体或人工进行审批，而且get_procedure还要求该流程至少已被应用过三次。record_application_result用于记录该流程被应用的频率以及成功的频率，而format_for_injection则负责展示流程步骤、成功率，并提醒用户这些内容仅作为起点，而非完整脚本：
# harness/memory/procedural.py
import boto3
import json
import time
from typing import List, Optional
from dataclasses import dataclass
@dataclass
class Procedure:
procedure_id: str
task_type: str # "contract_review", "security_audit", etc.
title: str
steps: List[str]
success_rate: float # updated as procedure is applied to new runs
times_applied: int
times_succeeded: int
created_at: float
last_updated_at: float
validated: bool # human-validated procedures get priority
class ProceduralMemoryStore:
"""
Stores and retrieves learned procedures for recurring task types.
Write path has a validation gate: procedures are stored as PENDING
and must be validated (manually or by a second agent) before use.
This prevents a single bad run from poisoning future behavior.
"""
def __init__(
self,
table_name: str = "agent-procedural-memory",
region: str = "us-east-1",
):
dynamodb = boto3.resource("dynamodb", region_name=region)
self.table = dynamodb.Table(table_name)
def get_procedure(self, task_type: str) -> Optional[Procedure]:
"""
Returns the best validated procedure for a task type, if one exists.
Prefers human-validated procedures, then sorts by success rate.
"""
response = self.table.query(
KeyConditionExpression="task_type = :tt",
FilterExpression="validated = :v AND times_applied >= :min",
ExpressionAttributeValues={
":tt": task_type,
":v": True,
":min": 3, # must have been used at least 3 times
}
)
items = response.get("Items", [])
if not items:
return None
# Sort by success rate, prefer human-validated
best = sorted(
items,
key=lambda x: (x.get("validated", False), x.get("success_rate", 0)),
reverse=True
)[0]
return Procedure(**best)
def propose_procedure(
self,
task_type: str,
episode_artifacts: dict,
extractor_model,
) -> Optional[str]:
"""
After a successful run, proposes a new procedure based on
what worked. Stored as PENDING until validated.
Returns procedure_id if a procedure was proposed, None otherwise.
"""
if episode_artifacts.get("outcome") != "COMPLETED":
return None
from langchain_core.messages import SystemMessage, HumanMessage
response = extractor_model.invoke([
SystemMessage(content=f"""
Based on this successful agent run, extract a reusable procedure for
task type: {task_type}
The procedure should be:
- A numbered list of clear, actionable steps
- Generalizable to similar future tasks, not specific to this one
- Based only on what actually worked, not what was attempted and failed
Return JSON:
{{
"title": "short title for this procedure",
"steps": ["step 1", "step 2", ...]
}}
If the run does not contain a clearly reusable procedure, return null.
"""),
HumanMessage(content=json.dumps(episode_artifacts))
])
try:
procedure_data = json.loads(response.content)
except (json.JSONDecodeError, TypeError):
return None
if not procedure_data or not procedure_data.get("steps"):
return None
import uuid
now = time.time()
procedure_id = str(uuid.uuid4())
self.table.put_item(Item={
"procedure_id": procedure_id,
"task_type": task_type,
"title": procedure_data["title"],
"steps": procedure_data["steps"],
"success_rate": 1.0,
"times_applied": 0,
"times_succeeded": 0,
"created_at": now,
"last_updated_at": now,
"validated": False, # starts as PENDING
"source_run_id": episode_artifacts.get("run_id", "unknown"),
})
return procedure_id
def record_application_result(
self,
procedure_id: str,
succeeded: bool,
):
"""
Updates the success rate after a procedure is applied.
If success rate drops below threshold, flag for review.
"""
self.table.update_item(
Key={"procedure_id": procedure_id},
UpdateExpression="""
SET times_applied = times_applied + :one,
times_succeeded = times_succeeded + :inc,
last_updated_at = :now,
success_rate = (times_succeeded + :inc) / (times_applied + :one)
""",
ExpressionAttributeValues={
":one": 1,
":inc": 1 if succeeded else 0,
":now": time.time(),
}
)
def format_for_injection(self, procedure: Procedure) -> str:
"""Formats a procedure for injection into agent system prompt."""
lines = [
f"[PROCEDURAL MEMORY: recommended approach for {procedure.task_type}]",
f"{procedure.title}",
f"(Success rate: {procedure.success_rate:.0%} over {procedure.times_applied} uses)",
"",
]
for i, step in enumerate(procedure.steps, 1):
lines.append(f"{i}. {step}")
lines.append(
"\nThis procedure is a starting point. Adapt it if the current task requires it."
)
return "\n".join(lines)
请将此代码片段视为设计草图而非可直接使用的代码，因为还有些细节需要调整。DynamoDB的更新表达式仅支持加法和减法操作，因此success_rate的除法运算必须在应用程序代码中（或读取数据时）进行，而不能放在UpdateExpression内部。查询时以task_type作为键，而更新操作则使用procedure_id，因此该表需要一种能够支持这两种访问方式的键结构或辅助索引。浮点型时间戳和success_rate同样需要经过前面提到的Decimal类型转换。最后，“已验证”与“至少被应用三次”这两个条件结合在一起可能会引发先有鸡还是先有蛋的问题：例如，需要确定新验证通过的流程如何获得首次应用机会，是通过试用模式还是通过人工审批的例外情况。
刻意遗忘：记忆衰减
将所有记忆永久保存会带来两个问题。在DynamoDB和S3中存储数十万条运行摘要确实需要成本。更隐蔽的问题是，过时的记忆会带来误导：六个月前有效的方案可能因为领域发展而不再适用。合理的遗忘机制其实是优点而非缺陷。
衰减管理器针对不同层级采用不同的策略。对于超过最低保留期（默认为七天）的记录，它会删除结果为FAILED的记录。而对于流程类数据，则采取标记而非删除的方式：60天内未更新的内容，或使用次数至少三次且成功率低于0.6的内容，都会被标记为needs_review，以便人工核查。尽管被标注为SQL，但实际上这段代码是Python编写的：
# harness/memory/decay.py
import boto3
import time
from decimal import Decimal
from typing import List
class MemoryDecayManager:
"""
Manages the lifecycle of stored memories.
Applies different decay strategies per memory tier.
"""
def __init__(self, region: str = "us-east-1"):
self.dynamodb = boto3.resource("dynamodb", region_name=region)
self.region = region
def apply_episodic_decay(
self,
agent_id: str,
table_name: str = "agent-episodic-memory",
max_episodes: int = 500,
min_retention_days: int = 7,
):
"""
Prunes old episodic records when the store gets large.
Keeps: recent episodes, COMPLETED outcomes, high-similarity clusters.
Removes: old FAILED runs, duplicates of similar episodes.
"""
table = self.dynamodb.Table(table_name)
now = time.time()
min_age = now - (min_retention_days * 86400)
response = table.query(
IndexName="agent-time-index",
KeyConditionExpression="agent_id = :aid AND created_at < :cutoff",
ExpressionAttributeValues={
":aid": agent_id,
":cutoff": Decimal(str(min_age)),
},
ScanIndexForward=True, # oldest first
Limit=max_episodes,
)
candidates_for_removal = [
item for item in response.get("Items", [])
if item.get("outcome") == "FAILED"
]
with table.batch_writer() as batch:
for item in candidates_for_removal:
batch.delete_item(Key={
"agent_id": item["agent_id"],
"run_id": item["run_id"],
})
def flag_stale_procedures(
self,
table_name: str = "agent-procedural-memory",
stale_after_days: int = 60,
min_success_rate: float = 0.6,
) -> List[str]:
"""
Identifies procedures that should be reviewed:
- Not used in a long time (might be outdated)
- Success rate has dropped below acceptable threshold
Does NOT delete automatically. Flags for human review.
Returns list of procedure_ids that need attention.
"""
table = self.dynamodb.Table(table_name)
stale_cutoff = time.time() - (stale_after_days * 86400)
response = table.scan(
FilterExpression=(
"last_updated_at < :cutoff OR "
"(times_applied >= :min AND success_rate < :threshold)"
),
ExpressionAttributeValues={
":cutoff": Decimal(str(stale_cutoff)),
":min": 3,
":threshold": Decimal(str(min_success_rate)),
}
)
flagged_ids = []
for item in response.get("Items", []):
table.update_item(
Key={"procedure_id": item["procedure_id"]},
UpdateExpression="SET needs_review = :v",
ExpressionAttributeValues={":v": True},
)
flagged_ids.append(item["procedure_id"])
return flagged_ids
丢失一集只会损失少量上下文，而默默删除某个流程则可能丢弃经过人工验证的宝贵方法，因此两者存在差异。flag_stale_procedures采用表扫描方式，适合定时批处理任务（如Lambda函数），但不适用于高频访问路径。
在运行开始与结束时整合信息
所有层级会在每次运行开始时汇聚在一起。引导类会从剧集存储中查询最多三集相似内容，从知识库中获取最多五条领域相关事实，再从流程存储中找出与任务类型匹配的已验证流程。随后它会将找到的内容用明确的标记包裹起来，以便模型能够区分检索到的记忆与任务本身：
# 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 |
+-----------------------------+---------------------+------------------------------+
关键要点
- 将上下文窗口视为一块白板，为那些需要在运行过程中保留的知识构建独立的持久化笔记本。
- 将内存分为工作记忆、情景记忆、语义记忆和程序记忆四个层级，因为它们分别需要不同的存储方式、检索机制及故障处理策略。
- 将运行过程提炼为简洁的摘要和事实，绝不能以原始日志的形式作为内存存储或检索。
- 对那些最容易改变行为的层级设置更严格的管控措施：事实层面依赖确认机制，程序层面则需人工验证。
- 在首次写入任何内存数据之前，先在存储层去除个人身份信息并确保租户之间的隔离。
- 从一开始就为数据遗忘、检索监控、数据摄入延迟以及最终的重新嵌入做好预算准备。