# Uso de DynamoDB como almacén de puntos de control para los agentes de LangGraph [LangGraph](https://langchain-ai.github.io/langgraph/) es un marco para desarrollar aplicaciones de IA con estado y múltiples actores utilizando modelos de lenguaje de gran tamaño (LLM). Los agentes de LangGraph requieren un almacenamiento persistente para mantener el estado de la conversación, habilitar flujos de trabajo con intervención humana, garantizar la tolerancia a errores y proporcionar capacidades de depuración con retroceso en el tiempo. La arquitectura sin servidor de DynamoDB, su latencia de menos de diez milisegundos y su escalabilidad automática la convierten en un almacén de puntos de control ideal para implementar LangGraph en producción en AWS. El paquete de `langgraph-checkpoint-aws` proporciona una clase de `DynamoDBSaver` que implementa la interfaz de puntos de control de LangGraph, lo que le permite conservar el estado del agente en DynamoDB con la descarga opcional de Amazon Simple Storage Service para puntos de control grandes. ## Características principales de Persistencia del estado Guarda automáticamente el estado del agente tras cada paso, lo que permite a los agentes reanudar su actividad tras una interrupción y recuperarse de los errores. Limpieza basada en el periodo de vida Elimine automáticamente los puntos de control antiguos utilizando el periodo de vida de DynamoDB para administrar los costos de almacenamiento. Compresión Si lo desea, comprima los datos de los puntos de control con gzip para reducir los costos de almacenamiento y mejorar el rendimiento. Descarga de Amazon S3 Descargue automáticamente los puntos de control de gran tamaño (superiores a 350 KB) a Amazon Simple Storage Service para cumplir con los límites de tamaño de los elementos de DynamoDB. Compatibilidad con sincronización y asincronización API sincrónicas y asincrónicas para ofrecer flexibilidad en diferentes arquitecturas de aplicaciones. ## Requisitos previos + Python 3.10 o posterior + Una Cuenta de AWS con permisos para crear tablas de DynamoDB (y, opcionalmente, buckets de Amazon S3) + Credenciales de AWS configuradas (consulte la documentación de AWS para ver las opciones de configuración de credenciales) **importante** Esta guía crea recursos de AWS que pueden conllevar gastos. DynamoDB utiliza, de forma predeterminada, un modelo de facturación por solicitud, y se aplicarán las tarifas de Amazon S3 si activa la descarga de puntos de control de gran tamaño. Siga las instrucciones de la sección [Limpieza](#langgraph-cleanup) para eliminar los recursos cuando haya terminado. ## Instalación Instale el paquete de puntos de control desde PyPI: ``` pip install langgraph-checkpoint-aws ``` ## Uso básico En el siguiente ejemplo se muestra cómo configurar DynamoDB como almacén de puntos de control para un agente de 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) ``` El `thread_id` en la configuración actúa como clave de partición en DynamoDB, lo que le permite mantener hilos de conversación independientes y recuperar estados históricos de cualquier hilo. ## Configuración de producción En las implementaciones de producción, puede habilitar el Periodo de vida, la compresión y la descarga a Amazon S3. También puede utilizar el parámetro `endpoint_url` para indicar una instancia local de DynamoDB con fines de prueba: ``` 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" ) ``` ## Configuración de la tabla de DynamoDB El protector de puntos de control requiere una tabla de DynamoDB con una clave primaria compuesta. Puede crear la tabla con la siguiente plantilla de 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 ``` Implemente la plantilla con la AWS CLI: ``` aws cloudformation deploy \ --template-file template.yaml \ --stack-name langgraph-checkpoint \ --parameter-overrides TableName=langgraph-checkpoints ``` ## Permisos de IAM necesarios La siguiente política de IAM proporciona los permisos mínimos necesarios para el protector de puntos de control de DynamoDB. Sustituya {{111122223333}} por su ID de Cuenta de AWS y actualice la región para que coincida con su entorno. ``` { "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" } ] } ``` Si habilita la descarga de Amazon S3, añada la siguiente instrucción a la política: ``` { "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" } ``` ## Uso asíncrono Para aplicaciones asíncronas, utilice los métodos asíncronos proporcionados por el protector de puntos de control: ``` 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()) ``` ## Limpieza Para evitar que se sigan cobrando cargos, elimine los recursos que ha creado: ``` # 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 ``` ## Gestión de errores Situaciones de error comunes: + **No se encontró la tabla**: compruebe que el `table_name` y el `region_name` coincide con su tabla de DynamoDB. + **Limitación**: si ve `ProvisionedThroughputExceededException`, considere cambiar al modo de facturación bajo demanda o aumentar la capacidad asignada. + **Tamaño de elemento excedido**: si los puntos de control superan los 350 KB, habilite la descarga de Amazon S3 (consulte [Configuración de producción](#langgraph-production-config)). + **Errores de credenciales**: compruebe que sus credenciales de AWS sean válidas y que disponga de los [permisos necesarios](#langgraph-iam-permissions). ## Recursos adicionales + [langgraph-checkpoint-aws en PyPI](https://pypi.org/project/langgraph-checkpoint-aws/) + [langgraph-checkpoint-aws en GitHub](https://github.com/langchain-ai/langchain-aws/blob/main/libs/langgraph-checkpoint-aws/docs/dynamodb/DynamoDBSaver.md) + [Documentación de LangGraph](https://langchain-ai.github.io/langgraph/) + [Prácticas recomendadas de DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/best-practices.html) + [Desarrollo de agentes de IA duraderos con LangGraph y Amazon DynamoDB](https://aws.amazon.com/blogs/database/build-durable-ai-agents-with-langgraph-and-amazon-dynamodb/)