# DynamoDB에서 에이전틱 AI 사용
Amazon DynamoDB는 어떤 규모에서든 10밀리초 미만의 성능을 제공하는 완전 관리형 서버리스 분산형 NoSQL 데이터베이스입니다. DynamoDB는 처리량이 많은 워크로드에 최적화되어 있으며 생성형 AI 모델과 통합하여 기능을 확장할 수 있습니다. 생성형 AI 모델을 사용하면 DynamoDB 테이블에 저장된 데이터를 실시간으로 작업하고 상황에 맞게 인식되고 고도로 맞춤화된 애플리케이션을 구축할 수 있습니다. 또한 비즈니스, 사용자 및 애플리케이션 데이터를 완전하게 활용하여 생성형 AI 솔루션을 사용자 지정함으로써 최종 사용자 경험을 향상할 수 있습니다.
생성형 AI 및 AWS가 생성형 AI 애플리케이션 구축을 지원하는 솔루션에 대해 자세히 알아보려면 [생성형 AI를 사용하여 비즈니스 전환](https://aws.amazon.com/ai/generative-ai/)을 참조하시기 바랍니다.
**Topics**
+ [DynamoDB용 생성형 AI 사용 사례](#gen-ai-use-case-ddb)
+ [DynamoDB용 생성형 AI 블로그](#gen-ai-blogs)
+ [OpenSearch Service와 DynamoDB의 제로 ETL 통합 활용](ddb-and-amazon-bedrock.md)
+ [DynamoDB를 LangGraph 에이전트의 체크포인트 스토어로 사용](ddb-langgraph-checkpoint.md)
## DynamoDB용 생성형 AI 사용 사례
DynamoDB는 [파운데이션 모델(FM)](https://aws.amazon.com/what-is/foundation-models/)로 구축된 챗봇 및 고객 상담 센터와 같은 AI 기반 대화형 애플리케이션에 널리 사용됩니다. Amazon Bedrock, Amazon SageMaker AI 또는 기타 모델 공급업체를 통해 FM에 액세스할 수 있습니다. 이러한 애플리케이션은 일반적으로 DynamoDB를 사용하여 3가지 데이터 패턴(애플리케이션 데이터, 비즈니스 데이터 및 사용자 데이터) 전반에서 맞춤화를 개선하고 사용자 경험을 향상시킵니다. 이러한 데이터 패턴의 몇 가지 예는 다음과 같습니다.
+ [LangChain](https://js.langchain.com/v0.1/docs/integrations/chat_memory/dynamodb/), [LlamaIndex](https://docs.llamaindex.ai/en/stable/examples/docstore/DynamoDBDocstoreDemo/) 또는 사용자 지정 코드와의 통합을 통해 채팅 메시지 기록 등의 애플리케이션 데이터를 저장합니다. 이 컨텍스트를 통해 모델이 사용자와 전후로 *대화*할 수 있도록 하여 사용자 경험을 향상시킵니다.
+ 인벤토리, 요금 및 설명서와 같은 비즈니스 데이터를 활용하여 사용자 지정 사용자 환경을 조성합니다.
+ 웹 기록, 과거 주문 및 사용자 기본 설정 등의 사용자 데이터를 적용하여 맞춤화된 답변을 제공합니다.
예를 들어 보험 회사는 DynamoDB로 챗봇을 빌드하여 거의 실시간 데이터에 대한 [검색 증강 생성(RAG)](https://docs.aws.amazon.com/sagemaker/latest/dg/jumpstart-foundation-models-customize-rag.html) 기반 생성형 AI 모델에 대한 액세스를 제공할 수 있습니다. 이러한 데이터의 예로는 실시간 모기지 요금, 제품 요금, 규정 준수/표준 계약 사본, 사용자 웹 기록 및 사용자 기본 설정이 있습니다. 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/)
# OpenSearch Service와 DynamoDB의 제로 ETL 통합 활용
Amazon Bedrock을 DynamoDB와 함께 사용하여 Amazon Titan 및 기타 서드 파티 모델 등 [파운데이션 모델(FM)](https://aws.amazon.com/what-is/foundation-models/)에 대한 서버리스 액세스를 제공할 수 있습니다. 생성형 AI 애플리케이션을 구축할 때 Amazon OpenSearch Service와의 제로 ETL 통합을 활용하여 벡터 검색 기능을 사용하는 것이 가능합니다. [DynamoDB 제로 ETL에서 OpenSearch로의 생성형 AI 통합 및 Amazon Bedrock](https://catalog.workshops.aws/dynamodb-labs/en-US/dynamodb-opensearch-zetl) 워크숍에서는 OpenSearch와의 DynamoDB 제로 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)의 7개 모듈 중 하나입니다. 모든 AWS 계정에서 이 워크숍을 실행할 수 있습니다.
또한 DynamoDB와 OpenSearch Service 간의 제로 ETL 통합을 설정하는 방법에 대한 다음 블로그 게시물을 참조하시기 바랍니다. 이 블로그 게시물에서는 수신 데이터에 Amazon Bedrock을 사용하여 임베딩을 자동으로 생성하도록 OpenSearch Service에서 모델 커넥터를 설정하는 방법도 설명해 줍니다. [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/)는 대규모 언어 모델(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/)