Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Utilisation de DynamoDB comme backend de stockage pour Strands Agents

Strands Agents est un SDK open source qui adopte une approche basée sur des modèles pour créer des agents d'IA en quelques lignes de code, avec un support de premier ordre pour Amazon Bedrock et d'autres fournisseurs de modèles. Parmi ses éléments constitutifs figure une interface de stockage unifiée : un contrat orienté octet (write, readdelete,list) défini par chaque sous-système avec état du SDK. Le gestionnaire de session conserve les instantanés des conversations par son intermédiaire, le gestionnaire de mémoire y stocke les mémoires à long terme, et le déchargeur de contexte et les transcriptions utilisent les mêmes opérations.

Le package strands-dynamodb-storage implémente ce contrat sur une seule table DynamoDB, à la fois pour Python et. TypeScript Les clés de stockage / séparées par le SDK correspondent directement au modèle de clé DynamoDB : les deux premiers segments d'une clé deviennent la clé de partition et le reste devient la clé de tri. Les opérations ponctuelles sont donc des appels à élément unique et la liste d'un préfixe est une partition Query native, jamais une analyse de table.

Fonctionnalités principales

Conditions préalables

Installation

Installez le package depuis PyPI :

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

Ou depuis npm for TypeScript (le TypeScript package est un miroir de parité de fonctionnalités) :

npm install strands-dynamodb-storage

Création de la table

Le package ne détient aucune CreateTable autorisation et ne crée jamais d'infrastructure : vous créez la table à l'avance, en appliquant vos propres paramètres de balisage, de sauvegarde et de cryptage. Une table avec une clé de partition de chaîne pk et une clé de tri de chaîne sk est la seule exigence :

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

Si vous envisagez d'utiliser la mémoire sémantique, déclarez l'index vectoriel dans cette même commande : le nom, les dimensions et la fonction de distance d'un index ne peuvent pas être modifiés après sa création. Dimensionnez donc les dimensions en fonction de votre modèle d'intégration. Le package README affiche l'appel complet qui déclare l'index.

Persister les sessions des agents

Confiez l'espace de stockage au gestionnaire de sessions de votre agent. Le SDK espace les clés, crée un instantané de la conversation à chaque appel et la restaure au retour de la même session :

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

Vous pouvez également définir le stockage une fois sur Agent lui-même. Chaque sous-système qui accepte un stockage en hérite ensuite, chacun étant placé sous son propre préfixe de clé, de sorte qu'une table contient l'état complet de l'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/
)

Le contrat d'octets est également disponible directement. Le contrat est asynchrone : à l'intérieur d'un agent, le SDK le gère pour vous, et dans un script simple, vous encapsulez les appels dans : 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())

Mémoire sémantique à long terme avec index vectoriels

La persistance des sessions résout la moitié du problème de mémoire : votre agent survit à un redémarrage et reprend la conversation. La partie la plus difficile consiste à se remémorer quelque chose qu'un utilisateur a dit à l'agent il y a quelques semaines, lors d'une nouvelle conversation qui ne partage aucune clé avec l'ancienne, et qui nécessite de rechercher dans les souvenirs par sens plutôt que par clé. Les index vectoriels DynamoDB redirigent la recherche par le voisin le plus proche vers la même table qui contient l'état de votre agent (voir). Utilisation d'index vectoriels dans DynamoDB

Les exemples suivants intègrent du texte avec Amazon Titan Text Embeddings V2 via Amazon Bedrock. Le modèle renvoie des vecteurs de 1 024 dimensions par défaut. L'index de ces exemples est donc créé avec 1 024 dimensions. Le code suivant définit la embed() fonction utilisée dans les autres exemples :

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

L'écriture d'une mémoire associe une intégration et des métadonnées facultatives aux octets. Le code suivant construit le magasin avec un préfixe par locataire, de sorte que la clé memories/m1 atterrisse dans la partition physique : 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"},
)

Le rappel signifie un appel, limité à la même partition :

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

Remarquez l'pkargument. L'index vectoriel est partitionné de la même manière que la table, et chaque recherche est limitée à une partition, de sorte que la recherche d'un locataire ne s'étend jamais sur les mémoires d'un autre locataire, et le travail effectué par chaque recherche suit la taille de la mémoire de ce locataire plutôt que la taille de la table entière. N'oubliez pas que la valeur de la partition est fournie par l'appelant. Il s'agit donc d'une portée de requête plutôt que d'une limite d'autorisation : un principal figurant dynamodb:SearchVectors sur la table peut rechercher n'importe quelle partition, et le contrôle d'accès des locataires appartient à IAM et à votre couche d'application.

Les résultats renvoient les résultats les plus similaires en premier. La direction du score brut suit la fonction de distance de l'indice : la valeur la plus faible est la plus proche pour le cosinus et la distance euclidienne, et la valeur la plus élevée est plus similaire pour le produit scalaire.

Transférez la mémoire à un agent

Dans un véritable agent, vous souhaitez que les mémoires récupérées atteignent automatiquement le modèle, et le gestionnaire de mémoire du SDK s'en charge : il récupère les entrées pertinentes avant chaque appel de modèle et les intègre dans l'entrée du modèle, et il enregistre un search_memory outil que le modèle peut appeler à la demande. Le Memory Manager accepte tout objet implémentant le MemoryStore protocole du SDK. Le package n'en fournit pas. Vous définissez donc une petite classe dans votre propre application qui s'intègre lors de l'écriture, s'intègre à la recherche et renvoie MemoryEntry des valeurs :

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
        ]

Le code suivant génère trois mémoires et connecte le magasin à l'agent, de sorte que les mémoires récupérées atteignent le modèle sans code d'orchestration de votre part :

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

En le comparant à une table en direct, l'agent appelle son search_memory outil, fait correspondre le trajet à Tokyo avec la préférence de siège côté fenêtre de DynamoDB, et recommande un siège côté fenêtre pour le vol. Le passage permet add_tool_config=True également d'enregistrer un add_memory outil, de sorte que le modèle peut stocker de nouveaux faits dans le même tableau que celui dont il se souvient.

Autorisations IAM requises

Au moment de l'exécution, le package émet quatre opérations DynamoDB, auxquelles s'ajoutent les opérations de recherche sémantique SearchVectors lorsque vous utilisez la recherche sémantique, et les opérations Amazon S3 uniquement lorsque vous configurez le déchargement, de sorte que la politique IAM du moindre privilège est courte. Remplacez-le 111122223333 par votre Compte AWS identifiant et mettez à jour la région en fonction de votre environnement :

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

Le package README contient la politique complète, y compris les instructions supplémentaires pour la recherche sémantique, le déchargement d'Amazon S3 et l'invocation du modèle d'intégration.

Considérations

Ressources supplémentaires