本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。 # 使用 DynamoDB 做為 Strands Agents 的儲存後端 [Strands Agents](https://strandsagents.com/) 是一種開放原始碼 SDK，採用模型驅動的方法，在幾行程式碼中建置 AI 代理器，並支援 Amazon Bedrock 和其他模型供應商。在其建置區塊中是統一的儲存界面：開發套件中每個具狀態子系統都會說出的位元組導向合約 (`write``read`、`delete`、、`list`)。Session Manager 會透過對話快照保留對話快照，記憶體管理員會透過對話快照存放長期的記憶，而內容卸載器和文字記錄會使用相同的操作。 [strands-dynamodb-storage](https://github.com/aws/strands-dynamodb-storage) 套件會在單一 DynamoDB 資料表上實作該合約，適用於 Python 和 TypeScript。SDK 的 `/`分隔儲存金鑰會直接對應至 DynamoDB 金鑰模型：金鑰的前兩個區段會成為分割區金鑰，其餘區段則會成為排序金鑰，因此點操作是單一項目呼叫，而列出字首是原生分割區`Query`，而不是資料表掃描。 ## 主要功能 一個資料表，用於所有客服人員狀態 單一`DynamoDBStorage`執行個體會傳回工作階段持久性、長期記憶體、文字記錄和內容卸載，每個名稱都以自己的金鑰字首分隔。 語意記憶體搜尋 透過資料表上的向量索引，可以使用內嵌編寫記憶體，並透過 `SearchVectors` API 的意義來回收記憶體。請參閱 [具有向量索引的語意長期記憶體](#strands-semantic-memory)。 大型值的 Amazon S3 卸載 卸載為選擇加入：在建構函數上傳遞儲存貯體名稱，超過 400 KB DynamoDB 項目大小限制的值會透明地卸載至 Amazon Simple Storage Service，資料表中會保留一個較小的指標項目。 選用壓縮和存留時間 選用的 gzip 壓縮可讓可壓縮值以較低的成本內嵌。選用的存留時間會為 DynamoDB 原生過期屬性加上戳記，並讀取和列出已過期的篩選條件項目。 多租戶字首 建構函式繫結金鑰字首會在自己的金鑰空間內固定每個操作，因此共享資料表的兩個租用戶會將相同的邏輯金鑰解析為物理上不同的分割區。 ## 先決條件 + AWS 帳戶 具有建立 DynamoDB 資料表許可的 （以及選擇性 Amazon S3 儲存貯體，如果您設定大型值卸載） + Python 3.10 或更新版本搭配 `strands-agents` 1.48.0 或更新版本，或 Node.js 20 或更新版本搭配 `@strands-agents/sdk` 1.10.0 或更新版本 + AWS 設定的登入資料 （請參閱登入資料設定選項 AWS 的文件） + 對於語意記憶體：存取內嵌模型，例如 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) 會顯示宣告索引的完整呼叫。 ## 持久性代理程式工作階段 將儲存體交給代理程式的工作階段管理員。軟體開發套件會命名金鑰、快照每次叫用時的對話，並在相同的工作階段傳回時將其還原： ``` 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 中使用向量索引](VectorSearch.md))。 下列範例透過 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](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/)