Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

DynamoDB als Speicher-Backend für Strands Agents verwenden

Strands Agents ist ein Open-Source-SDK, das einen modellgesteuerten Ansatz zur Erstellung von KI-Agenten in wenigen Codezeilen verfolgt und erstklassige Unterstützung für Amazon Bedrock und andere Modellanbieter bietet. Zu seinen Bausteinen gehört eine einheitliche Speicherschnittstelle: ein byteorientierter Vertrag (write,,,list) readdelete, den jedes statusbehaftete Subsystem im SDK abdeckt. Der Sitzungsmanager speichert Konversations-Snapshots, der Memory Manager speichert Langzeiterinnerungen darin, und der Kontext-Offloader und die Transkripte verwenden dieselben Operationen.

Das Paket strands-dynamodb-storage implementiert diesen Vertrag in einer einzelnen DynamoDB-Tabelle, sowohl für Python als auch. TypeScript Die durch das SDK / getrennten Speicherschlüssel sind direkt dem DynamoDB-Schlüsselmodell zugeordnet: Die ersten beiden Segmente eines Schlüssels werden zum Partitionsschlüssel und der Rest zum Sortierschlüssel. Punktoperationen sind also Einzelelementaufrufen und das Auflisten eines Präfixes ist eine native Partition, niemals ein Tabellenscan. Query

Schlüssel-Features

Voraussetzungen

Installation

Installieren Sie das Paket von PyPI:

pip install strands-dynamodb-storage "strands-agents>=1.48.0"

Oder von npm für TypeScript (das TypeScript Paket ist ein Feature-Parity-Mirror):

npm install strands-dynamodb-storage

Erstellen der Tabelle

Das Paket besitzt keine CreateTable Berechtigungen und erstellt niemals eine Infrastruktur: Sie erstellen die Tabelle im Voraus, wobei Ihre eigenen Tagging-, Backup- und Verschlüsselungseinstellungen angewendet werden. Eine Tabelle mit einem String-Partitionsschlüssel pk und einem String-Sortierschlüssel sk ist die einzige Voraussetzung:

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

Wenn Sie den semantischen Speicher verwenden möchten, deklarieren Sie den Vektorindex mit demselben Befehl: Name, Abmessungen und Distanzfunktion eines Indexes können nach der Erstellung nicht mehr geändert werden. Passen Sie die Dimensionen daher an Ihr Einbettungsmodell an. Das Paket README zeigt den vollständigen Aufruf, der den Index deklariert.

Agentensitzungen beibehalten

Übergeben Sie den Speicher an den Sitzungsmanager Ihres Agenten. Das SDK benennt die Schlüssel, erstellt bei jedem Aufruf einen Snapshot der Konversation und stellt sie wieder her, wenn dieselbe Sitzung zurückkehrt:

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?")

Sie können den Speicher auch einmal selbst einrichten. Agent Jedes Subsystem, das einen Speicher akzeptiert, erbt ihn dann. Jeder Namespace hat sein eigenes Schlüsselpräfix, sodass eine Tabelle den gesamten Status des Agenten enthält:

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

Der Byte-Vertrag ist auch direkt verfügbar. Der Vertrag ist asynchron: In einem Agenten steuert das SDK ihn für Sie, und in einem einfachen Skript packen Sie die Aufrufe ein inasyncio.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())

Semantisches Langzeitgedächtnis mit Vektorindizes

Sitzungspersistenz löst die Hälfte des Speicherproblems: Ihr Agent überlebt einen Neustart und nimmt die Konversation wieder auf. Die schwierigere Hälfte besteht darin, sich an etwas zu erinnern, das ein Benutzer dem Agenten vor Wochen erzählt hat, und zwar in einer neuen Konversation, die keinen Schlüssel wie die alte hat, und bei der Erinnerungen nach ihrer Bedeutung und nicht nach Schlüsseln durchsucht werden müssen. DynamoDB-Vektorindizes führen die Suche nach dem nächsten Nachbarn zu derselben Tabelle, die den Status Ihres Agenten enthält (siehe). Verwenden von Vektorindizes in DynamoDB

In den folgenden Beispielen wird Text mit Amazon Titan Text Embeddings V2 über Amazon Bedrock eingebettet. Das Modell gibt standardmäßig 1.024-dimensionale Vektoren zurück, sodass der Index für diese Beispiele mit 1.024 Dimensionen erstellt wird. Der folgende Code definiert die embed() Funktion, die in den übrigen Beispielen verwendet wird:

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"]

Beim Schreiben eines Speichers werden neben den Bytes auch eine Einbettung und optionale Metadaten hinzugefügt. Der folgende Code erstellt den Speicher mit einem mandantenspezifischen Präfix, sodass der Schlüssel in der physischen Partition memories/m1 landet: 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"},
)

Das Abrufen nach Bedeutung ist ein Aufruf, der sich auf dieselbe Partition bezieht:

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"},
))

Beachten Sie das Argument. pk Der Vektorindex ist genauso partitioniert wie die Tabelle, und jede Suche ist auf eine Partition beschränkt, sodass die Suche eines Mandanten niemals die Speicher eines anderen Mandanten umfasst, und die Arbeit, die jede Suche durchführt, verfolgt die Größe des Speichers dieses Mandanten und nicht die gesamte Tabelle. Denken Sie daran, dass der Partitionswert vom Aufrufer bereitgestellt wird, es handelt sich also um den Umfang der Abfrage und nicht um eine Autorisierungsgrenze: Ein Principal, der dynamodb:SearchVectors an der Tabelle festhält, kann jede Partition durchsuchen, und die Mandantenzugriffskontrolle gehört zu IAM und Ihrer Anwendungsebene.

Die Ergebnisse werden zuerst am ähnlichsten zurückgegeben. Die Richtung des Rohwerts folgt der Distanzfunktion des Indexes: niedriger ist näher für den Kosinus- und den euklidischen Abstand, und höher ist ähnlicher für das Punktprodukt.

Verbinden Sie den Speicher mit einem Agenten

Bei einem echten Agenten möchten Sie, dass abgerufene Erinnerungen das Modell automatisch erreichen, und der Memory Manager des SDK kümmert sich darum: Er ruft relevante Einträge vor jedem Modellaufruf ab und faltet sie in die Modelleingabe ein, und er registriert ein search_memory Tool, das das Modell bei Bedarf aufrufen kann. Der Memory Manager akzeptiert jedes Objekt, das das MemoryStore SDK-Protokoll implementiert. Das Paket liefert keine aus. Sie definieren also eine kleine Klasse in Ihrer eigenen Anwendung, die beim Schreiben und beim Suchen einbettet und Werte zurückgibtMemoryEntry:

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
        ]

Der folgende Code leitet drei Speicher aus und verbindet den Speicher mit dem Agenten, sodass abgerufene Erinnerungen das Modell ohne Orchestrierungscode auf Ihrer Seite erreichen:

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?")

Wenn der Agent dies mit einem Live-Tisch vergleicht, ruft er sein search_memory Tool auf, gleicht die Tokio-Reise und den von DynamoDB bevorzugten Fensterplatz ab und empfiehlt einen Fensterplatz für den Flug. Beim Übergeben wird add_tool_config=True auch ein add_memory Tool registriert, sodass das Modell neue Fakten in derselben Tabelle speichern kann, aus der es abgerufen wird.

Erforderliche IAM-Berechtigungen

Zur Laufzeit führt das Paket vier DynamoDB-Operationen aus, und zwar zusätzlich, SearchVectors wenn Sie die semantische Suche verwenden, und Amazon S3-Operationen nur, wenn Sie Offload konfigurieren, sodass die IAM-Richtlinie mit den geringsten Rechten kurz ist. Ersetzen Sie diese durch Ihre AWS-Konto ID und 111122223333 aktualisieren Sie die Region entsprechend Ihrer Umgebung:

{
  "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"
    }
  ]
}

Das Paket README enthält die vollständige Richtlinie, einschließlich der zusätzlichen Anweisungen für die semantische Suche, den Amazon S3-Offload und den Aufruf des Einbettungsmodells.

Überlegungen

Weitere Ressourcen