Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Utilisation de DynamoDB comme magasin de points de contrôle pour les agents LangGraph

LangGraphest un framework permettant de créer des applications d'IA multi-acteurs dynamiques avec de grands modèles linguistiques ()LLMs. LangGraph les agents ont besoin d'un stockage permanent pour maintenir l'état des conversations, activer les human-in-the-loop flux de travail, garantir la tolérance aux pannes et fournir des fonctionnalités de débogage permettant de voyager dans le temps. L'architecture sans serveur de DynamoDB, sa latence à un chiffre en millisecondes et sa mise à l'échelle automatique en font un magasin de points de contrôle idéal pour les déploiements de production sur. LangGraph AWS

Le langgraph-checkpoint-aws package fournit une DynamoDBSaver classe qui implémente l'interface de LangGraph point de contrôle, vous permettant de conserver l'état de l'agent dans DynamoDB avec le déchargement facultatif d'Amazon Simple Storage Service pour les points de contrôle importants.

Fonctions principales

Conditions préalables

Installation

Installez le package checkpoint depuis PyPI :

pip install langgraph-checkpoint-aws

Utilisation de base

L'exemple suivant montre comment configurer DynamoDB en tant que magasin de points de contrôle pour un agent : 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)

La configuration thread_id intégrée fait office de clé de partition dans DynamoDB, ce qui vous permet de gérer des fils de conversation distincts et de récupérer les états historiques de n'importe quel fil.

Configuration de production

Pour les déploiements de production, vous pouvez activer Time to Live, la compression et le déchargement d'Amazon S3. Vous pouvez également utiliser le endpoint_url paramètre pour pointer vers une instance DynamoDB locale à des fins de 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"
)

Configuration de la table DynamoDB

L'économiseur de points de contrôle nécessite une table DynamoDB avec une clé primaire composite. Vous pouvez créer le tableau à l'aide du AWS CloudFormation modèle suivant :

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

Déployez le modèle avec la AWS CLI :

aws cloudformation deploy \
  --template-file template.yaml \
  --stack-name langgraph-checkpoint \
  --parameter-overrides TableName=langgraph-checkpoints

Autorisations IAM requises

La politique IAM suivante fournit les autorisations minimales requises pour l'économiseur de points de contrôle DynamoDB. 111122223333Remplacez-le par votre Compte AWS identifiant et mettez à jour la région en fonction de votre environnement.

{
  "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 vous activez le déchargement sur Amazon S3, ajoutez la déclaration suivante à la politique :

{
  "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"
}

Utilisation asynchrone

Pour les applications asynchrones, utilisez les méthodes asynchrones fournies par l'économiseur de points de contrôle :

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())

Nettoyage

Pour éviter des frais permanents, supprimez les ressources que vous avez créées :

# 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

Gestion des erreurs

Scénarios d'erreur courants :

Ressources supplémentaires