# 将 DynamoDB 用作 Strands Agents 的存储后端 [Strands Agents](https://strandsagents.com/) 是一款开源 SDK，采用模型驱动的方式，仅需几行代码即可构建智能体，并为 Amazon Bedrock 及其它模型提供商提供一流的支持。其构建模块中包含一个统一的存储接口：一个面向字节的合约（`write`、`read`、`delete`、`list`），此 SDK 中的每个有状态子系统均使用该合约。会话管理器通过该接口保存对话快照，记忆管理器通过该接口存储长期记忆，而上下文卸载器和记录也使用相同的操作。 [strands-dynamodb-storage](https://github.com/aws/strands-dynamodb-storage) 程序包在单个 DynamoDB 表上实施该合约，同时适用于 Python 和 TypeScript。此 SDK 的以 `/` 分隔的存储键直接映射到 DynamoDB 键模型：键的前两个段成为分区键，其余部分成为排序键。因此，点操作为单项目调用，而列出某个前缀则是原生分区 `Query`，从不进行表扫描。 ## 主要功能 一个表对应于所有智能体状态 单个 `DynamoDBStorage` 实例支持会话持久化、长期记忆、记录和上下文卸载，每项功能均通过各自的键前缀限定命名空间。 语义记忆搜索 在表上创建向量索引后，可以使用嵌入写入记忆，并通过 `SearchVectors` API 按语义含义检索记忆。请参阅[带有向量索引的语义长期记忆](#strands-semantic-memory)。 Amazon S3 卸载大值 卸载是选择加入的：在构造函数中传入存储桶名称后，超过 DynamoDB 项目大小限制（400 KB）的值将透明地卸载至 Amazon Simple Storage Service，表中仅保留一个小型指针项目。 可选压缩和生存时间 可选的 gzip 压缩以较低的成本将可压缩的值保持为内联。可选的生存时间功能会在 DynamoDB 原生过期属性上标记过期时间，读取和列出操作自动筛选掉已过期的项目。 多租户前缀 绑定到构造函数的键前缀将每个操作固定在其自身的键空间内，因此共享一个表的两个租户可将相同的逻辑键解析为物理上截然不同的分区。 ## 先决条件 + 一个 Amazon Web Services 账户，具有创建 DynamoDB 表（以及可选的 Amazon S3 存储桶，如果您配置了大值卸载）的权限 + Python 3.10 或更高版本以及 `strands-agents` 1.48.0 或更高版本，或 Node.js 20 或更高版本以及 `@strands-agents/sdk` 1.10.0 或更高版本 + 配置了 Amazon 凭证（有关凭证设置选项，请参阅 Amazon 文档） + 对于语义记忆：访问嵌入模型（例如 Amazon Bedrock 中的 Amazon Titan Text Embeddings V2），以及使用向量索引创建的表（请参阅[在 DynamoDB 中使用向量索引](VectorSearch.md)） ## 安装 从 PyPI 安装程序包： ``` pip install strands-dynamodb-storage "strands-agents>=1.48.0" ``` 或者从适用于 TypeScript 的 npm 安装（TypeScript 程序包是功能对等的镜像）： ``` npm install strands-dynamodb-storage ``` ## 创建表 该程序包没有 `CreateTable` 权限，从不会创建基础设施：您需要提前创建表，并应用自己的标记、备份和加密设置。唯一的要求是具有字符串分区键 `pk` 和字符串排序键 `sk` 的表： ``` aws dynamodb create-table \ --table-name agent-storage \ --attribute-definitions AttributeName=pk,AttributeType=S AttributeName=sk,AttributeType=S \ --key-schema AttributeName=pk,KeyType=HASH AttributeName=sk,KeyType=RANGE \ --billing-mode PAY_PER_REQUEST ``` 如果您计划使用语义记忆，请在该同一命令中声明向量索引：索引的名称、维度和距离函数在创建后无法更改，因此请根据您的嵌入模型设置合适的维度大小。[程序包 README](https://github.com/aws/strands-dynamodb-storage) 展示了声明索引的完整调用示例。 ## 持久化智能体会话 将存储传递给智能体的会话管理器。SDK 为键指定命名空间，在每次调用时为对话创建快照，并在同一会话返回时将其还原： ``` from strands import Agent from strands.session import SnapshotSessionManager from strands_dynamodb_storage import DynamoDBStorage storage = DynamoDBStorage("agent-storage", region_name="us-east-1") session = SnapshotSessionManager(session_id="user-42", storage=storage) agent = Agent(session_manager=session) agent("Where did we leave off?") ``` 您也可以在 `Agent` 自身上设置存储一次。之后，每个接受存储的子系统都会继承该存储，每个子系统以其自己的键前缀限定命名空间，从而使一个表承载整个智能体的状态： ``` from strands.vended_plugins.context_offloader import ContextOffloader agent = Agent( storage=storage, session_manager=SnapshotSessionManager(), # persists under session/ plugins=[ContextOffloader()], # offloads oversized tool results under offloader/ ) ``` 字节合约也可供直接使用。合约是异步的：在智能体内部，SDK 会为您采用合约；在普通脚本中，您可以将调用封装在 `asyncio.run` 中： ``` import asyncio async def main(): await storage.write("session/user-42/notes", b"prefers aisle seats") keys = await storage.list("session/user-42/") asyncio.run(main()) ``` ## 带有向量索引的语义长期记忆 会话持久化可以解决一半的记忆问题：您的智能体在重启后得以保留并恢复对话。更难的另一半是：在一次与旧对话没有共享任何键的新对话中，检索用户数周前告诉智能体的内容，这需要按语义而非按键来搜索记忆。DynamoDB 向量索引将最近邻搜索引入到保存智能体状态的同一个表中（请参阅[在 DynamoDB 中使用向量索引](VectorSearch.md)）。 以下示例通过 Amazon Bedrock 使用 Amazon Titan Text Embeddings V2 嵌入文本。默认情况下，模型返回 1024 维向量，因此这些示例的索引是使用 1024 个维度创建的。以下代码定义了其余示例使用的 `embed()` 函数： ``` import json import boto3 bedrock = boto3.client("bedrock-runtime", region_name="us-east-1") def embed(text): response = bedrock.invoke_model( modelId="amazon.titan-embed-text-v2:0", body=json.dumps({"inputText": text, "dimensions": 1024}), ) return json.loads(response["body"].read())["embedding"] ``` 写入记忆时，会在字节旁边附加嵌入和可选的元数据。以下代码使用每租户前缀来构造存储，因此键 `memories/m1` 落在物理分区 `user/u1` 中： ``` from strands_dynamodb_storage import DynamoDBStorage, SearchQuery storage = DynamoDBStorage("agent-storage", region_name="us-east-1", prefix="user/u1") await storage.write( "memories/m1", b"prefers window seats on long flights", vector=embed("prefers window seats on long flights"), metadata={"kind": "preference"}, ) ``` 按含义检索是一次调用，范围限于同一分区： ``` results = await storage.search(SearchQuery( vector=embed("what are this user's seating preferences?"), top_k=5, pk="user/u1", # the physical partition: the full key's first two segments filter={"kind": "preference"}, )) ``` 请注意 `pk` 参数。向量索引的分区方式与表相同，且每次搜索都限定在一个分区内，因此某个租户的搜索范围从不会跨越其它租户的记忆，每次搜索所执行的工作都会跟踪该租户的记忆的大小（而非整个表的大小）。请注意，分区值由调用方提供，因此这是查询范围限定，而非授权边界：对表持有 `dynamodb:SearchVectors` 的主体可以搜索任意分区，而租户访问控制应在 IAM 和应用程序层实现。 结果按相似性从高到低返回。原始分数的方向遵循索引的距离函数：对于余弦距离和欧几里得距离，分数越低表示越相近；对于点积，分数越高表示越相似。 ## 将记忆连接至智能体 在实际智能体中，您希望检索到的记忆能够自动到达模型，SDK 的记忆管理器负责处理此过程：它会在每次模型调用之前检索相关条目并将其合并到模型输入中，同时注册一个 `search_memory` 工具，供模型按需调用。记忆管理器接受任何实现 SDK 的 `MemoryStore` 协议的对象。该程序包不附带此功能，因此，您需要在自己的应用程序中定义一个小类，用于在写入时嵌入、在搜索时嵌入，以及返回 `MemoryEntry` 值： ``` import uuid from strands.memory import MemoryEntry from strands_dynamodb_storage import DynamoDBStorage, SearchQuery class DynamoDBMemoryStore: def __init__(self, storage, partition, embed): self.storage = storage self.partition = partition self.embed = embed # the embed() function defined earlier self.name = "dynamodb" self.description = "Long-term memories in DynamoDB, searched by meaning" self.max_search_results = 3 self.writable = True self.extraction = None async def add(self, content, metadata=None): await self.storage.write( f"memories/{uuid.uuid4().hex[:8]}", content.encode(), vector=self.embed(content), metadata=metadata, ) async def search(self, query, options=None): results = await self.storage.search(SearchQuery( vector=self.embed(query), top_k=self.max_search_results, pk=self.partition, include_values=True, )) return [ MemoryEntry(content=r.data.decode(), metadata=r.metadata) for r in results if r.data is not None ] ``` 以下代码为三个记忆做种并将存储连接到智能体，从而使检索到的记忆无需您编写任何编排代码即可访问模型： ``` import asyncio from strands import Agent from strands.memory import MemoryManager storage = DynamoDBStorage("agent-storage", region_name="us-east-1", prefix="user/u1") store = DynamoDBMemoryStore(storage, partition="user/u1", embed=embed) async def seed(): await store.add("Prefers window seats on long flights") await store.add("Planning a trip to Tokyo in December") await store.add("Allergic to peanuts") asyncio.run(seed()) memory = MemoryManager(stores=[store], add_tool_config=True) agent = Agent(memory_manager=memory) agent("Book me a flight seat for my December trip. Which seat should I pick?") ``` 针对实时表运行时，智能体会调用其 `search_memory` 工具，匹配 DynamoDB 中的东京行程和靠窗座位偏好，并推荐该航班的靠窗座位。传入 `add_tool_config=True` 还会注册一个 `add_memory` 工具，因此模型可以通过其用于检索的同一个表来存储新的事实。 ## 所需的 IAM 权限 在运行时，该程序包发出四个 DynamoDB 操作，使用语义搜索时额外发出 `SearchVectors`，且仅当配置卸载时才发出 Amazon S3 操作，因此最低权限 IAM 策略很简短。将 {{111122223333}} 替换为您的 Amazon Web Services 账户 ID，并更新区域以匹配您的环境： ``` { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "dynamodb:PutItem", "dynamodb:GetItem", "dynamodb:DeleteItem", "dynamodb:Query" ], "Resource": "arn:aws:dynamodb:us-east-1:111122223333:table/agent-storage" } ] } ``` [程序包 README](https://github.com/aws/strands-dynamodb-storage) 包含完整的策略，其中包括用于语义搜索、Amazon S3 卸载和嵌入模型调用的附加语句。 ## 注意事项 + 向量索引的名称、维度和距离函数在创建后不可变，且新创建的索引在可供搜索之前会先回填。一个表最多支持五个向量索引，因此，如以后需采用不同的配置，则意味着只需添加索引，而无需重建表。 + 向量索引是最终一致的，与全局二级索引的模型相同。刚写入的记忆可能需要短暂的时间才可供搜索。 + 生存时间过期筛选适用于读取和列出操作。由于生存时间删除是异步的，因此搜索可能会短暂地返回已过期但 DynamoDB 尚未实际移除的项目。 + 列出操作需要一个至少涵盖完整范围和标识符的前缀。诸如空前缀等宽泛的列出操作将遭到拒绝；该程序包从不会回退到全表扫描。 + 如果您对卸载的值启用生存时间，请添加 Amazon S3 生命周期规则：DynamoDB 移除过期的指针项目，而生命周期规则是回收 Amazon S3 对象的内容。 ## 其他资源 + [GitHub 上的 strands-dynamodb-storage](https://github.com/aws/strands-dynamodb-storage) + [PyPI 上的 strands-dynamodb-storage](https://pypi.org/project/strands-dynamodb-storage/) + [npm 上的 strands-dynamodb-storage](https://www.npmjs.com/package/strands-dynamodb-storage) + [Strands Agents 文档](https://strandsagents.com/)