企业官网建设流程全解析

Hindsight AgentCore 集成实战：为 Amazon Bedrock AgentCore Runtime 智能体构建跨会话持久记忆

【免费下载链接】hindsightHindsight: Agent Memory That Learns项目地址: https://gitcode.com/GitHub_Trending/hindsight2/hindsight

Amazon Bedrock AgentCore Runtime 的会话天然是短命的——会话因不活动而终止、环境被重新供给，导致智能体在每次新会话中"失忆"。Hindsight 仓库中的hindsight-agentcore包正是为解决这个问题而生：它以before_turn()（回忆相关记忆）和after_turn()（异步保留输出）两个钩子包裹 AgentCore Runtime 调用，把记忆绑定到稳定的用户身份而非runtimeSessionId，让智能体在任意多次会话切换之后依然记得用户、决策与已学习的模式。读完本文，你将掌握该包的完整配置参数、两种检索模式（recall/reflect）、自定义 bank 解析、失败降级行为，并能将其落地到自己的 AgentCore Runtime handler 中。

问题背景：Runtime 会话的"失忆"困境

根据 hindsight-integrations/agentcore/README.md 的说明，AgentCore Runtime 的会话是显式临时（ephemeral）的：会话在不活动时终止，环境重新供给为全新状态。hindsight-agentcore在其上叠加了一层持久的跨会话记忆。其工作机制如下：

AgentCore Runtime invocation │ ▼ before_turn() ← 从 Hindsight 召回相关记忆 │ ▼ Agent executes ← Prompt 被先前上下文增强 │ ▼ after_turn() ← 将输出异步保留到 Hindsight

这里的关键设计是：记忆键控（keyed）于稳定的用户身份，而不是runtimeSessionId。Bank（Hindsight 中的记忆库）在会话更替（session churn）中存活。默认 bank 格式为：

tenant:{tenant_id}:user:{user_id}:agent:{agent_name}

安装与核心 API

pip install hindsight-agentcore

运行前提（见 pyproject.toml）：

• Python 3.10+（requires-python = ">=3.10"，classifiers 覆盖 3.10/3.11/3.12）
• 依赖hindsight-client>=0.4.0
• MIT 许可证，当前版本 0.1.1，Development Status 为 Beta

包的公开 API 定义在 hindsight_agentcore/init.py 的__all__中，包括HindsightRuntimeAdapter、RecallPolicy、RetentionPolicy、TurnContext、BankResolver、default_bank_resolver、configure、get_config、reset_config、HindsightAgentCoreConfig以及异常类型HindsightAgentCoreError/BankResolutionError。

快速上手：配置 + 适配器 + Handler 接线

推荐的接入方式是使用 Hindsight Cloud（注册后一分钟即可获取 API key），无需自托管；也支持自托管。以下是文档给出的 Quick Start 完整示例：

import os from hindsight_agentcore import HindsightRuntimeAdapter, TurnContext, configure configure( hindsight_api_url="https://api.hindsight.vectorize.io", api_key=os.environ["HINDSIGHT_API_KEY"], ) adapter = HindsightRuntimeAdapter(agent_name="support-agent") # Your AgentCore Runtime handler async def handler(event: dict) -> dict: context = TurnContext( runtime_session_id=event["sessionId"], user_id=event["userId"], # 来自已验证的认证——绝不接受客户端自报值 agent_name="support-agent", tenant_id=event.get("tenantId"), request_id=event.get("requestId"), ) result = await adapter.run_turn( context=context, payload={"prompt": event["prompt"]}, agent_callable=run_my_agent, ) return result async def run_my_agent(payload: dict, memory_context: str) -> dict: prompt = payload["prompt"] if memory_context: prompt = f"Past context:\n{memory_context}\n\nCurrent request: {prompt}" output = await call_bedrock(prompt) return {"output": output}

仓库中还提供了一个可直接本地运行的完整示例 examples/basic_runtime_handler.py，它模拟了同一用户在两个不同sessionId下的两轮对话——第二轮开启新 Runtime 会话后，第一轮记住的偏好（"user-alex 偏好邮件而非电话"）依然可以被召回，直观演示了"bank 在会话更替中存活"这一核心能力。

从源码看，run_turn()的执行链在 adapter.py 中清晰可见：先从payload取出query_key（默认"prompt"）作为查询，调用before_turn()得到memory_context字符串，再将其注入你的agent_callable，最后从结果字典的result_key（默认"output"）提取输出并调用after_turn()。两个 key 均可通过参数覆盖，便于适配不同形状的事件负载。

低级钩子：手动控制 recall → execute → retain

如果你需要更细粒度的控制（例如想在两次回忆之间插入其他逻辑），可以直接调用三个钩子：

# 手动 recall → execute → retain memory_context = await adapter.before_turn(context, query=user_message) result = await run_my_agent(payload, memory_context=memory_context) await adapter.after_turn(context, result=result["output"], query=user_message)

before_turn()的行为细节（见 adapter.py）：

• 空 query（仅空白）直接返回""，不发起任何 Hindsight 调用；
• recall 模式下调用客户端的arecall(bank_id, query, budget, max_tokens)，结果经_format_memories()格式化为项目符号列表（每项形如- 文本 [类型] (提及时间)，多条之间以空行分隔）；
• 任何异常（网络不可达、超时等）都会被捕获，记录 warning 后返回""——"记忆是增强，不是基础设施"（graceful degradation）；
• after_turn()对空结果直接跳过；保留的内容默认会把用户消息拼进正文，格式为User: {query}\nAssistant: {result}（可通过RetentionPolicy.include_user_message=False关闭）。

这些行为在 tests/test_adapter.py 中有逐条对应的测试用例，如test_empty_query_returns_empty_string、test_gracefully_degrades_on_exception、test_retained_content_includes_user_message等。

检索模式：recall 与 reflect

适配器支持两种记忆检索策略，通过RecallPolicy控制（定义见 adapter.py）：

Recall（默认）

快速的多策略检索（语义 + 关键词 + 图 + 时间）：

from hindsight_agentcore import RecallPolicy adapter = HindsightRuntimeAdapter( recall_policy=RecallPolicy(mode="recall", budget="mid", max_tokens=1500) )

Reflect

由 LLM 合成的上下文，适合复杂推理任务：

adapter = HindsightRuntimeAdapter( recall_policy=RecallPolicy(mode="reflect") )

文档明确建议选择性使用 reflect——它更慢，应保留给显式规划步骤或路由决策。源码中，mode == "reflect"时before_turn()会改走客户端的areflect()并直接返回resp.answer；测试用例test_reflect_mode_calls_reflect验证了此时arecall不会被调用。

异步保留：默认不阻塞用户回合

默认情况下，after_turn()将保留（retention）作为后台任务发起——用户回合永远不会被记忆写入拖慢：

configure(retain_async=True) # 默认 configure(retain_async=False) # 返回前等待保留完成

从源码实现看（adapter.py），retain_async=True时通过asyncio.create_task()发起 fire-and-forget 任务，并用一个self._pending: set[asyncio.Task]持有强引用——因为 asyncio 对任务只保持弱引用，不加保护的话后台保留任务可能在执行中途被 GC 回收。任务完成后的 done callback 会将其移出集合。测试test_pending_task_tracked_and_completes专门验证了"任务被跟踪 → 完成后自动移除 →aretain恰好被调用一次"的完整生命周期。

保留时写入 Hindsight 的document_id默认为request_id（若提供），否则回退为{runtime_session_id}:{user_id}的组合，用于追踪与去重。

跨会话的长时任务

对于跨越多个 Runtime 会话的作业（例如多天的 QBR 分析），文档建议在任务开始与完成时各保留一次：

# 任务开始 await adapter.after_turn( context, result="Started QBR analysis for Acme Corp", query=task_description, ) # ... 跨越多个潜在会话的长时工作 ... # 任务完成 await adapter.after_turn( context, result=f"Completed QBR analysis. Finding: {summary}", query=task_description, )

这样，即使任务中途 Runtime 会话被重新供给，下一次会话开始时的 recall 也能命中"任务已开始"的记录，智能体得以续接上下文。

身份与认证：bank 键控的三条铁律

绝不要把runtimeSessionId用作 bank ID。会话会过期，记忆必须扛过会话更替。文档给出的身份来源优先级为：

1. 来自 AgentCore JWT/OAuth 上下文的已验证用户 ID；
2. X-Amzn-Bedrock-AgentCore-Runtime-User-Id请求头；
3. 受信服务端部署中由应用提供的用户 ID。

context = TurnContext( runtime_session_id=event["sessionId"], user_id=jwt_claims["sub"], # 来自已验证令牌的稳定身份 agent_name="support-agent", tenant_id=jwt_claims.get("tenant"), )

TurnContext的字段语义（见 bank.py）值得注意：runtime_session_id仅作为 metadata/标签存在，不作为 bank 主键；它通过as_metadata()写入每条保留记忆的 metadata（同时写入channel: "agentcore-runtime"、user_id、agent_name，可选tenant_id/request_id），通过as_tags()生成tenant:*、user:*、agent:*、session:*形式的标签——tenant 标签总是排在首位以便正确过滤。这些行为均有 tests/test_bank.py 中的用例锁定。

配置参考

全局配置通过应用启动时调用一次configure()完成（在创建任何 adapter 之前）。完整参数表（含环境变量回退与默认值）：

结合 config.py 源码可以补充两个实现细节：

• api_key除了HINDSIGHT_API_KEY外，还会回退读取HINDSIGHT_API_TOKEN（resolved_key的三级回退：显式参数 →HINDSIGHT_API_KEY→HINDSIGHT_API_TOKEN）；
• 未调用configure()时，adapter 自身还有兜底默认值（URL 回退到 Hindsight Cloud、budget"mid"、max_tokens1500、retain_async=True、timeout=15.0），因此最小化接入可以完全省略configure()。

配置优先级整体为：adapter 构造参数 >configure()全局配置 > 环境变量 > 内置默认。全局状态可被reset_config()重置（主要供测试使用，tests/ 中的每个测试类都在 setup/teardown 中调用它做隔离）。

自定义 Bank 解析与失败关闭（Fail Closed）

默认的 default_bank_resolver 按"每 (tenant, user, agent) 元组一个 bank"的规则工作：

有 tenant: tenant:{tenant_id}:user:{user_id}:agent:{agent_name} 无 tenant: user:{user_id}:agent:{agent_name}

测试 tests/test_bank.py 断言了两种形态的精确输出，并专门验证了runtimeSessionId永远不会出现在 bank ID 中。

要覆盖默认的tenant:user:agent格式，只需实现BankResolver协议（TurnContext -> str）：

from hindsight_agentcore import TurnContext def my_resolver(context: TurnContext) -> str: return f"acme:{context.user_id}:{context.agent_name}" adapter = HindsightRuntimeAdapter(bank_resolver=my_resolver)

安全规则：解析器必须失败关闭（fail closed）——当身份缺失时抛出BankResolutionError，而不是让记忆跨用户泄漏。默认的default_bank_resolver正是如此实现：user_id或agent_name为空/仅空白时立即抛BankResolutionError（异常定义在 errors.py）。而在适配层，bank 解析失败会被捕获：before_turn()记录日志并返回""，_retain()记录日志并跳过写入——两种情况下记忆操作都被跳过，但绝不会回退到可能串用户的 bank。

失败模式与降级行为

文档给出的失败行为契约（由源码与测试双重印证）：

源码中before_turn()与_retain()分别用except Exception包裹客户端调用并记录exc_info（adapter.py、adapter.py），保证记忆子系统的所有故障都停留在日志层面，不向用户回合冒泡。对应的测试用例包括test_gracefully_degrades_on_exception（before_turn 与 after_turn 各有一条）和test_session_id_not_in_bank_id。

部署方式与保留内容细节

部署选项（来自 README）：

• Hindsight Cloud：注册后把hindsight_api_url指向你的 Cloud endpoint；
• AWS 上自托管：在 ECS/EKS 上运行 Hindsight 并搭配 RDS PostgreSQL（pgvector），网络路径完全留在你的 AWS 账户内。

此外，如果你想在保留的记忆上附加更多上下文，可用RetentionPolicy（adapter.py）控制：

最终aretain()的标签集合 =context.as_tags()+ 全局tags+policy.extra_tags，metadata 则是context.as_metadata()与policy.extra_metadata的合并（adapter.py），测试test_tags_include_user_agent_session与test_extra_tags_from_retention_policy锁定了这一合并顺序。

小结

hindsight-agentcore用一套极薄的钩子（before_turn/after_turn/run_turn）把 Hindsight 的持久记忆接入了 AgentCore Runtime：

1. 身份解耦会话：bank 键控于(tenant, user, agent)而非临时runtimeSessionId，会话更替不丢失记忆；
2. 双模式检索：默认recall（多策略、快），复杂场景切reflect（LLM 合成、慢）；
3. 不阻塞用户回合：默认异步保留，且 fire-and-forget 任务有强引用保护；
4. 失败全部降级：记忆层任何故障都只影响日志，绝不影响 agent 主链路，bank 解析失败则严格失败关闭。

完整代码、测试与示例可分别查看 hindsight_agentcore/ 源码目录、tests/ 测试目录（含test_adapter.py、test_bank.py、test_config.py及一个需真实 Hindsight 服务的test_live_integration.py）以及 examples/basic_runtime_handler.py 可直接本地运行的冒烟示例。

【免费下载链接】hindsightHindsight: Agent Memory That Learns项目地址: https://gitcode.com/GitHub_Trending/hindsight2/hindsight