Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Utilizzo di DynamoDB come archivio di checkpoint per gli agenti LangGraph

LangGraphè un framework per la creazione di applicazioni di intelligenza artificiale multiattore con Large Language Models (). LLMs LangGraph gli agenti richiedono uno storage persistente per mantenere lo stato della conversazione, abilitare i human-in-the-loop flussi di lavoro, supportare la tolleranza agli errori e fornire funzionalità di debug con viaggi nel tempo. L'architettura serverless di DynamoDB, la latenza a una cifra di millisecondi e la scalabilità automatica lo rendono un checkpoint store ideale per le implementazioni di produzione su. LangGraph AWS

Il langgraph-checkpoint-aws pacchetto fornisce una DynamoDBSaver classe che implementa l'interfaccia LangGraph checkpoint, che consente di mantenere lo stato dell'agente in DynamoDB con l'offload opzionale di Amazon Simple Storage Service per checkpoint di grandi dimensioni.

Funzionalità principali

Prerequisiti

Installazione

Installa il pacchetto checkpoint da PyPI:

pip install langgraph-checkpoint-aws

Utilizzo di base

L'esempio seguente dimostra come configurare DynamoDB come checkpoint store per un agente: 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_idNella configurazione funge da chiave di partizione in DynamoDB, consentendoti di mantenere thread di conversazione separati e recuperare gli stati storici per ogni thread.

Configurazione di produzione

Per le implementazioni di produzione, puoi abilitare Time to Live, compressione e Amazon S3 offload. Puoi anche utilizzare il endpoint_url parametro per puntare a un'istanza locale di DynamoDB per il 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"
)

Configurazione della tabella DynamoDB

Il checkpoint saver richiede una tabella DynamoDB con una chiave primaria composita. È possibile creare la tabella utilizzando il seguente modello: 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

Distribuisci il modello con la AWS CLI:

aws cloudformation deploy \
  --template-file template.yaml \
  --stack-name langgraph-checkpoint \
  --parameter-overrides TableName=langgraph-checkpoints

Autorizzazioni IAM richieste

La seguente policy IAM fornisce le autorizzazioni minime richieste per il checkpoint saver di DynamoDB. 111122223333Sostituiscila con il tuo Account AWS ID e aggiorna la regione in base al tuo ambiente.

{
  "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"
    }
  ]
}

Se abiliti l'offload di Amazon S3, aggiungi la seguente dichiarazione alla policy:

{
  "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"
}

Utilizzo asincrono

Per le applicazioni asincrone, usa i metodi asincroni forniti dal 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())

Eliminazione

Per evitare addebiti continui, elimina le risorse che hai creato:

# 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

Gestione degli errori

Scenari di errore comuni:

Risorse aggiuntive