本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。 # 搭配 DynamoDB 使用代理式 AI Amazon DynamoDB 是無伺服器、全受管的分散式 NoSQL 資料庫，具有任何規模的單一位數毫秒效能。DynamoDB 已針對高輸送量工作負載進行最佳化，您可以整合生成式 AI 模型來擴展其功能。使用生成式 AI 模型，您可以即時處理存放在 DynamoDB 資料表中的資料，並建置內容感知及高度個人化的應用程式。您也可以徹底運用業務、使用者和應用程式資料，自訂生成式 AI 解決方案，以強化最終使用者體驗。 如需有關 gen AI 和 解決方案 AWS 提供以建置 gen AI 應用程式的詳細資訊，請參閱[使用生成式 AI 轉換您的業務](https://aws.amazon.com/ai/generative-ai/)。 **Topics** + [ ## DynamoDB 的生成式 AI 使用案例 ](#gen-ai-use-case-ddb) + [ ## DynamoDB 的生成式 AI 部落格 ](#gen-ai-blogs) + [ # 利用 DynamoDB 與 OpenSearch Service 的零 ETL 整合 ](ddb-and-amazon-bedrock.md) + [ # 使用 DynamoDB 做為 LangGraph 代理程式的檢查點存放區 ](ddb-langgraph-checkpoint.md) ## DynamoDB 的生成式 AI 使用案例 DynamoDB 廣泛用於 AI 支援的對話式應用程式中，例如使用[基礎模型 (FM)](https://aws.amazon.com/what-is/foundation-models/) 建置的聊天機器人和客服中心。您可以透過 Amazon Bedrock、Amazon SageMaker AI 或其他模型供應商存取 FM。這類應用程式通常會使用 DynamoDB 來提升個人化效果，並強化應用程式資料、業務資料及使用者資料這三種資料模式的使用者體驗。以下是這些資料模式的範例： + 透過整合 [LangChain](https://js.langchain.com/v0.1/docs/integrations/chat_memory/dynamodb/)、[LlamaIndex](https://docs.llamaindex.ai/en/stable/examples/docstore/DynamoDBDocstoreDemo/) 或自訂程式碼，來儲存應用程式資料，例如聊天訊息歷程記錄。此內容可讓模型和使用者互相*交談*以強化使用者體驗。 + 利用庫存、定價和文件等業務資料，建立自訂使用者體驗。 + 套用使用者資料，例如 Web 歷史記錄、過去的訂單和使用者偏好設定，以提供個人化的答案。 例如，保險公司可以使用 DynamoDB 建置聊天機器人，供[檢索增強生成 (RAG)](https://docs.aws.amazon.com/sagemaker/latest/dg/jumpstart-foundation-models-customize-rag.html) 型 AI 模型近即時存取資料。此類資料的範例包括即時貸款利率、產品定價、合規/標準合約副本、使用者 Web 歷史記錄和使用者偏好設定。結合 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 部落格 下列文章提供詳細的使用案例、最佳實務和逐步指南，協助您利用 DynamoDB 功能來建置進階的 AI 驅動應用程式。 + [適用於生成式 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/) # 利用 DynamoDB 與 OpenSearch Service 的零 ETL 整合 您可以搭配使用 Amazon Bedrock 與 DynamoDB，提供對[基礎模型 (FM) ](https://aws.amazon.com/what-is/foundation-models/)的無伺服器存取，例如 Amazon Titan 和其他第三方模型。您可以運用 Amazon OpenSearch Service 零 ETL 整合，在建置生成式 AI 應用程式時啟用向量搜索功能。[採用零 ETL 整合 DynamoDB、OpenSearch 和 Amazon Bedrock 的生成式 AI](https://catalog.workshops.aws/dynamodb-labs/en-US/dynamodb-opensearch-zetl) 工作坊提供實作體驗，讓您設定 DynamoDB 與 OpenSearch 的零 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) 的七個單元之一。您可以在任何 中執行此研討會 AWS 帳戶。 您也可以參閱下列部落格文章，了解如何在 DynamoDB 和 OpenSearch Service 之間設定零 ETL 整合。此部落格文章也說明如何在 OpenSearch Service 中設定模型連接器，以使用 Amazon Bedrock 自動產生傳入資料的嵌入。[Amazon DynamoDB 與 Amazon OpenSearch Service 的零 ETL 向量搜索](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/) 是使用大型語言模型 (LLMs) 建置具狀態、多動作 AI 應用程式的架構。LangGraph 代理程式需要持久性儲存來維持對話狀態、啟用human-in-the-loop工作流程、支援容錯能力，並提供時間歷程偵錯功能。DynamoDB 的無伺服器架構、單一位數毫秒延遲和自動擴展，使其成為 上生產 LangGraph 部署的理想檢查點存放區 AWS。 `langgraph-checkpoint-aws` 套件提供實作 LangGraph 檢查點介面的`DynamoDBSaver`類別，可讓您在 DynamoDB 中保留代理程式狀態，並為大型檢查點提供選用的 Amazon Simple Storage Service 卸載。 ## 主要功能 狀態持久性 在每個步驟後自動儲存代理程式狀態，讓代理程式從中斷中恢復並從失敗中復原。 即時清理的時間 使用 DynamoDB 存留時間自動使舊檢查點過期，以管理儲存成本。 壓縮 選擇性地使用 gzip 壓縮檢查點資料，以降低儲存成本並改善輸送量。 Amazon S3 卸載 自動將大型檢查點 （大於 350 KB) 卸載至 Amazon Simple Storage Service，以在 DynamoDB 項目大小限制內運作。 同步和非同步支援 同步和非同步 APIs皆可在不同應用程式架構中提供彈性。 ## 先決條件 + Python 3.10 或更新版本 + AWS 帳戶 具有建立 DynamoDB 資料表 （以及選用 Amazon S3 儲存貯體） 許可的 + AWS 已設定的登入資料 （請參閱登入資料設定選項 AWS 的文件） **重要** 本指南會建立可能會產生費用 AWS 的資源。DynamoDB 預設使用pay-per-request，如果您啟用大型檢查點卸載，則會收取 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/)