# DynamoDB를 LangGraph 에이전트의 체크포인트 스토어로 사용 [LangGraph](https://langchain-ai.github.io/langgraph/)는 대규모 언어 모델(LLM)을 사용하여 상태를 유지하고 여러 참여자가 있는 AI 애플리케이션을 구축하기 위한 프레임워크입니다. LangGraph 에이전트는 대화 상태를 유지하고, 인간 참여형 워크플로를 활성화하고, 내결함성을 지원하고, 시간 이동 디버깅 기능을 제공하기 위해 영구 스토리지가 필요합니다. DynamoDB의 서버리스 아키텍처, 한 자릿수 밀리초 지연 시간 및 오토 스케일링은 AWS의 프로덕션 LangGraph 배포에 이상적인 체크포인트 스토어입니다. `langgraph-checkpoint-aws` 패키지는 LangGraph 체크포인트 인터페이스를 구현하는 `DynamoDBSaver` 클래스를 제공하므로 대규모 체크포인트에 대한 선택적 Amazon Simple Storage Service 오프로드로 DynamoDB에서 에이전트 상태를 유지할 수 있습니다. ## 주요 기능 상태 지속성 각 단계 후 에이전트 상태를 자동으로 저장하여 에이전트가 중단에서 재개하고 실패에서 복구할 수 있도록 합니다. 라이브 기반 정리까지의 시간 스토리지 비용을 관리하기 위해 DynamoDB Time to Live를 사용하여 이전 체크포인트를 자동으로 만료시킵니다. 압축 선택적으로 gzip으로 체크포인트 데이터를 압축하여 스토리지 비용을 절감하고 처리량을 개선할 수 있습니다. Amazon S3 오프로드 DynamoDB 항목 크기 제한 내에서 작동하도록 대용량 체크포인트(350KB 초과)를 Amazon Simple Storage Service로 자동으로 오프로드합니다. 동기화 및 비동기 지원 다양한 애플리케이션 아키텍처의 유연성을 위한 동기 및 비동기 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의 파티션 키 역할을 하므로 별도의 대화 스레드를 유지하고 모든 스레드의 기록 상태를 검색할 수 있습니다. ## 프로덕션 구성 프로덕션 배포의 경우 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 ``` ## 오류 처리 일반적인 오류 시나리오: + **테이블을 찾을 수 없음**: `table_name` 및 `region_name`을 확인하고 DynamoDB 테이블과 일치시킵니다. + **스로틀링**: `ProvisionedThroughputExceededException`이 표시되면 온디맨드 결제 모드로 전환하거나 프로비저닝된 용량을 늘리는 것이 좋습니다. + **항목 크기 초과**: 체크포인트가 350KB를 초과하는 경우 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/)