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 archivio di checkpoint per gli agenti LangGraph [LangGraph](https://langchain-ai.github.io/langgraph/)è un framework per la creazione di applicazioni di intelligenza artificiale multiattore con Large Language Models (). LLMs LangGraph gli agenti richiedono uno storage persistente per mantenere lo stato della conversazione, abilitare i human-in-the-loop flussi di lavoro, supportare la tolleranza agli errori e fornire funzionalità di debug con viaggi nel tempo. L'architettura serverless di DynamoDB, la latenza a una cifra di millisecondi e la scalabilità automatica lo rendono un checkpoint store ideale per le implementazioni di produzione su. LangGraph AWS Il `langgraph-checkpoint-aws` pacchetto fornisce una `DynamoDBSaver` classe che implementa l'interfaccia LangGraph checkpoint, che consente di mantenere lo stato dell'agente in DynamoDB con l'offload opzionale di Amazon Simple Storage Service per checkpoint di grandi dimensioni. ## Funzionalità principali Persistenza dello stato Salva automaticamente lo stato dell'agente dopo ogni passaggio, permettendo agli agenti di riprendere le attività dopo le interruzioni e riprendersi dai guasti. Pulizia basata su Time to Live Fai scadere automaticamente i vecchi checkpoint utilizzando DynamoDB Time to Live per gestire i costi di storage. Compression Opzionalmente, comprimi i dati dei checkpoint con gzip per ridurre i costi di storage e migliorare il throughput. Offload di Amazon S3 Trasferisci automaticamente checkpoint di grandi dimensioni (superiori a 350 KB) su Amazon Simple Storage Service per lavorare entro i limiti di dimensione degli articoli di DynamoDB. Supporto per la sincronizzazione e l'asincrono Sia sincrono che asincrono APIs per la flessibilità in diverse architetture applicative. ## Prerequisiti + Python 3.10 o successivo + E Account AWS con autorizzazioni per creare tabelle DynamoDB (e facoltativamente bucket Amazon S3) + AWS credenziali configurate (consulta la documentazione per le opzioni di configurazione delle credenziali) AWS **Importante** Questa guida crea AWS risorse che potrebbero comportare costi. DynamoDB pay-per-request utilizza la fatturazione per impostazione predefinita e, se abiliti l'offload da checkpoint di grandi dimensioni, vengono applicati i costi di Amazon S3. Segui la sezione [Pulizia per](#langgraph-cleanup) eliminare le risorse quando hai finito. ## Installazione Installa il pacchetto checkpoint da PyPI: ``` pip install langgraph-checkpoint-aws ``` ## Utilizzo di base L'esempio seguente dimostra come configurare DynamoDB come checkpoint store per un agente: LangGraph ``` from langgraph.graph import StateGraph from langgraph_checkpoint_aws import DynamoDBSaver from typing import TypedDict # Define your state schema class State(TypedDict): input: str result: str # Initialize the DynamoDB checkpoint saver checkpointer = DynamoDBSaver( table_name="langgraph-checkpoints", region_name="us-east-1" ) # Build your LangGraph workflow builder = StateGraph(State) builder.add_node("process", lambda state: {"result": "processed"}) builder.set_entry_point("process") builder.set_finish_point("process") # Compile the graph with the DynamoDB checkpointer graph = builder.compile(checkpointer=checkpointer) # Invoke the graph with a thread ID to enable state persistence config = {"configurable": {"thread_id": "session-123"}} result = graph.invoke({"input": "data"}, config) ``` `thread_id`Nella configurazione funge da chiave di partizione in DynamoDB, consentendoti di mantenere thread di conversazione separati e recuperare gli stati storici per ogni thread. ## Configurazione di produzione Per le implementazioni di produzione, puoi abilitare Time to Live, compressione e Amazon S3 offload. Puoi anche utilizzare il `endpoint_url` parametro per puntare a un'istanza locale di DynamoDB per il test: ``` import boto3 from botocore.config import Config from langgraph_checkpoint_aws import DynamoDBSaver # Production configuration session = boto3.Session( profile_name="production", region_name="us-east-1" ) checkpointer = DynamoDBSaver( table_name="langgraph-checkpoints", session=session, ttl_seconds=86400 * 7, # Expire checkpoints after 7 days enable_checkpoint_compression=True, # Enable gzip compression boto_config=Config( retries={"mode": "adaptive", "max_attempts": 6}, max_pool_connections=50 ), s3_offload_config={ "bucket_name": "my-checkpoint-bucket" } ) # Local testing with DynamoDB Local local_checkpointer = DynamoDBSaver( table_name="langgraph-checkpoints", region_name="us-east-1", endpoint_url="http://localhost:8000" ) ``` ## Configurazione della tabella DynamoDB Il checkpoint saver richiede una tabella DynamoDB con una chiave primaria composita. È possibile creare la tabella utilizzando il seguente modello: AWS CloudFormation ``` AWSTemplateFormatVersion: '2010-09-09' Description: 'DynamoDB table for LangGraph checkpoint storage' Parameters: TableName: Type: String Default: langgraph-checkpoints Resources: CheckpointTable: Type: AWS::DynamoDB::Table DeletionPolicy: Retain UpdateReplacePolicy: Retain Properties: TableName: !Ref TableName BillingMode: PAY_PER_REQUEST AttributeDefinitions: - AttributeName: PK AttributeType: S - AttributeName: SK AttributeType: S KeySchema: - AttributeName: PK KeyType: HASH - AttributeName: SK KeyType: RANGE TimeToLiveSpecification: AttributeName: ttl Enabled: true PointInTimeRecoverySpecification: PointInTimeRecoveryEnabled: true SSESpecification: SSEEnabled: true ``` Distribuisci il modello con la AWS CLI: ``` aws cloudformation deploy \ --template-file template.yaml \ --stack-name langgraph-checkpoint \ --parameter-overrides TableName=langgraph-checkpoints ``` ## Autorizzazioni IAM richieste La seguente policy IAM fornisce le autorizzazioni minime richieste per il checkpoint saver di DynamoDB. {{111122223333}}Sostituiscila con il tuo Account AWS ID e aggiorna la regione in base al tuo ambiente. ``` { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "dynamodb:GetItem", "dynamodb:PutItem", "dynamodb:Query", "dynamodb:BatchGetItem", "dynamodb:BatchWriteItem" ], "Resource": "arn:aws:dynamodb:us-east-1:111122223333:table/langgraph-checkpoints" } ] } ``` Se abiliti l'offload di Amazon S3, aggiungi la seguente dichiarazione alla policy: ``` { "Effect": "Allow", "Action": [ "s3:PutObject", "s3:GetObject", "s3:DeleteObject", "s3:PutObjectTagging" ], "Resource": "arn:aws:s3:::my-checkpoint-bucket/*" }, { "Effect": "Allow", "Action": [ "s3:GetBucketLifecycleConfiguration", "s3:PutBucketLifecycleConfiguration" ], "Resource": "arn:aws:s3:::my-checkpoint-bucket" } ``` ## Utilizzo asincrono Per le applicazioni asincrone, usa i metodi asincroni forniti dal checkpoint saver: ``` import asyncio from langgraph.graph import StateGraph from langgraph_checkpoint_aws import DynamoDBSaver from typing import TypedDict class State(TypedDict): input: str result: str async def main(): checkpointer = DynamoDBSaver( table_name="langgraph-checkpoints", region_name="us-east-1" ) builder = StateGraph(State) builder.add_node("process", lambda state: {"result": "processed"}) builder.set_entry_point("process") builder.set_finish_point("process") graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "async-session-123"}} result = await graph.ainvoke({"input": "data"}, config) return result asyncio.run(main()) ``` ## Eliminazione Per evitare addebiti continui, elimina le risorse che hai creato: ``` # Delete the DynamoDB table aws dynamodb delete-table --table-name langgraph-checkpoints # Delete the CloudFormation stack (if you used the template above) aws cloudformation delete-stack --stack-name langgraph-checkpoint # If you created an S3 bucket for large checkpoint offloading, empty and delete it aws s3 rm s3://my-checkpoint-bucket --recursive aws s3 rb s3://my-checkpoint-bucket ``` ## Gestione degli errori Scenari di errore comuni: + **Tabella non trovata**: verifica `table_name` e `region_name` abbina la tabella DynamoDB. + **Limitazione:** se vedi`ProvisionedThroughputExceededException`, valuta la possibilità di passare alla modalità di fatturazione su richiesta o di aumentare la capacità fornita. + **Dimensione dell'articolo superata**: se i checkpoint superano i 350 KB, abilita l'offload di Amazon S3 (vedi). [Configurazione di produzione](#langgraph-production-config) + **Errori nelle credenziali**[: verifica che AWS le tue credenziali siano valide e dispongano delle autorizzazioni richieste.](#langgraph-iam-permissions) ## Risorse aggiuntive + [langgraph-checkpoint-aws su PyPI](https://pypi.org/project/langgraph-checkpoint-aws/) + [langgraph-checkpoint-aws su GitHub](https://github.com/langchain-ai/langchain-aws/blob/main/libs/langgraph-checkpoint-aws/docs/dynamodb/DynamoDBSaver.md) + [Documentazione di LangGraph](https://langchain-ai.github.io/langgraph/) + [Le migliori pratiche di DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/best-practices.html) + [Crea agenti AI durevoli con LangGraph Amazon DynamoDB](https://aws.amazon.com/blogs/database/build-durable-ai-agents-with-langgraph-and-amazon-dynamodb/)