本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。 # 使用 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/)