本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

使用 DynamoDB 做為 LangGraph 代理程式的檢查點存放區

LangGraph 是使用大型語言模型 (LLMs) 建置具狀態、多動作 AI 應用程式的架構。LangGraph 代理程式需要持久性儲存來維持對話狀態、啟用human-in-the-loop工作流程、支援容錯能力，並提供時間歷程偵錯功能。DynamoDB 的無伺服器架構、單一位數毫秒延遲和自動擴展，使其成為 上生產 LangGraph 部署的理想檢查點存放區 AWS。

langgraph-checkpoint-aws 套件提供實作 LangGraph 檢查點介面的DynamoDBSaver類別，可讓您在 DynamoDB 中保留代理程式狀態，並為大型檢查點提供選用的 Amazon Simple Storage Service 卸載。

相同的套件也提供實作 LangGraph 存放區界面的DynamoDBStore類別，用於長期、跨執行緒代理程式記憶體，並由 DynamoDB 向量索引支援語意搜尋。如需詳細資訊，請參閱LangGraph 代理程式的語意長期記憶體。本主題涵蓋檢查點節省程式。

主要功能

先決條件

安裝

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

錯誤處理

常見錯誤案例：

其他資源