DynamoDB を LangGraph エージェントのチェックポイントストアとして使用する

LangGraph は、大規模言語モデル (LLM) を用いて、ステートフルなマルチアクター AI アプリケーションを構築するためのフレームワークです。LangGraph エージェントには、会話状態を維持し、ヒューマンインザループワークフローを有効にし、耐障害性をサポートし、タイムトラベルデバッグ機能を提供するために永続的ストレージが必要です。DynamoDB のサーバーレスアーキテクチャ、1 桁ミリ秒のレイテンシー、自動スケーリングは、AWS での本番稼働用 LangGraph デプロイのチェックポイントストアとして理想的です。

langgraph-checkpoint-aws パッケージには、LangGraph チェックポイントインターフェイスを実装する DynamoDBSaver クラスを提供し、エージェントの状態を DynamoDB に永続化できます。また、大規模なチェックポイントの場合は、オプションで Amazon Simple Storage Service へのオフロードも可能です。

主な特徴

前提条件

インストール

PyPI からチェックポイントパッケージをインストールします。

pip install langgraph-checkpoint-aws

基本的な使用法

次の例は、DynamoDB を 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_id は DynamoDB のパーティションキーとして機能し、個別の会話スレッドを維持し、任意のスレッドの履歴状態を取得できます。

本番稼働設定

本番デプロイでは、有効期限、圧縮、Amazon S3 オフロードを有効にできます。endpoint_url パラメータを使用して、テスト用のローカル DynamoDB インスタンスを指定することもできます。

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 テーブル設定

チェックポイントセーバーには、複合プライマリキーを持つ DynamoDB テーブルが必要です。次の 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

AWS CLI を使用してテンプレートをデプロイします。

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

必要な IAM 許可

次の IAM ポリシーは、DynamoDB チェックポイントセーバーに必要な最小限のアクセス許可を提供します。111122223333 を AWS アカウント ID に置き換え、環境に合わせてリージョンを更新します。

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

Amazon S3 へのオフロードを有効にする場合は、次のステートメントをポリシーに追加します。

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

非同期使用

非同期アプリケーションの場合は、チェックポイントセーバーが提供する非同期メソッドを使用します。

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

クリーンアップ

作成したリソースは、継続的な料金の発生を回避するために削除してください。

# 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

エラー処理

一般的なエラーシナリオ。

その他のリソース