Hindsight AgentCore 集成实战为 Amazon Bedrock AgentCore Runtime 智能体构建跨会话持久记忆【免费下载链接】hindsightHindsight: Agent Memory That Learns项目地址: https://gitcode.com/GitHub_Trending/hindsight2/hindsightAmazon 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。BankHindsight 中的记忆库在会话更替session churn中存活。默认 bank 格式为tenant:{tenant_id}:user:{user_id}:agent:{agent_name}安装与核心 APIpip install hindsight-agentcore运行前提见 pyproject.tomlPython 3.10requires-python 3.10classifiers 覆盖 3.10/3.11/3.12依赖hindsight-client0.4.0MIT 许可证当前版本 0.1.1Development 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_urlhttps://api.hindsight.vectorize.io, api_keyos.environ[HINDSIGHT_API_KEY], ) adapter HindsightRuntimeAdapter(agent_namesupport-agent) # Your AgentCore Runtime handler async def handler(event: dict) - dict: context TurnContext( runtime_session_idevent[sessionId], user_idevent[userId], # 来自已验证的认证——绝不接受客户端自报值 agent_namesupport-agent, tenant_idevent.get(tenantId), request_idevent.get(requestId), ) result await adapter.run_turn( contextcontext, payload{prompt: event[prompt]}, agent_callablerun_my_agent, ) return result async def run_my_agent(payload: dict, memory_context: str) - dict: prompt payload[prompt] if memory_context: prompt fPast 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, queryuser_message) result await run_my_agent(payload, memory_contextmemory_context) await adapter.after_turn(context, resultresult[output], queryuser_message)before_turn()的行为细节见 adapter.py空 query仅空白直接返回不发起任何 Hindsight 调用recall 模式下调用客户端的arecall(bank_id, query, budget, max_tokens)结果经_format_memories()格式化为项目符号列表每项形如- 文本 [类型] (提及时间)多条之间以空行分隔任何异常网络不可达、超时等都会被捕获记录 warning 后返回——记忆是增强不是基础设施graceful degradationafter_turn()对空结果直接跳过保留的内容默认会把用户消息拼进正文格式为User: {query}\nAssistant: {result}可通过RetentionPolicy.include_user_messageFalse关闭。这些行为在 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字段类型默认值含义modestrrecallrecall为确定性多策略检索reflect为 LLM 合成上下文budgetstr \| NoneNone解析为midHindsight 检索深度low/mid/highmax_tokensint \| NoneNone解析为1500召回记忆块的最大 token 数Recall默认快速的多策略检索语义 关键词 图 时间from hindsight_agentcore import RecallPolicy adapter HindsightRuntimeAdapter( recall_policyRecallPolicy(moderecall, budgetmid, max_tokens1500) )Reflect由 LLM 合成的上下文适合复杂推理任务adapter HindsightRuntimeAdapter( recall_policyRecallPolicy(modereflect) )文档明确建议选择性使用 reflect——它更慢应保留给显式规划步骤或路由决策。源码中mode reflect时before_turn()会改走客户端的areflect()并直接返回resp.answer测试用例test_reflect_mode_calls_reflect验证了此时arecall不会被调用。异步保留默认不阻塞用户回合默认情况下after_turn()将保留retention作为后台任务发起——用户回合永远不会被记忆写入拖慢configure(retain_asyncTrue) # 默认 configure(retain_asyncFalse) # 返回前等待保留完成从源码实现看adapter.pyretain_asyncTrue时通过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, resultStarted QBR analysis for Acme Corp, querytask_description, ) # ... 跨越多个潜在会话的长时工作 ... # 任务完成 await adapter.after_turn( context, resultfCompleted QBR analysis. Finding: {summary}, querytask_description, )这样即使任务中途 Runtime 会话被重新供给下一次会话开始时的 recall 也能命中任务已开始的记录智能体得以续接上下文。身份与认证bank 键控的三条铁律绝不要把runtimeSessionId用作 bank ID。会话会过期记忆必须扛过会话更替。文档给出的身份来源优先级为来自 AgentCore JWT/OAuth 上下文的已验证用户 IDX-Amzn-Bedrock-AgentCore-Runtime-User-Id请求头受信服务端部署中由应用提供的用户 ID。context TurnContext( runtime_session_idevent[sessionId], user_idjwt_claims[sub], # 来自已验证令牌的稳定身份 agent_namesupport-agent, tenant_idjwt_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 之前。完整参数表含环境变量回退与默认值选项环境变量默认值说明hindsight_api_urlHINDSIGHT_API_URLHindsight CloudHindsight 服务器 URLapi_keyHINDSIGHT_API_KEY—Hindsight Cloud 的 API keyrecall_budget—mid检索深度low、mid、highrecall_max_tokens—1500召回记忆的最大 token 数retain_async—True非阻塞保留timeout—15.0Hindsight API 调用的 HTTP 超时秒tags—[]附加到所有保留记忆上的标签verbose—False记录记忆操作日志结合 config.py 源码可以补充两个实现细节api_key除了HINDSIGHT_API_KEY外还会回退读取HINDSIGHT_API_TOKENresolved_key的三级回退显式参数 →HINDSIGHT_API_KEY→HINDSIGHT_API_TOKEN未调用configure()时adapter 自身还有兜底默认值URL 回退到 Hindsight Cloud、budgetmid、max_tokens1500、retain_asyncTrue、timeout15.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 - strfrom hindsight_agentcore import TurnContext def my_resolver(context: TurnContext) - str: return facme:{context.user_id}:{context.agent_name} adapter HindsightRuntimeAdapter(bank_resolvermy_resolver)安全规则解析器必须失败关闭fail closed——当身份缺失时抛出BankResolutionError而不是让记忆跨用户泄漏。默认的default_bank_resolver正是如此实现user_id或agent_name为空/仅空白时立即抛BankResolutionError异常定义在 errors.py。而在适配层bank 解析失败会被捕获before_turn()记录日志并返回_retain()记录日志并跳过写入——两种情况下记忆操作都被跳过但绝不会回退到可能串用户的 bank。失败模式与降级行为文档给出的失败行为契约由源码与测试双重印证失败场景行为Hindsight 不可用before_turn()返回agent 继续执行Recall 超时返回agent 继续执行Retain 失败记录为 warning用户回合不受影响Bank 解析失败失败关闭——无跨用户记忆泄漏源码中before_turn()与_retain()分别用except Exception包裹客户端调用并记录exc_infoadapter.py、adapter.py保证记忆子系统的所有故障都停留在日志层面不向用户回合冒泡。对应的测试用例包括test_gracefully_degrades_on_exceptionbefore_turn 与 after_turn 各有一条和test_session_id_not_in_bank_id。部署方式与保留内容细节部署选项来自 READMEHindsight Cloud注册后把hindsight_api_url指向你的 Cloud endpointAWS 上自托管在 ECS/EKS 上运行 Hindsight 并搭配 RDS PostgreSQLpgvector网络路径完全留在你的 AWS 账户内。此外如果你想在保留的记忆上附加更多上下文可用RetentionPolicyadapter.py控制字段默认值含义context_labelagentcore-runtime:conversation_turn随每条保留记忆存储的来源标签extra_tags[]在默认 TurnContext 标签之外的附加标签extra_metadata{}在默认 metadata 之外的附加元数据include_user_messageTrue是否把用户消息拼接到保留内容前最终aretain()的标签集合 context.as_tags() 全局tagspolicy.extra_tagsmetadata 则是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身份解耦会话bank 键控于(tenant, user, agent)而非临时runtimeSessionId会话更替不丢失记忆双模式检索默认recall多策略、快复杂场景切reflectLLM 合成、慢不阻塞用户回合默认异步保留且 fire-and-forget 任务有强引用保护失败全部降级记忆层任何故障都只影响日志绝不影响 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考