Usar o DynamoDB como um armazenamento de ponto de verificação para agentes do LangGraph

O 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

Pré-requisitos

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:

Recursos adicionais