# DynamoDB を LangGraph エージェントのチェックポイントストアとして使用する [LangGraph](https://langchain-ai.github.io/langgraph/) は、大規模言語モデル (LLM) を用いて、ステートフルなマルチアクター AI アプリケーションを構築するためのフレームワークです。LangGraph エージェントには、会話状態を維持し、ヒューマンインザループワークフローを有効にし、耐障害性をサポートし、タイムトラベルデバッグ機能を提供するために永続的ストレージが必要です。DynamoDB のサーバーレスアーキテクチャ、1 桁ミリ秒のレイテンシー、自動スケーリングは、AWS での本番稼働用 LangGraph デプロイのチェックポイントストアとして理想的です。 `langgraph-checkpoint-aws` パッケージには、LangGraph チェックポイントインターフェイスを実装する `DynamoDBSaver` クラスを提供し、エージェントの状態を DynamoDB に永続化できます。また、大規模なチェックポイントの場合は、オプションで Amazon Simple Storage Service へのオフロードも可能です。 ## 主な特徴 状態永続性 各ステップの後にエージェントの状態を自動的に保存し、エージェントが中断から再開したり、障害から回復したりできるようにします。 有効期限に基づくクリーンアップ DynamoDB の有効期限を使用して古いチェックポイントを自動的に期限切れにし、ストレージコストを管理します。 圧縮 必要に応じて、gzip でチェックポイントデータを圧縮してストレージコストを削減し、スループットを向上させます。 Amazon S3 へのオフロード 大きなチェックポイント (350 KB 以上) を Amazon Simple Storage Service に自動的にオフロードして、DynamoDB 項目のサイズ制限内で機能させます。 同期と非同期のサポート 同期 API と非同期 API を使用して、さまざまなアプリケーションアーキテクチャに柔軟に対応できます。 ## 前提条件 + Python 3.10 以降 + DynamoDB テーブル (およびオプションで Amazon S3 バケット) を作成するアクセス許可を持つ AWS アカウント。 + AWS 認証情報の設定 (認証情報の設定オプションについては、AWS ドキュメントを参照してください) **重要** このガイドでは、料金が発生する可能性のある AWS リソースを作成します。DynamoDB はデフォルトでリクエストごとの支払料金を使用し、大規模なチェックポイントオフロードを有効にすると Amazon S3 の料金が適用されます。完了したら、「[クリーンアップ](#langgraph-cleanup)」セクションに従ってリソースを削除します。 ## インストール 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 ``` ## エラー処理 一般的なエラーシナリオ。 + **テーブルが見つかりません**: `table_name` と `region_name` が DynamoDB テーブルと一致していることを確認してください。 + **スロットリング**: `ProvisionedThroughputExceededException` が表示された場合は、オンデマンド請求モードに切り替えるか、プロビジョンドキャパシティを増やすことを検討してください。 + **項目サイズ超過**: チェックポイントが 350 KB を超える場合は、Amazon S3 へのオフロードを有効にします (「[本番稼働設定](#langgraph-production-config)」を参照)。 + **認証情報エラー**: AWS 認証情報が有効で、[必要なアクセス許可](#langgraph-iam-permissions)があることを確認します。 ## その他のリソース + [PyPI の langgraph-checkpoint-aws](https://pypi.org/project/langgraph-checkpoint-aws/) + [GitHub の langgraph-checkpoint-aws](https://github.com/langchain-ai/langchain-aws/blob/main/libs/langgraph-checkpoint-aws/docs/dynamodb/DynamoDBSaver.md) + [LangGraph ドキュメント](https://langchain-ai.github.io/langgraph/) + [DynamoDB のベストプラクティス](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/best-practices.html) + [LangGraph と Amazon DynamoDB を使用して耐久性の高い AI エージェントを構築する](https://aws.amazon.com/blogs/database/build-durable-ai-agents-with-langgraph-and-amazon-dynamodb/)