将 DynamoDB 用作 Strands Agents 的存储后端

Strands Agents 是一款开源 SDK，采用模型驱动的方式，仅需几行代码即可构建智能体，并为 Amazon Bedrock 及其它模型提供商提供一流的支持。其构建模块中包含一个统一的存储接口：一个面向字节的合约（write、read、delete、list），此 SDK 中的每个有状态子系统均使用该合约。会话管理器通过该接口保存对话快照，记忆管理器通过该接口存储长期记忆，而上下文卸载器和记录也使用相同的操作。

strands-dynamodb-storage 程序包在单个 DynamoDB 表上实施该合约，同时适用于 Python 和 TypeScript。此 SDK 的以 / 分隔的存储键直接映射到 DynamoDB 键模型：键的前两个段成为分区键，其余部分成为排序键。因此，点操作为单项目调用，而列出某个前缀则是原生分区 Query，从不进行表扫描。

主要功能

先决条件

安装

从 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 展示了声明索引的完整调用示例。

持久化智能体会话

将存储传递给智能体的会话管理器。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 中使用向量索引）。

以下示例通过 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 替换为您的 AWS 账户 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 包含完整的策略，其中包括用于语义搜索、Amazon S3 卸载和嵌入模型调用的附加语句。

注意事项

其他资源