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 [LangGraph](https://langchain-ai.github.io/langgraph/)est 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 Persistance de l'État Enregistre automatiquement l'état de l'agent après chaque étape, ce qui permet aux agents de reprendre leurs activités après une interruption et de récupérer après une panne. Nettoyage basé sur Time to Live Expirez automatiquement les anciens points de contrôle à l'aide de DynamoDB Time to Live pour gérer les coûts de stockage. Compression Compressez éventuellement les données des points de contrôle avec gzip pour réduire les coûts de stockage et améliorer le débit. Déchargement d'Amazon S3 Déchargez automatiquement les points de contrôle volumineux (supérieurs à 350 Ko) vers Amazon Simple Storage Service afin de respecter les limites de taille des éléments DynamoDB. Support de synchronisation et d'asynchronisation Synchrone et asynchrone APIs pour plus de flexibilité dans les différentes architectures d'applications. ## Conditions préalables + Python 3.10 ou version ultérieure + Et Compte AWS avec les autorisations nécessaires pour créer des tables DynamoDB (et éventuellement des compartiments Amazon S3) + AWS informations d'identification configurées (voir la AWS documentation pour les options de configuration des informations d'identification) **Important** Ce guide crée AWS des ressources susceptibles d'entraîner des frais. DynamoDB pay-per-request utilise la facturation par défaut, et les frais Amazon S3 s'appliquent si vous activez le déchargement à des points de contrôle importants. Suivez la section [Nettoyage](#langgraph-cleanup) pour supprimer des ressources lorsque vous avez terminé. ## 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. {{111122223333}}Remplacez-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 : + **Table introuvable** : vérifiez que la `table_name` table DynamoDB `region_name` correspond à celle-ci. + **Limitation : si c'**est le cas`ProvisionedThroughputExceededException`, envisagez de passer en mode de facturation à la demande ou d'augmenter la capacité allouée. + **Taille de l'article dépassée** : si les points de contrôle dépassent 350 Ko, activez le déchargement sur Amazon S3 (voir[Configuration de production](#langgraph-production-config)). + **Erreurs d'identification** : vérifiez que vos AWS informations d'identification sont valides et que vous disposez des [autorisations requises](#langgraph-iam-permissions). ## Ressources supplémentaires + [langgraph-checkpoint-aws sur PyPI](https://pypi.org/project/langgraph-checkpoint-aws/) + [langgraph-checkpoint-aws sur GitHub](https://github.com/langchain-ai/langchain-aws/blob/main/libs/langgraph-checkpoint-aws/docs/dynamodb/DynamoDBSaver.md) + [Documentation LangGraph](https://langchain-ai.github.io/langgraph/) + [Bonnes pratiques DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/best-practices.html) + [Créez des agents d'IA durables avec LangGraph Amazon DynamoDB](https://aws.amazon.com/blogs/database/build-durable-ai-agents-with-langgraph-and-amazon-dynamodb/)