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 zustandsorientierter KI-Anwendungen mit mehreren Akteuren mit Large Language Models (LLMs). LangGraph Agenten benötigen persistenten Speicher, um den Gesprächsstatus aufrechtzuerhalten, menschliche Arbeitsabläufe zu ermöglichen, Fehlertoleranz zu unterstützen und Debugging-Funktionen für Zeitreisen bereitzustellen. Die serverlose Architektur von DynamoDB, die Latenz im einstelligen Millisekundenbereich und die automatische Skalierung machen es 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 und es Ihnen ermöglicht, den Agentenstatus in DynamoDB beizubehalten, wobei das optionale Amazon Simple Storage Service-Offloading für große Checkpoints möglich ist.

Das gleiche Paket enthält auch eine DynamoDBStore Klasse, die das LangGraph Store-Interface für langfristigen, threadübergreifenden Agentenspeicher implementiert, wobei die semantische Suche durch DynamoDB-Vektorindizes unterstützt wird. Weitere Informationen finden Sie unter Semantisches Langzeitgedächtnis für Agenten LangGraph. Dieses Thema behandelt den Checkpoint Saver.

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

Der thread_id in der Konfiguration dient als Partitionsschlüssel in DynamoDB, sodass Sie separate Konversations-Threads verwalten und historische Status für jeden Thread abrufen können.

Konfiguration für die Produktion

Für Produktionsbereitstellungen können Sie Time to Live, Komprimierung und Amazon S3-Offloading aktivieren. Sie können den endpoint_url Parameter auch verwenden, um zum Testen auf eine lokale DynamoDB-Instance 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 mithilfe 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 bietet die Mindestberechtigungen, die für den DynamoDB-Checkpoint Saver erforderlich sind. Ersetzen Sie 111122223333 diese 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-Offloading aktivieren, fügen Sie der Richtlinie die folgende Anweisung 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())

Fehlerbehandlung

Häufige Fehlerszenarien:

Weitere Ressourcen