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

Menggunakan DynamoDB sebagai penyimpanan pos pemeriksaan untuk agen LangGraph

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

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

Paket yang sama juga menyediakan DynamoDBStore kelas yang mengimplementasikan antarmuka LangGraph penyimpanan untuk memori agen lintas-thread jangka panjang, dengan pencarian semantik yang didukung oleh indeks vektor DynamoDB. Untuk informasi selengkapnya, lihat Memori jangka panjang semantik untuk LangGraph agen. Topik ini mencakup penghemat pos pemeriksaan.

Fitur utama

Prasyarat

Penginstalan

Instal paket pos pemeriksaan dari PyPI:

pip install langgraph-checkpoint-aws

Penggunaan dasar

Contoh berikut menunjukkan cara mengkonfigurasi DynamoDB sebagai penyimpanan 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)

Konfigurasi thread_id dalam bertindak sebagai kunci partisi di DynamoDB, memungkinkan Anda untuk mempertahankan utas percakapan terpisah dan mengambil status historis untuk thread apa pun.

Konfigurasi produksi

Untuk penerapan produksi, Anda dapat mengaktifkan Time to Live, kompresi, dan offloading 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

Penghemat pos pemeriksaan memerlukan 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

Terapkan 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 menyediakan 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 offloading 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())

Penanganan kesalahan

Skenario kesalahan umum:

Sumber daya tambahan