# Usar o DynamoDB como um armazenamento de ponto de verificação para agentes do LangGraph O [LangGraph](https://langchain-ai.github.io/langgraph/) é uma estrutura para criar aplicações de IA multiator e de estado com grandes modelos de linguagem (LLMs). Os agentes do LangGraph exigem armazenamento persistente para manter o estado da conversa, viabilizar fluxos de trabalho human-in-the-loop, oferecer suporte à tolerância a falhas e fornecer recursos de depuração de viagem no tempo. A arquitetura de tecnologia sem servidor, a latência de um dígito em milissegundos e o escalonamento automático do DynamoDB o tornam um ponto de verificação ideal para implantações de produção do LangGraph na AWS. O pacote `langgraph-checkpoint-aws` fornece uma classe `DynamoDBSaver` que implementa a interface de ponto de verificação do LangGraph, permitindo que você mantenha o estado do agente no DynamoDB com o descarregamento opcional do Amazon Simple Storage Service para grandes pontos de verificação. ## Atributos principais Persistência de estado Salva automaticamente o estado do agente após cada etapa, permitindo que eles retomem de interrupções e se recuperem das falhas. Limpeza baseada em vida útil Expire automaticamente pontos de verificação antigos usando o tempo de vida útil do DynamoDB para gerenciar os custos de armazenamento. Compactação Opcionalmente, compacte os dados do ponto de verificação com o gzip para reduzir os custos de armazenamento e melhorar o throughput. Descarregamento do Amazon S3 Transfira automaticamente grandes pontos de verificação (maiores do que 350 KB) para o Amazon Simple Storage Service para trabalhar dentro dos limites de tamanho de itens do DynamoDB. Suporte síncrono e assíncrono APIs síncronas e assíncronas para flexibilidade em diferentes arquiteturas de aplicativos. ## Pré-requisitos + Python 3.10 ou posterior + Uma Conta da AWS com permissões para criar tabelas do DynamoDB (e, opcionalmente, buckets do Amazon S3) + Credenciais da AWS configuradas (consulte a documentação da AWS para ver as opções de configuração de credenciais) **Importante** Este guia cria recursos da AWS que podem gerar cobranças. O DynamoDB usa o faturamento por solicitação por padrão, e cobranças do Amazon S3 se aplicam se você habilitar o descarregamento de grandes pontos de verificação. Siga para a seção [Limpar](#langgraph-cleanup) para excluir recursos quando terminar. ## Instalação Instale o pacote de ponto de verificação do PyPI: ``` pip install langgraph-checkpoint-aws ``` ## Uso básico O seguinte exemplo demonstra como configurar o DynamoDB como um ponto de verificação para um agente do 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) ``` O `thread_id` na configuração atua como a chave de partição no DynamoDB, permitindo que você mantenha threads de conversa separados e recupere estados históricos de qualquer thread. ## Configuração de produção Para implantações de produção, você pode ativar o tempo de vida útil, a compressão e o descarregamento do Amazon S3. Você também pode usar o parâmetro `endpoint_url` para apontar para uma instância local do DynamoDB para testes: ``` 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" ) ``` ## Configuração da tabela do DynamoDB O salvador do ponto de verificação requer uma tabela do DynamoDB com uma chave primária composta. Você pode criar a tabela usando o seguinte modelo do 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 ``` Implante o modelo com a CLI da AWS: ``` aws cloudformation deploy \ --template-file template.yaml \ --stack-name langgraph-checkpoint \ --parameter-overrides TableName=langgraph-checkpoints ``` ## Permissões obrigatórias do IAM A política do IAM a seguir fornece as permissões mínimas necessárias para o salvador de ponto de verificação do DynamoDB. Substitua *111122223333* pelo ID da Conta da AWS e atualize a região para que corresponda ao seu 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 você habilitar o descarregamento para o Amazon S3, adicione a seguinte declaração à 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 assíncrono Para aplicações assíncronas, use os métodos assíncronos fornecidos pelo salvador do ponto de verificação: ``` 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()) ``` ## Limpeza Para evitar cobranças contínuas, exclua os recursos que você criou: ``` # 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 ``` ## Tratamento de erros Cenários de erro comuns: + **Tabela não encontrada**: verifique se `table_name` e `region_name` correspondem à sua tabela do DynamoDB. + **controle de utilização**: se você ver `ProvisionedThroughputExceededException`, considere mudar para o modo de cobrança sob demanda ou aumentar a capacidade provisionada. + **Tamanho de item excedido**: se os pontos de verificação excederem 350 KB, habilite o descarregamento do Amazon S3 (consulte [Configuração de produção](#langgraph-production-config)). + **Erros de credencial**: verifique se suas credenciais da AWS são válidas e têm as [permissões necessárias](#langgraph-iam-permissions). ## Recursos adicionais + [langgraph-checkpoint-aws no PyPI](https://pypi.org/project/langgraph-checkpoint-aws/) + [langgraph-checkpoint-aws no GitHub](https://github.com/langchain-ai/langchain-aws/blob/main/libs/langgraph-checkpoint-aws/docs/dynamodb/DynamoDBSaver.md) + [Documentação do LangGraph](https://langchain-ai.github.io/langgraph/) + [Práticas recomendadas do DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/best-practices.html) + [Criar agentes de IA duráveis com o LangGraph e o Amazon DynamoDB](https://aws.amazon.com/blogs/database/build-durable-ai-agents-with-langgraph-and-amazon-dynamodb/)