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 avec état à l'aide de grands modèles de langage (LLM). LangGraph les agents ont besoin d'un stockage permanent pour maintenir l'état des conversations, activer des flux de travail humains, garantir la tolérance aux pannes et fournir des fonctionnalités de débogage en fonction du temps. L'architecture sans serveur de DynamoDB, sa latence en millisecondes à un chiffre et sa mise à l'échelle automatique en font un magasin de points de contrôle idéal pour les déploiements en production. LangGraph AWS

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

Le même package fournit également une DynamoDBStore classe qui implémente l'interface de LangGraph stockage pour la mémoire d'agent interthread à long terme, avec une recherche sémantique soutenue par des index vectoriels DynamoDB. Pour de plus amples informations, veuillez consulter Mémoire sémantique à long terme pour les agents LangGraph. Cette rubrique traite de l'économiseur de points de contrôle.

Fonctionnalités 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)

thread_idDans la configuration, elle 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 l'état historique de chaque fil.

Configuration de production

Pour les déploiements en production, vous pouvez activer Time to Live, la compression et le déchargement 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 des tables 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 à l'aide de 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. Remplacez-le 111122223333 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 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 le Checkpoint Saver :

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

Gestion des erreurs

Scénarios d'erreur courants :

Ressources supplémentaires