本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

使用 DynamoDB 做為 Strands Agents 的儲存後端

Strands Agents 是一種開放原始碼 SDK，採用模型驅動的方法，在幾行程式碼中建置 AI 代理器，並支援 Amazon Bedrock 和其他模型供應商。在其建置區塊中是統一的儲存界面：開發套件中每個具狀態子系統都會說出的位元組導向合約 (writeread、delete、、list)。Session Manager 會透過對話快照保留對話快照，記憶體管理員會透過對話快照存放長期的記憶，而內容卸載器和文字記錄會使用相同的操作。

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 會顯示宣告索引的完整呼叫。

持久性代理程式工作階段

將儲存體交給代理程式的工作階段管理員。軟體開發套件會命名金鑰、快照每次叫用時的對話，並在相同的工作階段傳回時將其還原：

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/
)

位元組合約也可以直接取得。合約是非同步的：在 代理程式中，開發套件會為您驅動它，而在純指令碼中，您會在 中包裝呼叫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 內嵌文字。模型預設會傳回 1，024 維度向量，因此這些範例的索引會以 1，024 個維度建立。下列程式碼定義其餘範例使用的 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 和您的應用程式層。

結果會先傳回最相似的結果。原始分數的方向遵循索引的距離函數：對於餘弦和歐幾里得距離，較低的 較接近，而對於點產品，較高的 較相似。

將記憶體連接至代理程式

在實際的代理程式中，您希望擷取的記憶體自動到達模型，開發套件的 Memory Manager 會處理：它會在每個模型呼叫之前擷取相關項目，並將其摺疊到模型輸入中，並註冊模型可隨需呼叫search_memory的工具。Memory Manager 接受實作 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 卸載和內嵌模型調用的其他陳述式。

考量事項

其他資源