# DynamoDB でのエージェンティック AI の使用
Amazon DynamoDB は、サーバーレスのフルマネージド分散 NoSQL データベースで、あらゆる規模で 1 桁ミリ秒のパフォーマンスを実現します。DynamoDB はハイスループットワークロード用に最適化されており、生成 AI モデルと統合することで機能を拡張できます。生成 AI モデルを使用すると、DynamoDB テーブルに保存されているデータをリアルタイムで操作し、コンテキストを認識して高度にパーソナライズされたアプリケーションを構築できます。また、ビジネス、ユーザー、アプリケーションデータを最大限に活用して生成 AI ソリューションをカスタマイズすることで、エンドユーザーエクスペリエンスを向上させることもできます。
生成 AI と、生成 AI アプリケーションを構築するために AWS が提供するソリューションの詳細については、「[生成 AI でビジネスを変革する](https://aws.amazon.com/ai/generative-ai/)」を参照してください。
**Topics**
+ [
## DynamoDB の生成 AI ユースケース
](#gen-ai-use-case-ddb)
+ [
## DynamoDB の生成 AI ブログ
](#gen-ai-blogs)
+ [
# OpenSearch Service と DynamoDB のゼロ ETL 統合を活用する
](ddb-and-amazon-bedrock.md)
+ [
# DynamoDB を LangGraph エージェントのチェックポイントストアとして使用する
](ddb-langgraph-checkpoint.md)
## DynamoDB の生成 AI ユースケース
DynamoDB は、チャットボットや[基盤モデル (FM)](https://aws.amazon.com/what-is/foundation-models/) で構築されたコールセンターなど、AI を活用した会話アプリケーションに広く使用されています。Amazon Bedrock、Amazon SageMaker AI、またはその他のモデルプロバイダーから FM にアクセスできます。このようなアプリケーションは、通常、DynamoDB を使用して、パーソナライゼーションを向上させ、アプリケーションデータ、ビジネスデータ、ユーザーデータの 3 つのデータパターンにわたってユーザーエクスペリエンスを向上させます。これらのデータパターンの例は次のとおりです。
+ [LangChain](https://js.langchain.com/v0.1/docs/integrations/chat_memory/dynamodb/)、 [LlamaIndex](https://docs.llamaindex.ai/en/stable/examples/docstore/DynamoDBDocstoreDemo/)、またはカスタムコードとの統合による、チャットメッセージ履歴などのアプリケーションデータのストレージ。このコンテキストは、モデルがユーザーと対話できるようにすることで、ユーザーエクスペリエンスを向上させます。**
+ インベントリ、料金、ドキュメントなどのビジネスデータを活用して、カスタマイズされたユーザーエクスペリエンスを作成します。
+ ウェブ履歴、過去の注文、ユーザー設定などのユーザーデータを適用して、パーソナライズされた回答を提供します。
例えば、保険会社は DynamoDB を使用して、チャットボットを構築し、[検索拡張検索拡張生成 (RAG)](https://docs.aws.amazon.com/sagemaker/latest/dg/jumpstart-foundation-models-customize-rag.html) ベースの生成 AI モデルにほぼリアルタイムのデータへのアクセスを提供することができます。このようなデータの例としては、住宅ローンの金利、商品の価格設定、準拠/標準契約書、ユーザーのウェブ履歴、ユーザーの好みなどのリアルタイムデータが挙げられます。DynamoDB を RAG と組み合わせることで、保険商品やユーザーデータに関する詳細かつ最新の情報が追加されます。これにより、プロンプトと回答が充実し、エンドユーザーに正確でパーソナライズされたほぼリアルタイムのエクスペリエンスを提供できます。
同様に、金融サービス業界のお客様は、DynamoDB、[Amazon Bedrock ナレッジベース](https://docs.aws.amazon.com/bedrock/latest/userguide/knowledge-base.html)、[Amazon Bedrock エージェント](https://aws.amazon.com/bedrock/agents/)を使用して、RAG ベースの生成 AI アプリケーションを構築しています。これらのアプリケーションは、オープンソースの収益レポートと通話トランスクリプトを使用できます。また、ユーザー固有のポートフォリオとトランザクション履歴を使用して、将来の見通しを含むポートフォリオのオンデマンドの概要を生成することもできます。
## DynamoDB の生成 AI ブログ
以下の記事では、高度な AI を活用したアプリケーションの構築に DynamoDB の機能を活用するのに役立つ詳細なユースケース、ベストプラクティス、ステップバイステップガイドを提供します。
+ [生成 AI チャットボット用の Amazon DynamoDB データモデル](https://aws.amazon.com/blogs/database/amazon-dynamodb-data-models-for-generative-ai-chatbots/)
+ [Amazon DynamoDB、Amazon Bedrock、LangChain を使用して、スケーラブルなコンテキスト対応チャットボットを構築する](https://aws.amazon.com/blogs/database/build-a-scalable-context-aware-chatbot-with-amazon-dynamodb-amazon-bedrock-and-langchain/)
+ [LangGraph と Amazon DynamoDB を使用して耐久性の高い AI エージェントを構築する](https://aws.amazon.com/blogs/database/build-durable-ai-agents-with-langgraph-and-amazon-dynamodb/)
# OpenSearch Service と DynamoDB のゼロ ETL 統合を活用する
Amazon Bedrock と DynamoDB を使用して、Amazon Titan やその他のサードパーティーモデルなどの[基盤モデル (FM)](https://aws.amazon.com/what-is/foundation-models/) へのサーバーレスアクセスを提供できます。Amazon OpenSearch Service とのゼロ ETL 統合を活用して、生成 AI アプリケーションを構築するときにベクトル検索機能を有効にすることができます。[生成 AI による DynamoDB ゼロ ETL から OpenSearch への統合と Amazon Bedrock](https://catalog.workshops.aws/dynamodb-labs/en-US/dynamodb-opensearch-zetl) ワークショップでは、OpenSearch と DynamoDB とのゼロ ETL 統合をセットアップする実践的な経験を提供します。このワークショップでは、次のタスクを実行します。
+ DynamoDB テーブルから OpenSearch にパイプラインを作成します。
+ OpenSearch に Amazon Bedrock コネクタを作成します。
+ OpenSearch をベクトルストアとして使用して Amazon Bedrock にクエリを実行します。
+ Amazon Bedrock の Claude FM を使用して、OpenSearch によって返された検索結果を説明する英語の書面によるレスポンスを作成します。
このワークショップでは、DynamoDB を OpenSearch と統合して生成 AI アプリケーションを構築できます。また、データベースエンジン全体で柔軟なクエリ機能を示し、従来のユースケースで DynamoDB と OpenSearch を統合するのに役立ちます。このワークショップは、[Amazon DynamoDB Immersion Day](https://catalog.workshops.aws/dynamodb-labs/en-US) の 7 つのモジュールの 1 つです。このワークショップは、どの AWS アカウントでも実行できます。
DynamoDB と OpenSearch Service 間のゼロ ETL 統合を設定する方法については、次のブログ記事を参照してください。このブログ記事では、OpenSearch Service でモデルコネクタをセットアップして、受信データに Amazon Bedrock を使用して、埋め込みを自動的に生成する方法についても説明します。[Amazon OpenSearch Service のゼロ ETL を使用した Amazon DynamoDB のベクトル検索](https://aws.amazon.com/blogs/database/vector-search-for-amazon-dynamodb-with-zero-etl-for-amazon-opensearch-service/)
# 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/)