Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich. # DynamoDB als Checkpoint-Speicher für Agenten verwenden LangGraph [LangGraph](https://langchain-ai.github.io/langgraph/)ist ein Framework für die Entwicklung zustandsbehafteter KI-Anwendungen mit mehreren Akteuren mit großen Sprachmodellen (). LLMs LangGraph Agenten benötigen persistenten Speicher, um den Konversationsstatus aufrechtzuerhalten, human-in-the-loop Workflows zu ermöglichen, Fehlertoleranz zu unterstützen und Debugging-Funktionen für Zeitreisen bereitzustellen. Die serverlose Architektur, die Latenz im einstelligen Millisekundenbereich und die automatische Skalierung machen DynamoDB zu einem idealen Checkpoint-Speicher für Produktionsbereitstellungen auf. LangGraph AWS Das `langgraph-checkpoint-aws` Paket enthält eine `DynamoDBSaver` Klasse, die die LangGraph Checkpoint-Schnittstelle implementiert, sodass Sie den Agentenstatus in DynamoDB beibehalten können, wobei Sie optional Amazon Simple Storage Service für große Checkpoints auslagern können. ## Schlüssel-Features Beibehaltung des Zustands Speichert automatisch den Agentenstatus nach jedem Schritt, sodass Agenten nach Unterbrechungen weitermachen und sich nach Ausfällen erholen können. Zeit bis zur Live-Bereinigung Lassen Sie alte Checkpoints mithilfe von DynamoDB Time to Live automatisch ablaufen, um die Speicherkosten zu verwalten. Komprimierung Optional können Sie Checkpoint-Daten mit gzip komprimieren, um die Speicherkosten zu senken und den Durchsatz zu verbessern. Amazon S3 S3-Auslagerung Automatisches Auslagern großer Checkpoints (mehr als 350 KB) an Amazon Simple Storage Service, um innerhalb der DynamoDB-Elementgrößenbeschränkungen zu arbeiten. Unterstützung für Synchronisation und Asynchron Sowohl synchron als auch asynchron APIs für Flexibilität in verschiedenen Anwendungsarchitekturen. ## Voraussetzungen + Python 3.10 oder höher + Und AWS-Konto mit Berechtigungen zum Erstellen von DynamoDB-Tabellen (und optional Amazon S3 S3-Buckets) + AWS konfigurierte Anmeldeinformationen (die Optionen zur Einrichtung der Anmeldeinformationen finden Sie in der AWS Dokumentation) **Wichtig** In diesem Handbuch werden AWS Ressourcen zusammengestellt, für die Gebühren anfallen können. DynamoDB verwendet standardmäßig die pay-per-request Fakturierung, und es fallen Amazon S3 S3-Gebühren an, wenn Sie das Offloading an großen Checkpoints aktivieren. Folgen Sie dem Abschnitt [Aufräumen](#langgraph-cleanup), um Ressourcen zu löschen, wenn Sie fertig sind. ## Installation Installieren Sie das Checkpoint-Paket von PyPI: ``` pip install langgraph-checkpoint-aws ``` ## Grundlegende Verwendung Das folgende Beispiel zeigt, wie DynamoDB als Checkpoint-Store für einen Agenten konfiguriert wird: 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) ``` Das `thread_id` in der Konfiguration fungiert als Partitionsschlüssel in DynamoDB, sodass Sie separate Konversations-Threads verwalten und historische Status für jeden Thread abrufen können. ## Produktionskonfiguration Für Produktionsbereitstellungen können Sie Time to Live, Komprimierung und Amazon S3 S3-Offloading aktivieren. Sie können den `endpoint_url` Parameter auch verwenden, um zu Testzwecken auf eine lokale DynamoDB-Instanz zu verweisen: ``` 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" ) ``` ## DynamoDB-Tabellenkonfiguration Der Checkpoint Saver benötigt eine DynamoDB-Tabelle mit einem zusammengesetzten Primärschlüssel. Sie können die Tabelle mit der folgenden Vorlage erstellen: 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 ``` Stellen Sie die Vorlage mit der AWS CLI bereit: ``` aws cloudformation deploy \ --template-file template.yaml \ --stack-name langgraph-checkpoint \ --parameter-overrides TableName=langgraph-checkpoints ``` ## Erforderliche IAM-Berechtigungen Die folgende IAM-Richtlinie stellt die Mindestberechtigungen bereit, die für den DynamoDB-Checkpoint Saver erforderlich sind. Ersetzen Sie es *111122223333* durch Ihre AWS-Konto ID und aktualisieren Sie die Region entsprechend Ihrer Umgebung. ``` { "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" } ] } ``` Wenn Sie Amazon S3 S3-Offloading aktivieren, fügen Sie der Richtlinie die folgende Erklärung hinzu: ``` { "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" } ``` ## Asynchrone Nutzung Verwenden Sie für asynchrone Anwendungen die vom Checkpoint Saver bereitgestellten asynchronen Methoden: ``` 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()) ``` ## Bereinigen Um laufende Gebühren zu vermeiden, löschen Sie die Ressourcen, die Sie erstellt haben: ``` # 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 ``` ## Fehlerbehandlung Häufige Fehlerszenarien: + **Tabelle nicht gefunden**: Überprüfen Sie die DynamoDB-Tabelle `table_name` und `region_name` passen Sie sie an. + **Drosselung**: Falls dies der Fall ist, sollten Sie in Erwägung ziehen`ProvisionedThroughputExceededException`, zum On-Demand-Abrechnungsmodus zu wechseln oder die bereitgestellte Kapazität zu erhöhen. + **Artikelgröße überschritten**: Wenn die Checkpoints 350 KB überschreiten, aktivieren Sie Amazon S3 S3-Offloading (siehe[Produktionskonfiguration](#langgraph-production-config)). + **Anmeldefehler**[: Stellen Sie sicher, dass Ihre AWS Anmeldeinformationen gültig sind und über die erforderlichen Berechtigungen verfügen.](#langgraph-iam-permissions) ## Weitere Ressourcen + [langgraph-checkpoint-aws auf PyPI](https://pypi.org/project/langgraph-checkpoint-aws/) + [langgraph-checkpoint-aws nein GitHub](https://github.com/langchain-ai/langchain-aws/blob/main/libs/langgraph-checkpoint-aws/docs/dynamodb/DynamoDBSaver.md) + [LangGraph-Dokumentation](https://langchain-ai.github.io/langgraph/) + [Bewährte Methoden für DynamoDB](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/best-practices.html) + [Entwickeln Sie langlebige LangGraph KI-Agenten mit Amazon DynamoDB](https://aws.amazon.com/blogs/database/build-durable-ai-agents-with-langgraph-and-amazon-dynamodb/)