Amazon Bedrock와 Boto3로 구현하는 코인 관리 AI 에이전트
-
2024년 11월 22일
시작하며
안녕하세요.
Cocone Engineering에서 주로 결제와 관련된 서버 사이드를 담당하고 있는 K입니다.
저의 주요 업무는 자사가 운영하는 게임의 결제 관련 시스템 개발, 유지보수 및 운영이며, 특히 게임 내 과금과 결제 시스템과 관련된 업무를 중점적으로 맡고 있습니다.
이번에는 자택에서 Agents for Amazon Bedrock을 사용하여 만든 AI 에이전트의 간단한 샘플 애플리케이션을 소개해드리겠습니다.
AI 에이전트란?
AI 에이전트란, 특정 작업이나 목적에 기반하여 자율적으로 의사 결정을 하고 행동을 실행하는 소프트웨어 프로그램을 말합니다. 이를 통해 인간을 대신하여 문제를 해결하거나 업무를 자동화하거나 사용자와 상호 작용할 수 있습니다.
AI 에이전트는 고객 지원, 데이터 분석, 게임 AI 등 다양한 분야에서 사용되고 있으며, 특히 복잡한 작업을 자동화하고 효율적으로 처리하는 능력이 평가받고 있습니다.
생성할 샘플 애플리케이션 개요
이 샘플 애플리케이션에서는 다음 목적에 맞는 AI 에이전트를 생성합니다.
이 AI 에이전트 애플리케이션은 코인에 관한 정보를 쉽게 확인하고 코인 거래를 지원하는 편리한 도구입니다. 사용자는 대화를 통해 코인의 잔액 확인이나 시세 문의를 할 수 있으며, 코인의 종류와 수량을 지정하여 즉시 구매도 가능합니다.
또한, 시장 전체의 코인 구매 이력을 조사하여 시장 동향을 파악하고 더 나은 거래 판단에 활용할 수 있습니다.
이 애플리케이션은 간단한 조작으로 코인의 관리와 거래를 원활하게 수행할 수 있도록 설계되었으며, 사용자의 의사 결정을 효율적으로 지원합니다.
Amazon Bedrock을 사용하여 생성 시작
먼저 AWS 콘솔에서 Amazon Bedrock 페이지를 엽니다.
처음으로 “모델 액세스” 섹션에서 이번에 사용할 Bedrock 기반 모델(Foundation Model: FM)에 대한 액세스를 활성화합니다. 이번에 사용할 모델은 Anthropic사의 Claude 3 Sonnet이며, 리전은 버지니아 북부입니다.
모델 활성화가 완료되면, “오케스트레이션”의 에이전트 섹션으로 이동하여 에이전트 생성을 시작합니다. 다음 설정으로 에이전트를 생성했습니다.
- 에이전트 이름: agent-coin-market-assistant
- 에이전트 설명: 코인 구매를 지원합니다.
- 에이전트를 위한 지시사항: 코인을 구매하려면 코인의 이름과 수량을 알아야 합니다. 고객에게는 정중히 설명해 주십시오.
에이전트의 세부 정보를 입력한 후, 다음으로 액션 그룹을 생성합니다. 액션 그룹이란, AI 에이전트가 특정 작업을 달성하기 위해 일련의 액션(작업 또는 활동)을 그룹화한 것입니다. 이번에는 액션 그룹 타입으로 “함수의 세부 정보 정의”를 선택했습니다.
- 액션 그룹 이름: action-group-coin-market-assistant
- 설명: 사용자는 현재 소유하고 있는 각 코인의 수량과 시세를 확인하고, 코인의 종류와 수량을 지정하여 구매할 수 있습니다.
「새 Lambda 함수를 빠르게 생성하기」를 선택한 후, 다음 3개의 액션 그룹을 생성합니다.
- get-current-coin-balances: 현재 소유하고 있는 각 코인의 수량을 표시합니다.
- get-current-coin-prices: 현재 각 코인의 시세(원)를 표시합니다.
- purchase-coin: 코인 이름과 수량을 전달하여 특정 코인을 원하는 수량만큼 구매할 수 있습니다.
이 중에서, “purchase-coin” 함수에는 인수를 설정합니다. 설정 내용은 아래와 같습니다.
- coin_name: 구매할 코인의 이름 (타입: string, 필수: True)
- quantity: 구매할 코인의 수량 (타입: integer, 필수: True)
다음으로, 생성한 액션 그룹을 기반으로 새로 생성된 Lambda 함수를 편집합니다. AWS 콘솔에서 해당 Lambda 함수를 열면, 샘플 코드가 표시되며, 이를 아래 코드로 변경합니다.
- AWS 콘솔의 Lambda 섹션으로 이동하여 생성된 Lambda 함수를 선택합니다.
- Lambda 함수 편집 화면으로 이동하면 샘플 코드가 표시됩니다. 이 샘플 코드를 삭제하고, 아래 코드로 교체합니다.
import json
import boto3
s3 = boto3.client('s3')
bucket_name = 'agent-coin-market-assistant-bucket'
def convert_params_to_dict(params_list):
params_dict = {}
for param in params_list:
name = param.get("name")
value = param.get("value")
if name is not None:
params_dict[name] = value
return params_dict
def get_balances():
file_name = 'balances.json'
response = s3.get_object(Bucket=bucket_name, Key=file_name)
return json.loads(response['Body'].read().decode('utf-8'))
def get_prices():
file_name = 'prices.json'
response = s3.get_object(Bucket=bucket_name, Key=file_name)
return json.loads(response['Body'].read().decode('utf-8'))
# 코인 구매 후, 지불한 금액의 합계를 반환
def purchase_coin(coin_name, quantity):
# 현재 잔액
balances = get_balances()
# 지정된 코인의 잔액을 찾아서 업데이트
coin_found = False
for coin in balances:
if coin['coin_name'] == coin_name:
coin['quantity'] += quantity
coin_found = True
break
if not coin_found:
raise ValueError(f"Coin {coin_name} not found in balances.")
file_name = 'balances.json'
# 업데이트된 balances 저장
# balances 데이터를 JSON 형식으로 직렬화
balances_data = json.dumps(balances)
# S3에 balances.json을 덮어쓰기 저장
s3.put_object(Bucket=bucket_name, Key=file_name, Body=balances_data)
# 구매로 인해 지불한 금액 계산
coin_price = None
coin_prices = get_prices()
for coin in coin_prices:
if coin['coin_name'] == coin_name:
coin_price = coin['price']
break
return coin_price * quantity
def lambda_handler(event, context):
print(event)
actionGroup = event['actionGroup']
function = event['function']
param = convert_params_to_dict(event.get('parameters', []))
if actionGroup == 'action-group-coin-market-assistant':
if function == 'get-current-coin-balances':
balances = get_balances()
body = f"Your current balances:\n{json.dumps(balances)}."
elif function == 'get-current-coin-prices':
prices = get_prices()
body = f"The prices of the coins available for sale:\n{json.dumps(prices)}."
elif function == 'purchase-coin':
coin_name = param['coin_name']
quantity = int(param['quantity'])
total_price = purchase_coin(coin_name, quantity)
body = (f"{coin_name} coin(s) were purchased in the amount of {quantity}."
f" The total price paid is {total_price} yen.")
else:
body = f"Unknown function {function} for action group {actionGroup}."
else:
body = f"Unknown action group {actionGroup}."
# 에이전트에서 기대하는 형식으로 출력을 포맷팅
responseBody = {
"TEXT": {
"body": body
}
}
action_response = {
'actionGroup': actionGroup,
'function': function,
'functionResponse': {
'responseBody': responseBody
}
}
function_response = {'response': action_response, 'messageVersion': event['messageVersion']}
print(f"Response: {function_response}")
return function_response
Lambda 함수 코드를 편집한 후, 변경 사항을 저장하고 “배포(Deploy)” 버튼을 눌러 Lambda 함수를 업데이트합니다. 일반적으로 이 Lambda 함수를 사용하여 외부 서비스나 자사 서비스 등 기존 시스템과 통합하지만, 이번에는 간단히 S3에 저장된 잔액 정보 및 코인 가격 정보 JSON 파일을 DB 대신 사용합니다. 이를 통해 잔액 정보를 업데이트하거나 코인 가격 정보를 가져옵니다.
S3에 생성한 버킷 이름
agent-coin-market-assistant-bucket
버킷에 저장한 파일
- balances.json:
[
{
"coin_name": "gold",
"quantity": 0
},
{
"coin_name": "silver",
"quantity": 0
},
{
"coin_name": "copper",
"quantity": 0
}
]
- prices.json
[
{
"coin_name": "gold",
"price": 10000
},
{
"coin_name": "silver",
"price": 5000
},
{
"coin_name": "copper",
"price": 1000
}
]
생성한 Lambda 함수 내에서, S3 데이터를 읽고 쓰기 위해 IAM 콘솔에서 “action-group-coin-market-assistant” 롤에 “AmazonS3FullAccess” 권한을 부여합니다. 이를 통해 Lambda 함수가 S3 버킷에 접근하여 잔액 정보와 코인 가격 정보를 처리할 수 있습니다.
저장이 완료되면, AWS 콘솔의 Amazon Bedrock 에이전트 화면으로 돌아가 테스트 화면의 “준비” 버튼을 누릅니다. 이를 통해 생성한 에이전트를 테스트할 준비가 완료됩니다.
잔액이 올바르게 가져와지는지 확인해봅시다.
잔액이 올바르게 가져와지며, 이를 통해 동작 확인이 완료되었습니다.
다음으로, “purchase_history_2024-10-19.json”이라는 파일을 준비하여 동일하게 S3 버킷에 저장합니다. 이는 2024년 10월 19일 모든 사용자의 구매 이력을 포함하고 있습니다.
- purchase_history_2024-10-19.json
[
{
"coin_name": "silver",
"purchase_time": "2024-10-19T00:15:40",
"price": 5000
},
{
"coin_name": "copper",
"purchase_time": "2024-10-19T00:23:54",
"price": 1000
},
{
"coin_name": "copper",
"purchase_time": "2024-10-19T00:52:48",
"price": 1000
}
]
앞서 생성한 에이전트에서 Code Interpreter를 활성화해두면, 에이전트가 데이터를 처리하거나 정보를 추출할 수 있습니다. 이를 통해 에이전트를 호출할 때 문서를 포함할 수 있으며, 테스트 화면에서도 여러 문서를 첨부하여 테스트할 수 있습니다.
이번에는 AWS SDK for Python (Boto3)와 Streamlit을 사용하여 에이전트를 호출합니다. 이를 위해 먼저 에이전트 별칭(Alias)과 버전을 생성해야 합니다. 에이전트 ID와 별칭 ID는 에이전트를 호출할 때 필요하므로 메모해둡니다. Boto3의 버전은 1.35.46을 사용하고 있습니다.
또한 “purchase_history_2024-10-19.json” 파일은 에이전트가 호출할 수 있도록 AmazonBedrockExecutionRoleForAgents IAM 롤에 “AmazonS3FullAccess” 권한을 부여했습니다.
다음은 Streamlit을 사용해 에이전트를 호출하는 소스 코드입니다.
import streamlit as st
import boto3
import uuid
import os
# Streamlit UI 생성
st.title("CoinMarketBot")
# 상수 정의
AGENT_ID = os.getenv('AGENT_ID')
AGENT_ALIAS_ID = os.getenv('AGENT_ALIAS_ID')
# 고유 사용자 식별자
MEMORY_ID = os.getenv('MEMORY_ID')
# Bedrock 에이전트 클라이언트
client = boto3.client("bedrock-agent-runtime")
# 세션 ID 생성
if "session_id" not in st.session_state:
st.session_state.session_id = str(uuid.uuid4())
# 대화 기록 초기화
if "messages" not in st.session_state:
st.session_state.messages = []
# 과거 대화 기록 표시
for message in st.session_state.messages:
with st.chat_message(message["role"]):
for content in message["content"]:
# 텍스트 처리
if content["type"] == "text":
st.write(content["content"])
# 파일 처리
elif content["type"] == "file":
st.image(content["bytes"], caption=content["name"])
# 채팅 입력 필드 정의
if prompt := st.chat_input("메시지를 입력하세요"):
# 사용자의 입력을 화면에 표시
with st.chat_message("user"):
st.markdown(prompt)
# 사용자의 메시지를 세션에 추가
st.session_state.messages.append({"role": "user", "content": [{"type": "text", "content": prompt}]})
# AI의 응답을 가져와 표시
with st.chat_message("assistant"):
# 사용자 입력을 가져와 에이전트로 전송
response = client.invoke_agent(
agentId=AGENT_ID,
agentAliasId=AGENT_ALIAS_ID,
sessionId=st.session_state.session_id,
inputText=prompt,
memoryId=MEMORY_ID,
enableTrace=False,
endSession=False,
sessionState={
"files": [
{
"name": "purchase_history_2024-10-19.json",
"source": {
"sourceType": "S3",
"s3Location": {
"uri": "s3://agent-coin-market-assistant-bucket/purchase_history_2024-10-19.json"
},
},
"useCase": "CODE_INTERPRETER",
}
]
},
)
if stream := response.get("completion"):
result = []
for event in stream:
# 파일이 있는 경우
if "files" in event:
files = event["files"]["files"]
for file in files:
# 이미지를 표시
st.image(file["bytes"], caption=file["name"])
# 파일 정보를 리스트에 저장
result.append(
{
"type": "file",
"name": file["name"],
"bytes": file["bytes"],
}
)
# 텍스트 청크가 있는 경우
if "chunk" in event:
chunk = event["chunk"]
answer = chunk["bytes"].decode()
st.markdown(answer)
# 텍스트를 리스트에 저장
result.append({"type": "text", "content": answer})
# 메시지 저장
st.session_state.messages.append(
{"role": "assistant", "content": result}
)
동작 확인을 진행하며 결과물을 확인해봅시다.
- 현재 잔액을 확인합니다
에이전트에게 현재 코인 잔액을 조회합니다.
⇩- 마켓 코인 가격을 확인합니다
현재 마켓에서 거래되는 코인의 가격 정보를 에이전트에게 요청합니다.
⇩- 코인을 구매합니다
구매할 코인의 종류와 수량을 지정하여 에이전트를 통해 코인을 구매합니다.
마지막으로, 코인 잔액이 방금 구매한 내용을 반영하여 증가한 것을 확인하고, 향후 구매를 위해 다른 사용자의 과거 구매 이력도 조사해봅시다.
- 현재 코인 잔액을 확인합니다
에이전트에게 현재 코인 잔액을 요청하여 구매 내용이 반영되었는지 확인합니다.
⇩- 2024년 10월 19일의 다른 사용자 구매 이력을 확인합니다
2024년 10월 19일에 다른 사용자가 구매한 코인 이력을 에이전트를 통해 조회하고, 이를 바탕으로 향후 구매를 계획합니다.
이로써 모든 작업이 문제없이 동작하는 것을 확인할 수 있었습니다.
이번 샘플 애플리케이션은 간단한 예제였지만, 이를 기반으로 다양한 기능을 확장할 수 있습니다. 예를 들어, 복잡한 데이터 분석이나 자동화된 거래 관리 등의 응용 범위를 넓힐 수 있습니다.
직접 테스트해 보시길 추천드립니다!
참고 자료
참고한 서적
- 「Amazon Bedrock 생성 AI 애플리케이션 개발 입문 [AWS 심화 가이드]」
참고한 웹사이트
- Amazon Web Services 블로그
「Amazon Bedrock의 에이전트가 메모리 유지와 코드 해석을 지원하게 되었습니다 (프리뷰 중)」
https://aws.amazon.com/jp/blogs/news/agents-for-amazon-bedrock-now-support-memory-retention-and-code-interpretation-preview/- Boto3 1.35.47 문서
「invoke_agent」
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/bedrock-agent-runtime/client/invoke_agent.html
마무리
이번 블로그에서는 Amazon Bedrock을 사용해 AI 에이전트를 구축하고, 코인 잔액 확인과 구매 이력 조회를 통해 에이전트의 동작을 확인했습니다.
이 샘플 애플리케이션은 저의 개인 학습을 목적으로 기본적인 기능을 조합하여 제작한 것으로, 실제 업무에서 이러한 AI 도구를 사용하는 것은 아닙니다. 따라서 업무에서 사용하려면 추가적인 검증이 필요합니다.
다만, 이러한 AI 기반 시스템은 앞으로 다양한 분야에서 활용될 가능성이 크기 때문에 최신 기술을 꾸준히 배우는 것이 엔지니어에게 매우 중요하다고 생각합니다.
저도 일상 업무에서 작은 부분이라도 AI를 활용할 수 있도록 계속 학습해 나갈 것입니다.
끝까지 읽어주셔서 감사합니다.
