View a markdown version of this page

LangGraph 智能体的语义长期记忆 - Amazon DynamoDB
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

LangGraph 智能体的语义长期记忆

LangGraph 将智能体状态分为两种类型。短期状态是会话线程本身,由检查指针持久化,以便线程可以继续、重放和恢复(请参阅将 DynamoDB 用作 LangGraph 代理的检查点存储)。长期记忆是智能体跨线程所掌握的内容,LangGraph 将其建模为存储:一个限定了命名空间的键值接口,智能体可有意写入并在之后读取。

langgraph-checkpoint-aws 程序包中的 DynamoDBStore 类是该存储的 DynamoDB 实现。它通过分层命名空间处理键值端,支持使用生存时间使陈旧的记忆过期,并提供基本筛选。配置向量索引后(请参阅在 DynamoDB 中使用向量索引),其 search() 方法可执行语义搜索:记忆在写入时嵌入,异步编制索引,并按语义从保存记忆的同一个表中检索(按相似性排序)。无需预调配单独的向量数据库,也无需通过管道将数据复制到其中。

先决条件

  • 有权创建 DynamoDB 表和调用 Amazon Bedrock 模型的 Amazon Web Services 账户

  • 可以访问您所在区域的 Amazon Bedrock 中的嵌入模型。这些示例使用 Amazon Titan Text Embeddings V2

  • Python 3.10 或更高版本,以及 langgraph-checkpoint-aws 1.2.2 或更高版本和 boto3 1.43.64 或更高版本(早期 boto3 版本没有 SearchVectors 操作)

使用 pip 安装库:

pip install langgraph langgraph-checkpoint-aws langchain-aws

设置存储

使用 index 块配置存储并调用 setup():

from langchain_aws import BedrockEmbeddings from langgraph_checkpoint_aws import DynamoDBStore store = DynamoDBStore( table_name="support-agent-memory", region_name="us-east-1", index={ "embed": BedrockEmbeddings(model_id="amazon.titan-embed-text-v2:0"), "dims": 1024, "fields": ["text"], "distance_function": "COSINE", }, ) store.setup()

index 块中有四个设置值得了解:

  • embed 接受任何 LangChain 嵌入对象,也接受一个将字符串列表映射到向量列表的普通可调用对象。

  • dims 必须与模型的输出大小相匹配。Titan Text Embeddings V2 默认返回 1024 个维度。如果这些不一致,存储会在第一次写入时引发维度不匹配错误并列出两个数字,而不是写入索引无法使用的向量。

  • fields 用于选择要嵌入值的哪些部分。此处仅嵌入 text 字段。默认值 ["$"] 会将整个值序列化为 JSON 并嵌入该值,这很方便,但也会嵌入您的记录属性。

  • distance_function 默认为 COSINE,也接受 EUCLIDEAN 和 DOT_PRODUCT。

setup() 会在表不存在时创建该表,附加向量索引,并在已配置生存时间的情况下启用生存时间。对于现有表,它仅添加缺少的内容,因此现有的 DynamoDBStore 部署无需重新创建表或迁移条目,即可采用语义搜索。请注意,setup() 会等到索引报告 ACTIVE 且不再回填后才返回:当仍回填时报告 ACTIVE 的索引会拒绝 SearchVectors 调用。对于新的空表,等待时间很短。对含有现有项目的表进行改造所需的时间,与回填所需的时间相同。

写入记忆

写入是普通的 put()。将发生嵌入:

namespace = ("memories", "acme-corp", "user-8812") store.put(namespace, "mem-1", { "text": "Account runs a proxy that closes idle sockets after 60 seconds", "source": "ticket-4471", }) store.put(namespace, "mem-2", { "text": "Customer prefers email, asked not to be called by phone", "source": "ticket-4471", }) store.put(namespace, "mem-3", { "text": "Uses a custom build of the SDK pinned to version 2.14", "source": "ticket-4502", })

按语义检索

客户发起了一个新对话,并表示其连接在大约一分钟后持续断开:

results = store.search( namespace, query="connection drops after a minute of inactivity", limit=3, ) for item in results: print(round(item.score, 3), item.value["text"])

输出:

0.324 Account runs a proxy that closes idle sockets after 60 seconds 0.039 Customer prefers email, asked not to be called by phone 0.031 Uses a custom build of the SDK pinned to version 2.14

相关的记忆排在第一位,而查询中的键与任何内容均不匹配。SearchItem.score 遵循 LangGraph 约定:排名越高,相关性越强。DynamoDB 返回的是距离,距离越小表示越相近,因此存储会进行转换:COSINE 的分数为 1 - distance,EUCLIDEAN 的分数为 1 / (1 + distance),DOT_PRODUCT 的分数则保持不变。如果您使用绝对分数来判断某条记忆是否足够相关,以便可以注入提示,请根据您自己的数据和所选的距离函数来校准该阈值。

使用搜索架构限定记忆范围

当 DynamoDBStore 创建向量索引时,它将表的分区键声明为索引搜索架构中的 HASH 元素。由于该元素存在,因此每次搜索都必须为其提供条件,而存储则提供命名空间:命名空间元组联接后形成分区键值,因此每个语义搜索都固定到恰好一个命名空间。由此产生三个结果:

  • 隔离是结构性的,而不是您记得要编写的筛选条件。搜索无法访问不同命名空间中的记忆,因此,为多个租户提供服务的存储,不存在任何可返回其它租户的记忆的查询方式。这是查询范围界定,而非授权:对表拥有 dynamodb:SearchVectors 权限的主体可以直接搜索任何命名空间值,因此,决定调用方可读取哪些命名空间仍属于 IAM 和应用程序授权层的职责范围。

  • 检索成本仅跟踪单个用户的记忆,而非整个表。搜索工作受一个命名空间可容纳的内容量所限制,而不是受整个产品可容纳的记忆数量限制,这使得延迟和向量搜索成本可随增长而保持较小和稳定。

  • 语义搜索基于完全的命名空间,而非前缀。搜索会精确到达您传递的命名空间,而不会触及其下的任何内容。搜索 ("memories", "acme-corp") 不会访问 ("memories", "acme-corp", "user-8812"),因为它们是不同的分区键值。您的命名空间即为检索范围,因此,请根据您希望单次搜索所覆盖的范围来选择命名空间。

下表总结了如何选择命名空间结构。

命名空间结构

search() 查询范围

适用场景

("memories", user_id)

该用户的记忆

单租户产品(按用户检索)

("memories", tenant_id, user_id)

指定租户下的该用户

多租户(常见情况)

("memories", tenant_id, user_id, agent_name)

某个智能体关于该用户的备注

多个专用智能体,且不应互相读取彼此的备注

("account_facts", tenant_id)

租户范围的知识

适用于账户中每位用户的事实

避免将每个记忆都放在一个命名空间中并对元数据进行筛选:筛选条件是在 DynamoDB 已从整个命名空间中选出最近邻匹配项之后才应用的,且单次搜索最多返回前 100 个匹配项。因此,一旦某个高负载租户的常见查询结果占满前 100 位,即使其它租户的记忆存在,较低负载租户的搜索返回的结果数也会少于请求的数量。对于任何影响正确性的场景,应优先使用命名空间范围,而非筛选条件。粒度细于检索边界同样是一种合理的设计:如果智能体同时需要用户特定的上下文和账户范围的上下文,则可同时搜索这两个命名空间并合并结果。

注意事项

  • 该索引是最终一致的,与全局二级索引的模型相同。紧跟在 put() 后发出的 search() 可能尚未包含新记忆。对于在同一轮次内写入记忆后又检索该记忆的智能体,请改为按键直接读取。

  • 单次搜索最多返回前 100 个匹配项。limit 与 offset 之和超过该上限的请求将遭拒绝并返回明确的错误,而不是在超出上限时,采用静默方式而不返回任何内容。

  • 如果您使用生存时间来使陈旧的记忆过期并启用读取时刷新,则智能体实际检索的记忆的到期时间会顺延,因此正在使用中的记忆不会悄悄消失。

  • 此时会引发向量搜索失败,以便您的应用程序能够做出响应。如果存储在失败时返回空列表,则节流、权限问题和真正为空的结果看起来将完全相同。

  • 向量操作以其自有单位计费,与基础表读写请求单位分开计量,且两者均随维度数量的增加而扩展。每次写入和搜索时,较小的嵌入维度成本更低,因此请使用能够保持检索质量的最小维度。

  • 在配置的 fields 字段中不含文本的情况下写入的记忆将在没有嵌入的情况下存储,不会出现在语义结果中。发生这种情况时,存储会记录一条警告。

其他资源