为AgentCore记忆设计生命周期策略：AWS CDK可部署方案

内存生命周期策略帮助在 Amazon Bedrock AgentCore 上长时间运行的 AI 智能体保持高效，通过系统性地管理其记忆内容来决定记住什么、遗忘什么。AI 智能体从每次对话中生成记忆。如果你不积极管理这些记忆，AI 智能体会积累过时的上下文，这可能导致响应质量下降，并为你的部署带来合规风险。

在生产环境中运行数月后，问题会逐渐显现。我们观察到，一个客服 AI 智能体将四个月前已解决的账单纠纷作为活跃案例引用。另一个 AI 智能体重复了过时的部署建议，因为它的记忆中仍然保留着已被取代的运维手册。

在本文中，我们介绍 AI 智能体的内存生命周期管理：系统性地对 AI 智能体记忆进行评分、整合和修剪的实践。我们将带你走过一套基于 AgentCore memory（Amazon Bedrock AgentCore 的一项能力）、AWS Step Functions 和 Amazon Bedrock 的可部署架构，运行夜间生命周期工作流。到最后，你将获得一个 AWS Cloud Development Kit（AWS CDK）堆栈以及一个将 AI 智能体记忆作为托管资源管理的框架。完整代码可在 GitHub 仓库中获取。

此解决方案面向在数周或数月内积累大量交互数据的高容量 AI 智能体，如客服 AI 智能体、销售顾问和 IT 服务台机器人。对于低容量 AI 智能体（如个人助手），你可能只需从存活时间（TTL）过期和《通用数据保护条例》（GDPR）合规开始。所有阈值都可配置以匹配你 AI 智能体的需求。

此解决方案将共享记忆分类法与三个作为夜间工作流运行的生命周期策略相结合。我们先从塑造这些策略的记忆类型开始。

记忆类型

在设计生命周期策略之前，我们需要一个关于 AI 智能体记忆内容的共享词汇表。我们将 AI 智能体记忆分为三种类型，每种都有不同的保留需求。

情景记忆（Episodic memory）：情景记忆捕捉发生的事情，是过去对话的记录。这些记录带有时间戳，绑定在会话中，且数量庞大。AgentCore memory 以两种策略存储这些信息：摘要（Summary）和情景（Episodic）。两种策略都将记忆存储为与特定 AI 智能体-用户会话绑定的单独条目。情节和摘要提供短期连续性，但随着时间推移，它们各自的相关性会逐渐降低。在设计生命周期策略时，优先让这些记忆首先过期。

语义记忆（Semantic memory）：语义记忆是从交互中提炼出来的事实和偏好，但与任何单一对话无关。「用户偏好使用美国东部（北弗吉尼亚）AWS 区域（us-east-1）进行部署。」这些是持久的、高价值的、紧凑的。在你的生命周期策略中，将语义记忆保留得比情景记忆更长时间。它们是整合的主要候选者，可以将多个情景观察合并为单一权威事实。

程序性记忆（Procedural memory）：程序性记忆编码了习得的工作流和工具使用模式。「当用户询问成本时，先查询 AWS Cost Explorer API，然后总结。」这些代表了 AI 智能体的操作专业知识。程序性记忆数量较少，但对某些用例而言是价值最高的类型。它们保留时间最长，修剪门槛最高。AgentCore memory 将程序性知识存储为与情景记忆绑定的反思（reflection）。在情景记忆深度解析博客中阅读更多相关内容。随着你的流程演进，你应该检查这些记忆的有效性。

三种生命周期策略

有了我们的分类法，我们就可以设计三种互补的生命周期策略。每个策略针对无限制记忆的不同失效模式。

策略一：TTL 过期

第一个策略自动删除超过配置 TTL 的记忆。我们默认将情景记忆的 TTL 设置为 90 天。TTL 不考虑记忆是否仍然有用，但它为积累量提供了硬性上限，对合规至关重要。

在生产环境中，按记忆类型区分 TTL。将摘要记忆配置为 30-60 天后过期，语义记忆 6-12 个月后过期，并考虑对程序性记忆不设置 TTL。本文提供了一个可配置的 memoryTtlDays 参数作为起点。TTL 过期首先运行，在评分或整合之前，这有助于避免在本应删除的记忆上浪费计算资源。

AgentCore memory 没有提供内置的自动删除 TTL。然而，它暴露了系统生成的时间戳字段，支持在 ListMemoryRecords 上使用 BEFORE 和 AFTER 过滤运算符。我们的修剪器使用 x-amz-agentcore-memory-createdAt 和 BEFORE 过滤器来仅检索早于配置 TTL 的记录，然后删除它们。

cutoff = (now - timedelta(days=ttl_days)).isoformat()

response = client.list_memory_records(
    memoryId=memory_id,
    namespace=agent_id,
    metadataFilters=[{
        "left": {"metadataKey": "x-amz-agentcore-memory-createdAt"},
        "operator": "BEFORE",
        "right": {"metadataValue": {"dateTimeValue": cutoff}},
    }],
)

策略二：相关性评分

并非所有记忆都以相同速度老化。昨天访问过的记忆比数周未触及的记忆更相关。我们使用一个三项加权公式对每条记忆进行评分，结合创建新鲜度、上次访问新鲜度和访问频率：

score = W_RECENCY * exp(-decay_rate * days_since_creation)
        + W_ACCESS * exp(-decay_rate * days_since_last_access)
        + W_FREQUENCY * min(access_count / MAX_ACCESS_BASELINE, 1.0)

我们不直接暴露原始衰减常数，而是提供一个直观的参数：pruneDays，即未访问记忆的评分降至相关阈值以下的大致天数：

import math

def decay_rate_from_prune_days(prune_days: int, threshold: float) -> float:
    """Convert pruneDays to an exponential decay rate.

decay_rate = -ln(threshold) / prune_days
    """
    if prune_days <= 0:
        raise ValueError(f"prune_days must be a positive integer, got: {prune_days}")
    if threshold <= 0 or threshold >= 1:
        raise ValueError(
            f"threshold must be in the open interval (0, 1), got: {threshold}"
        )
    return -math.log(threshold) / prune_days

使用默认值（pruneDays = 45，threshold = 0.3），这给出 decay_rate ≈ 0.02676。该公式产生的评分在 0.0–1.0 之间。当记忆评分低于你配置的相关阈值时，系统会根据你的策略设置将它们标记为待整合或待修剪。

该公式平衡了三种直觉：最近的记忆很重要，最近使用过的记忆更重要，频繁检索的记忆携带额外信号。指数衰减意味着评分在前几周急剧下降，然后趋于平稳。一条旧但最近且频繁访问的记忆仍然可以获得高分。

三个权重是可配置的，让运维人员可以根据其 AI 智能体的工作负载强调不同的信号：

• W_RECENCY（默认 0.4）：创建新鲜度的权重。较高的值更偏向较新的记忆。
• W_ACCESS（默认 0.35）：上次访问新鲜度的权重。较高的值更偏向最近检索的记忆。
• W_FREQUENCY（默认 0.25）：访问频率的权重。较高的值更偏向频繁检索的记忆。
• MAX_ACCESS_BASELINE（默认 50）：频率项饱和至 1.0 时的访问次数。设置为你回顾窗口内「重度使用」记忆累积的近似访问次数。

当三个权重之和为 1.0 时，评分将落在 [0.0, 1.0] 范围内。运维人员可以调整权重以匹配其 AI 智能体的需求。例如，为频繁访问的记忆最有价值的 AI 智能体增加 W_FREQUENCY（例如，一个反复引用同一故障排除手册的支持机器人），或为新鲜度最重要的 AI 智能体增加 W_RECENCY（例如，实时交易助手）。

正确的 pruneDays 值取决于你 AI 智能体的用例。下表为常见 AI 智能体原型提供了推荐起始点：

以下评分函数来自我们的 Memory Scorer AWS Lambda 函数（code/lambdas/memory_scorer/handler.py）：

def compute_relevance_score(
    created_at: datetime,
    last_accessed_at: datetime,
    access_count: int,
    decay_rate: float,
    now: datetime,
    w_recency: float = 0.4,
    w_access: float = 0.35,
    w_frequency: float = 0.25,
    max_access_baseline: int = 50,
) -> float:
    """Compute relevance score using the 3-term weighted decay formula.

score = w_recency * exp(-decay_rate * days_since_creation)
            + w_access * exp(-decay_rate * days_since_last_access)
            + w_frequency * min(access_count / max_access_baseline, 1.0)

当权重和为 1.0 时返回 [0.0, 1.0] 范围内的浮点数。 如果 max_access_baseline 为零或负数则抛出 ValueError。 """ if max_access_baseline <= 0: raise ValueError( f"max_access_baseline must be a positive integer, got: {max_access_baseline}" ) days_since_creation = max((now - created_at).total_seconds() / 86400, 0.0) days_since_last_access = max((now - last_accessed_at).total_seconds() / 86400, 0.0) recency_term = w_recency * math.exp(-decay_rate * days_since_creation) access_term = w_access * math.exp(-decay_rate * days_since_last_access) frequency_term = w_frequency * min(access_count / max_access_baseline, 1.0) score = recency_term + access_term + frequency_term return score

AgentCore 记忆 API 的 MemoryRecordSummary 中不包含 lastAccessedAt 字段。为了获取真实的访问数据，我们使用 AWS CloudTrail。CDK 堆栈配置了一条带有高级事件选择器的 CloudTrail 路径，用于捕获 GetMemoryRecord 数据事件。CloudTrail 配置会记录每次记忆检索及其 memoryRecordId 和时间戳，然后将日志传送到 Amazon Simple Storage Service（Amazon S3）存储桶。在每次评分调用开始时，Memory Scorer 从过去 25 小时内列出 CloudTrail 日志文件，解压缩并聚合 GetMemoryRecord 事件，生成每条记录的最后访问时间戳和访问计数查询表。为了在调用之间维护累积访问历史，评分器在 Amazon S3 中持久化一个访问台账。每次运行都会将新的 CloudTrail 计数与历史计数合并，使频率项获得真正的全生命周期信号，而不是狭窄的每日快照。

在修剪低分记忆之前，我们会给予它们最后一次机会。合并（Consolidation）使用 Amazon Bedrock 将相关记忆合并为一条紧凑的语义条目。五条关于部署偏好的情景记忆变成一条权威事实。在这一步骤中，大语言模型（LLM）会总结自己的记忆。合并提示词指示模型保留关键事实、去除冗余，并输出置信度分数：

```python
CONSOLIDATION_PROMPT_TEMPLATE = """You are a memory consolidation assistant.
Given the following agent memories, create a single concise summary that
preserves essential facts, user preferences, and actionable knowledge.
Remove redundancy and outdated information.

Memories:
{memory_contents}

Output a JSON object with:
- "summary": the consolidated memory text
- "confidence": a float 0.0-1.0 indicating consolidation quality
- "key_facts": list of preserved key facts"""

系统将合并后的记忆存储回 AgentCore 记忆，然后删除原始记忆。如果 Amazon Bedrock 失败，系统保留原始记忆不变。系统记录删除失败的操作供手动审查。合并本质上有损。LLM 将五条记忆总结为一条可能会丢失一些细节。模型返回的置信度分数有助于标记需要人工审查的低质量合并。对于高风险领域，考虑将原始记忆归档到冷存储而不是删除。

对于生产部署，请配置 Amazon Bedrock 护栏（Guardrails）以过滤有害内容，并使用接地检查（grounding checks）来验证合并后的记忆是否忠实于原始材料。这些控制是生产环境要求，不是可选附加项。

下图展示了夜间生命周期工作流架构。Amazon EventBridge 触发一个 AWS Step Functions 状态机，按顺序编排五个 Lambda 函数。

图 1：由 Amazon EventBridge 和 AWS Step Functions 编排的夜间记忆生命周期工作流

无障碍文本描述：Amazon EventBridge 规则每夜触发一个 Step Functions 状态机。状态机按顺序调用 Lambda 函数：Memory Pruner（TTL 过期）、Memory Scorer（使用 CloudTrail 访问数据进行相关性评分）、Memory Consolidator（通过 Amazon Bedrock 进行基于 LLM 的合并）、Metrics Emitter（Amazon CloudWatch 指标）和 Run Output Writer（S3 持久化）。失败路由到 Amazon Simple Notification Service（Amazon SNS）主题以发送告警。

工作流按以下步骤进行：

TTL 过期：Memory Pruner 查询 AgentCore 记忆中早于配置 TTL（默认：90 天）的记录并删除它们。

评分记忆：Memory Scorer 从 CloudTrail 日志构建每条记录的访问查找表，与持久化 S3 台账合并，计算相关性分数，并返回低于阈值的记忆。

合并：工作流将低分记忆分批（默认大小：10）发送到 Memory Consolidator，后者调用 Amazon Bedrock 将它们合并为紧凑的语义条目并删除原始记录。

发送指标：Metrics Emitter 将工作流指标（已处理、已合并、已修剪的记忆数）发布到 CloudWatch。

写入运行输出：Run Output Writer 将工作流结果持久化到 S3 以便追溯。如果任何步骤失败，Catch 块会路由到故障处理器，将错误详情发布到 Amazon SNS 主题。

在部署解决方案之前，请确认您具备以下条件：

一个 AWS 账户，拥有创建 Lambda 函数、Step Functions 状态机、Amazon EventBridge 规则、SNS 主题、CloudWatch 仪表板、CloudTrail 路径和 S3 存储桶的权限。

已安装 AWS CDK v2（npm install -g aws-cdk）。

Python 3.12 及 pip。

在目标区域已为 Claude Sonnet 4.5（anthropic.claude-sonnet-4-5-20250929-v1:0）启用 Amazon Bedrock 模型访问权限。请参阅 Amazon Bedrock 中的 AWS 区域支持的模型以验证可用性。

已配置至少一个启用记忆功能的 Amazon Bedrock AgentCore 代理。

已使用适当凭证配置 AWS Command Line Interface（AWS CLI）。

克隆代码库并安装依赖：

cd code
npm install

我们将整个生命周期编排为一个由 Amazon EventBridge 触发的夜间 AWS Step Functions 工作流。工作流按顺序运行五个阶段：TTL 过期、评分、合并、指标发送和运行输出写入。

CDK 堆栈详解

单个 CDK 堆栈（code/lib/memory-lifecycle-stack.ts）定义了整个基础设施。以下是关键部分。

Lambda 函数定义：每个处理程序使用 Python 3.12 并遵循最小权限 IAM 权限。堆栈将共享代码部署为 Lambda Layer，并通过环境变量传递可配置参数：

// Lambda Layer for the shared Python module (constants, models)
const sharedLayer = new lambda.LayerVersion(this, 'SharedLayer', {
  code: lambda.Code.fromAsset(
    path.join(__dirname, '..', 'lambdas', 'shared'),
    {
      bundling: {
        image: lambda.Runtime.PYTHON_3_12.bundlingImage,
        command: [
          'bash', '-c',
          'mkdir -p /asset-output/python/shared && cp -r . /asset-output/python/shared/',
        ],
      },
    },
  ),
  compatibleRuntimes: [lambda.Runtime.PYTHON_3_12],
  description: 'Shared constants and models for memory lifecycle Lambdas',
});

const memoryScorerFn = new lambda.Function(this, 'MemoryScorerFunction', {
  runtime: lambda.Runtime.PYTHON_3_12,
  handler: 'handler.handler',
  code: lambda.Code.fromAsset(
    path.join(__dirname, '..', 'lambdas', 'memory_scorer')
  ),
  layers: [sharedLayer],
  timeout: cdk.Duration.minutes(5),
  environment: {
    MEMORY_TTL_DAYS: String(memoryTtlDays),
    RELEVANCE_THRESHOLD: String(relevanceThreshold),
    CONSOLIDATION_BATCH_SIZE: String(consolidationBatchSize),
    BEDROCK_MODEL_ID: bedrockModelId,
    PRUNE_DAYS: String(pruneDays),
    TRAIL_BUCKET_NAME: trailBucket.bucketName,
    TRAIL_LOOKBACK_HOURS: '25',
    W_RECENCY: String(wRecency),
    W_ACCESS: String(wAccess),
    W_FREQUENCY: String(wFrequency),
    MAX_ACCESS_BASELINE: String(maxAccessBaseline),
  },
});

AWS Identity and Access Management（IAM）最小权限：Memory Scorer 只能列出记忆。Consolidator 可以读取、创建、删除记忆并调用 Amazon Bedrock。Pruner 可以列出和删除：

// Memory Scorer: list records only (read-only)
memoryScorerFn.addToRolePolicy(new iam.PolicyStatement({
  effect: iam.Effect.ALLOW,
  actions: ['bedrock-agentcore:ListMemoryRecords'],
  resources: [
    `arn:aws:bedrock-agentcore:${this.region}:${this.account}:memory/*`,
  ],
}));

// Memory Consolidator: full memory record CRUD + Bedrock
memoryConsolidatorFn.addToRolePolicy(new iam.PolicyStatement({
  effect: iam.Effect.ALLOW,
  actions: [
    'bedrock-agentcore:GetMemoryRecord',
    'bedrock-agentcore:BatchCreateMemoryRecords',
    'bedrock-agentcore:DeleteMemoryRecord',
  ],
  resources: [
    `arn:aws:bedrock-agentcore:${this.region}:${this.account}:memory/*`,
  ],
}));

memoryConsolidatorFn.addToRolePolicy(new iam.PolicyStatement({
  effect: iam.Effect.ALLOW,
  actions: ['bedrock:InvokeModel'],
  resources: [
    `arn:aws:bedrock:${this.region}::foundation-model/${bedrockModelId}`,
  ],
}));

Step Functions 工作流：状态机将 TTL 过期、评分、低分记忆的 Choice 分支、批量整合（Map 状态）、指标发送和运行输出写入串联在一起：

// Chain EmitMetrics -> WriteRunOutput once (both branches converge here)
const emitAndWrite = emitMetrics.next(writeRunOutput);

const definition = ttlExpiration
  .next(scoreMemories)
  .next(
    checkLowScoreMemories
      .when(
        sfn.Condition.isPresent('$.scoringResult.below_threshold[0]'),
        batchConsolidate.next(emitAndWrite),
      )
      .otherwise(emitAndWrite),
  );

const stateMachine = new sfn.StateMachine(this, 'MemoryLifecycleStateMachine', {
  definitionBody: sfn.DefinitionBody.fromChainable(definition),
  timeout: cdk.Duration.hours(1),
  tracingEnabled: true,
});

夜间触发器：每天凌晨 2:00 UTC，一条 Amazon EventBridge 规则触发该工作流：

new events.Rule(this, 'NightlyMemoryLifecycleRule', {
  schedule: events.Schedule.expression('cron(0 2 * * ? *)'),
  targets: [new targets.SfnStateMachine(stateMachine)],
});

所有可配置参数（memoryTtlDays、relevanceThreshold、consolidationBatchSize、pruneDays、bedrockModelId 以及评分权重）均从 CDK context 读取，因此可以在部署时调整这些参数而无需修改代码：

npx cdk deploy \
  -c memoryTtlDays=60 \
  -c relevanceThreshold=0.25 \
  -c consolidationBatchSize=15 \
  -c pruneDays=45 \
  -c wRecency=0.4 \
  -c wAccess=0.35 \
  -c wFrequency=0.25 \
  -c maxAccessBaseline=50

主要成本驱动因素是整合期间的 Amazon Bedrock 调用。对于拥有 1,000 条记忆且 20% 评分低于阈值的智能体，每次夜间运行预计约需 20 次 Bedrock 调用（约 $0.01–$0.02）。当记忆数量达到 100,000 条时，每月成本可能达到 $50–$100。建议从较高的相关性阈值开始以限制整合量，并查看 Amazon Bedrock 定价以了解具体工作负载的费用。

测试记忆质量

剪枝和整合只有在智能体之后仍能正确回答问题时才有用。我们使用回归测试套件来衡量生命周期操作是否降低了响应质量。

我们将测试用例定义为问答-标准对（code/test/test_regression_suite.py）。每个测试用例指定一个问题、智能体的响应应满足的标准，以及最低质量分数：

DEFAULT_TEST_FIXTURES = [
    {
        "question": "What are the user's preferred programming languages?",
        "expected_criteria": "Response mentions specific languages previously discussed with the user",
        "min_quality_score": 0.7,
    },
    {
        "question": "Summarize the last project we worked on together.",
        "expected_criteria": "Response includes project name, key milestones, and outcome",
        "min_quality_score": 0.6,
    },
]

回归测试套件遵循前后对比模式：

基线：在生命周期运行前，使用每个测试问题查询智能体。使用 AgentCore Evaluations（Amazon Bedrock AgentCore 的一项能力）记录质量分数。

运行生命周期：执行夜间工作流（评分、整合、剪枝）。

生命周期后：使用相同问题再次查询智能体。记录新的质量分数。

评估：如果生命周期后的分数达到或超过配置的最低分数，则测试用例通过。我们还计算质量变化量（post_lifecycle_score - baseline_score）用于报告。

def determine_pass_fail(test_case: RegressionTestCase) -> RegressionTestCase:
    if test_case.post_lifecycle_score is None:
        test_case.passed = None
        return test_case
    test_case.passed = test_case.post_lifecycle_score >= test_case.min_quality_score
    return test_case

回归测试套件与 Amazon Bedrock AgentCore Evaluations 集成，以编程方式计算质量分数。AgentCore Evaluations 作为 LLM 即评判系统工作：您提供智能体的响应和人类定义的标准，服务返回 0.0 到 1.0 之间的归一化质量分数。这使得套件完全自动化，适用于持续集成和持续交付（CI/CD）流水线。

运行套件会生成每个测试用例的报告，将基线和生命周期后的分数配对，以便一目了然地看到质量变化：

Memory regression suite (2 test cases)
------------------------------------------------------------
[PASS] Preferred programming languages
baseline=0.82 post=0.85 delta=+0.03 min=0.70
[PASS] Summary of last project
baseline=0.74 post=0.71 delta=-0.03 min=0.60
------------------------------------------------------------
Result: 2/2 passed

在此示例运行中，两个测试用例均保持在配置的最低分数之上。只有当生命周期后的分数低于其 min_quality_score 时，测试用例才会失败，这表示剪枝或整合过度。

隐私与合规

记忆生命周期管理不仅关乎性能，还是一项合规要求。当您的智能体在记忆中存储个人数据时，您需承担 GDPR 等法规下的义务。

专用的 GDPR 删除处理程序（code/lambdas/gdpr_deletion/handler.py）删除特定用户的所有记忆。它列出该用户在 AgentCore 记忆中的每一条记忆并逐一删除：

def handler(event: dict, context) -> dict:
    user_id = event["user_id"]
    memory_id = event["memory_id"]
    client = boto3.client("bedrock-agentcore")

response = client.list_memory_records(
        memoryId=memory_id,
        namespace=user_id,
    )
    memories = response.get("memoryRecordSummaries", [])

deleted_count = 0
    failed_memory_ids = []
    for memory in memories:
        record_id = memory["memoryRecordId"]
        try:
            client.delete_memory_record(memoryId=memory_id, memoryRecordId=record_id)
            deleted_count += 1
            logger.info(json.dumps({
                "action": "gdpr_delete",
                "user_id": user_id,
                "memory_id": record_id,
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }))
        except Exception as exc:
            failed_memory_ids.append(record_id)

status = "success" if le