DynamoDB를 LangGraph 에이전트의 체크포인트 스토어로 사용

LangGraph는 대규모 언어 모델(LLM)을 사용하여 상태를 유지하고 여러 참여자가 있는 AI 애플리케이션을 구축하기 위한 프레임워크입니다. LangGraph 에이전트는 대화 상태를 유지하고, 인간 참여형 워크플로를 활성화하고, 내결함성을 지원하고, 시간 이동 디버깅 기능을 제공하기 위해 영구 스토리지가 필요합니다. DynamoDB의 서버리스 아키텍처, 한 자릿수 밀리초 지연 시간 및 오토 스케일링은 AWS의 프로덕션 LangGraph 배포에 이상적인 체크포인트 스토어입니다.

langgraph-checkpoint-aws 패키지는 LangGraph 체크포인트 인터페이스를 구현하는 DynamoDBSaver 클래스를 제공하므로 대규모 체크포인트에 대한 선택적 Amazon Simple Storage Service 오프로드로 DynamoDB에서 에이전트 상태를 유지할 수 있습니다.

주요 기능

사전 조건

설치

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의 파티션 키 역할을 하므로 별도의 대화 스레드를 유지하고 모든 스레드의 기록 상태를 검색할 수 있습니다.

프로덕션 구성

프로덕션 배포의 경우 Time to Live, 압축 및 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

오류 처리

일반적인 오류 시나리오:

추가 리소스