Uso de DynamoDB como almacén de puntos de control para los agentes de LangGraph

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

Requisitos previos

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:

Recursos adicionales