Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Utilizzo di DynamoDB come backend di archiviazione per Strands Agents

Strands Agents è un SDK open source che adotta un approccio basato su modelli per creare agenti di intelligenza artificiale in poche righe di codice, con un supporto di prima classe per Amazon Bedrock e altri fornitori di modelli. Tra i suoi elementi costitutivi c'è un'interfaccia di storage unificata: un contratto orientato ai byte (write,, readdelete,list) di cui parla ogni sottosistema stateful dell'SDK. Il Session Manager conserva le istantanee delle conversazioni attraverso di esso, il Memory Manager memorizza memorie a lungo termine e l'offloader di contesto e le trascrizioni utilizzano le stesse operazioni.

Il pacchetto strands-dynamodb-storage implementa quel contratto su una singola tabella DynamoDB, sia per Python che. TypeScript Le chiavi di archiviazione / separate dall'SDK vengono mappate direttamente sul modello di chiavi DynamoDB: i primi due segmenti di una chiave diventano la chiave di partizione e il resto diventa la chiave di ordinamento, quindi le operazioni sui punti sono chiamate a elemento singolo e l'elenco di un prefisso è una partizione nativa, mai una scansione della tabella. Query

Funzionalità principali

Prerequisiti

Installazione

Installa il pacchetto da PyPI:

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

O da npm for TypeScript (il TypeScript pacchetto è un mirror con parità di funzionalità):

npm install strands-dynamodb-storage

Creazione della tabella

Il pacchetto non dispone di CreateTable autorizzazioni e non crea mai un'infrastruttura: si crea la tabella in anticipo, applicando le proprie impostazioni di tagging, backup e crittografia. L'unico requisito è una tabella con una chiave di partizione per pk le stringhe e una chiave sk di ordinamento delle stringhe:

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

Se prevedi di utilizzare la memoria semantica, dichiara l'indice vettoriale con questo stesso comando: il nome, le dimensioni e la funzione di distanza di un indice non possono essere modificati dopo la creazione, quindi ridimensiona le dimensioni in base al modello di incorporamento. Il pacchetto README mostra la chiamata completa che dichiara l'indice.

Persiste le sessioni degli agenti

Consegna lo spazio di archiviazione al gestore delle sessioni del tuo agente. L'SDK assegna i namespace alle chiavi, crea un'istantanea della conversazione a ogni chiamata e la ripristina quando ritorna la stessa sessione:

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

Puoi anche impostare l'archiviazione una volta su se stesso. Agent Ogni sottosistema che accetta uno storage lo eredita, ognuno con un namespace con il proprio prefisso di chiave, quindi una tabella riporta lo stato dell'intero agente:

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

Il contratto di byte è inoltre disponibile direttamente. Il contratto è asincrono: all'interno di un agente l'SDK lo gestisce al posto tuo e in un semplice script racchiudi le chiamate in: 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())

Memoria semantica a lungo termine con indici vettoriali

La persistenza della sessione risolve metà del problema della memoria: il tuo agente sopravvive a un riavvio e riprende la conversazione. La parte più difficile è ricordare qualcosa che un utente ha detto all'agente settimane fa, in una nuova conversazione che non condivide alcuna chiave con quella precedente e che richiede la ricerca nei ricordi per significato anziché per chiave. Gli indici vettoriali DynamoDB portano la ricerca del vicino più prossimo nella stessa tabella che contiene lo stato dell'agente (vedi). Utilizzo degli indici vettoriali in DynamoDB

Gli esempi seguenti incorporano testo con Amazon Titan Text Embeddings V2 tramite Amazon Bedrock. Il modello restituisce vettori di 1.024 dimensioni per impostazione predefinita, quindi l'indice per questi esempi viene creato con 1.024 dimensioni. Il codice seguente definisce la embed() funzione utilizzata dagli esempi rimanenti:

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

La scrittura di una memoria associa ai byte un incorporamento e dei metadati opzionali. Il codice seguente costruisce l'archivio con un prefisso per tenant, in modo che la chiave arrivi nella partizione fisica: 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"},
)

Il richiamo per significato è una chiamata, con ambito alla stessa partizione:

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

Notate l'argomento. pk L'indice vettoriale è partizionato allo stesso modo della tabella e ogni ricerca è limitata a una partizione, quindi la ricerca di un tenant non si estende mai sulle memorie di un altro tenant e il lavoro svolto da ciascuna ricerca tiene traccia delle dimensioni della memoria di quel tenant anziché dell'intera tabella. Tieni presente che il valore della partizione viene fornito dal chiamante, quindi si tratta dell'ambito della query anziché di un limite di autorizzazione: un elemento principale presente nella tabella può cercare qualsiasi partizione e il controllo dell'accesso del tenant appartiene a IAM e al livello dell'applicazione. dynamodb:SearchVectors

I risultati restituiscono per primi i risultati più simili. La direzione del punteggio grezzo segue la funzione di distanza dell'indice: più bassa è più vicina per il coseno e la distanza euclidea e più alta è più simile per il prodotto scalare.

Collega la memoria a un agente

In un agente reale si desidera che le memorie recuperate raggiungano automaticamente il modello e il Memory Manager dell'SDK se ne occupa: recupera le voci pertinenti prima di ogni chiamata del modello e le inserisce nell'input del modello e registra uno search_memory strumento che il modello può richiamare su richiesta. Il Memory Manager accetta qualsiasi oggetto che implementa il protocollo dell'SDK. MemoryStore Il pacchetto non ne fornisce uno, quindi definisci una piccola classe nella tua applicazione che incorpora in scrittura, incorpora in ricerca e restituisce valori: 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
        ]

Il codice seguente genera tre memorie e collega l'archivio all'agente, in modo che le memorie recuperate raggiungano il modello senza alcun codice di orchestrazione da parte dell'utente:

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

Confrontandolo con un tavolo in tempo reale, l'agente chiama il relativo search_memory strumento, confronta le preferenze relative alla tratta a Tokyo e al posto vicino al finestrino di DynamoDB e consiglia un posto vicino al finestrino per il volo. Il passaggio consente add_tool_config=True inoltre di registrare add_memory uno strumento, in modo che il modello possa memorizzare nuovi dati nella stessa tabella da cui richiama.

Autorizzazioni IAM richieste

In fase di esecuzione, il pacchetto esegue quattro operazioni DynamoDB, più SearchVectors quando si utilizza la ricerca semantica e le operazioni di Amazon S3 solo quando si configura l'offload, quindi la politica IAM con privilegi minimi è breve. Sostituiscila 111122223333 con il tuo Account AWS ID e aggiorna la regione in modo che corrisponda al tuo ambiente:

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

Il pacchetto README contiene la policy completa, incluse le istruzioni aggiuntive per la ricerca semantica, l'offload di Amazon S3 e l'invocazione del modello di incorporamento.

Considerazioni

Risorse aggiuntive