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

LangGraphist 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

Voraussetzungen

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:

Weitere Ressourcen