Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Menggunakan DynamoDB sebagai toko pos pemeriksaan untuk agen LangGraph

LangGraphadalah kerangka kerja untuk membangun aplikasi AI multi-aktor stateful dengan Large Language Models (). LLMs LangGraph agen memerlukan penyimpanan persisten untuk mempertahankan status percakapan, mengaktifkan human-in-the-loop alur kerja, mendukung toleransi kesalahan, dan menyediakan kemampuan debugging perjalanan waktu. Arsitektur tanpa server DynamoDB, latensi milidetik satu digit, dan penskalaan otomatis menjadikannya toko pos pemeriksaan yang ideal untuk penerapan produksi. LangGraph AWS

langgraph-checkpoint-awsPaket ini menyediakan DynamoDBSaver kelas yang mengimplementasikan antarmuka LangGraph pos pemeriksaan, memungkinkan Anda untuk mempertahankan status agen di DynamoDB dengan pembongkaran Amazon Simple Storage Service opsional untuk pos pemeriksaan besar.

Fitur utama

Prasyarat

Penginstalan

Instal paket pos pemeriksaan dari PyPI:

pip install langgraph-checkpoint-aws

Penggunaan dasar

Contoh berikut menunjukkan cara mengkonfigurasi DynamoDB sebagai toko pos pemeriksaan untuk agen: 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_idDalam konfigurasi bertindak sebagai kunci partisi di DynamoDB, memungkinkan Anda untuk mempertahankan utas percakapan terpisah dan mengambil status historis untuk utas apa pun.

Konfigurasi produksi

Untuk penerapan produksi, Anda dapat mengaktifkan Waktu untuk Hidup, kompresi, dan pembongkaran Amazon S3. Anda juga dapat menggunakan endpoint_url parameter untuk menunjuk ke instance DynamoDB lokal untuk pengujian:

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

Konfigurasi tabel DynamoDB

Checkpoint saver membutuhkan tabel DynamoDB dengan kunci primer komposit. Anda dapat membuat tabel menggunakan AWS CloudFormation template berikut:

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

Menyebarkan template dengan AWS CLI:

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

Izin IAM yang diperlukan

Kebijakan IAM berikut memberikan izin minimum yang diperlukan untuk penghemat pos pemeriksaan DynamoDB. Ganti 111122223333 dengan Akun AWS ID Anda dan perbarui Wilayah agar sesuai dengan lingkungan Anda.

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

Jika Anda mengaktifkan pembongkaran Amazon S3, tambahkan pernyataan berikut ke kebijakan:

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

Penggunaan asinkron

Untuk aplikasi asinkron, gunakan metode asinkron yang disediakan oleh penghemat pos pemeriksaan:

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

Bersihkan

Untuk menghindari tagihan yang sedang berlangsung, hapus sumber daya yang Anda buat:

# 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

Penanganan kesalahan

Skenario kesalahan umum:

Sumber daya tambahan