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 AI statiche e multiattore con Large Language Models (LLM). LangGraph gli agenti richiedono uno storage persistente per mantenere lo stato della conversazione, abilitare flussi di lavoro interattivi, supportare la tolleranza agli errori e fornire funzionalità di debug che viaggiano nel tempo. L'architettura serverless di DynamoDB, la latenza in millisecondi a una cifra e il ridimensionamento automatico 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, consentendoti di mantenere lo stato dell'agente in DynamoDB con l'offload opzionale di Amazon Simple Storage Service per checkpoint di grandi dimensioni.

Lo stesso pacchetto fornisce anche una DynamoDBStore classe che implementa l'interfaccia LangGraph store per la memoria a lungo termine tra agenti, con ricerca semantica supportata da indici vettoriali DynamoDB. Per ulteriori informazioni, consulta Memoria semantica a lungo termine per agenti LangGraph. Questo argomento tratta del checkpoint saver.

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 archivio di checkpoint 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)

La funzione thread_id in the configuration 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 distribuzioni di produzione, puoi abilitare Time to Live, la compressione e l'offloading di Amazon S3. Puoi anche utilizzare il endpoint_url parametro per puntare a un'istanza DynamoDB locale 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. Sostituiscilo 111122223333 con il tuo Account AWS ID e aggiorna la regione in modo che corrisponda 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, utilizzate 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())

Gestione degli errori

Scenari di errore comuni:

Risorse aggiuntive