Bedrock을 활용한 이미지 Docent 데모 프로그램 개발하는 방법
AWS Bedrock 기반 docent 시스템은 교육적 가치를 증대시키고 문화 체험을 강화하여, 사용자들에게 더욱 풍부하고 의미 있는 학습 및 감상 경험을 제공합니다.
1. Amazon Bedrock Docent 프로젝트 개요
AWS Bedrock을 활용한 이미지 인식 기반 docent 시스템은 다양한 측면에서 가치를 창출하고 폭넓은 사용 사례를 제공합니다. 우선, 이 시스템은 접근성 향상에 크게 기여합니다. 시각장애인들에게 시각적 정보를 음성으로 설명함으로써 문화 체험의 기회를 확대하고, 저시력자나 색맹인 사람들에게도 세부적인 설명을 제공하여 더 풍부한 경험을 가능하게 합니다.
교육적 측면에서도 이 시스템은 큰 가치를 지닙니다. 학생들에게 미술, 역사, 과학 등 다양한 분야의 시각 자료에 대한 상세한 설명을 제공함으로써 학습 효과를 높이고, 전문가 수준의 해설로 교육의 질을 향상시킵니다.
또한, 이 시스템은 문화 체험을 강화하는 데 중요한 역할을 합니다. 전시회나 박물관 방문객에게 개인화된 가이드 경험을 제공하여, 각자의 관심사와 페이스에 맞는 설명을 들을 수 있게 합니다. 이를 통해 작품에 대한 깊이 있는 이해와 감상이 가능해지며, 문화 체험의 질을 한층 높일 수 있습니다.
이렇게 AWS Bedrock 기반 docent 시스템은 접근성, 교육, 문화 체험 등 다양한 영역에서 혁신적인 솔루션을 제공하며, 사용자들에게 더욱 풍부하고 의미 있는 경험을 선사합니다.
2. 아키텍처 및 기술 구현
전체 아키텍처 개요
프로그램의 작동 방식은 다음과 같습니다. 우선, 시스템은 OpenSearch Serverless의 Vector Database를 조회하여 요청된 이미지 데이터의 존재 여부를 확인합니다. 만약 해당 이미지 정보가 이미 Vector Database에 저장되어 있다면, 시스템은 추가적인 이미지 분석 과정을 생략하고 즉시 저장된 이미지 설명을 활용합니다. 이 경우, 인공지능 기술을 활용하여 저장된 정보를 정제하고 조직화하여 사용자에게 최적화된 형태로 제공합니다.
반면, Vector Database에 해당 이미지 정보가 존재하지 않을 경우, 시스템은 Amazon Rekognition 서비스를 호출하여 실시간으로 이미지 분석을 수행합니다. Rekognition을 통해 얻은 분석 결과는 즉시 사용자에게 전달되며, 이를 통해 새로운 이미지에 대해서도 신속하고 정확한 설명을 제공할 수 있습니다.
이러한 이중 접근 방식을 통해, 본 프로그램은 기존 데이터의 재사용성을 극대화하면서도 새로운 정보에 대한 실시간 처리 능력을 갖추고 있어, 효율성과 정확성을 동시에 확보하고 있습니다.
1. Client에서 이미지를 업로드 합니다.
2. 업로드된 이미지는 Bedrock을 통해 임베딩 되어지고 이를 Opensearch Serverless를 통해 knn방식으로 검색합니다.
3.1. 유사한 이미지가 있는 경우 VectorDB에 저장되어있는 설명을 Bedrock에 전달합니다.
3.2. Bedrock에서 3-1에서 받은 내용을 토대로 스크립트를 생성하고 이를 polly로 보내 음성으로 변환합니다.
4.1. 유사한 이미지가 없는 경우, Rekognition을 통해 이미지 라벨링 및 속성을 분석합니다.
4.2. Rekognition을 통해 분석한 내용을 토대로 Bedrock에 전달하여 이미지를 분석합니다.
4.3. Polly를 이용하여 Bedrock에서 전달 받은 스크립트를 음성으로 전환합니다.
전제조건
AWS 자격증명 및 Bedrock 설정
- 해당 프로그램은 Bedrock, Opensearch Serverless, Rekognition, Polly를 사용중이기에 해당 권한이 필요합니다. Opensearch Serverless의 경우 아래의 링크를 통해 적절한 권한을 부여해야합니다. 이외에 AmazonBedrockFullAccess, AmazonRekognitionFullAccess, AmazonPollyFullAccess를 권한을 부여하여 자격증명을 진행합니다.
[+]구성 설정 지정 및 보기 https://docs.aws.amazon.com/ko_kr/cli/v1/userguide/cli-configure-files.html
- Bedrock의 Titan Embeddings G1 - Text, Claude 3 Haiku 액세스 권한을 요청해야합니다.
python IDE 설정
- PythonIDE의 터미널에서 아래의 명령어를 실행하여 라이브러리를 설치합니다.
pip install boto3, streamlit, requests, requests_aws4auth
- 폴더 구조는 아래와 같습니다. 아래와 같이 파이썬 파일을 생성합니다.
Opensearch Serverless 생성 및 데이터 저장
- Opensearch Serverless를 검색하여 벡터 검색을 클릭합니다.
- 표준 생성을 클릭하고 액세스 유형에 "퍼블릭"으로 설정하여 외부에서 접속할수있게 합니다.
- Opensearch Serverless를 생성한뒤에 아래의 방식으로 index에 설명과 이미지를 임베딩 합니다.
import json
import boto3
import base64
import requests
from requests_aws4auth import AWS4Auth
def generate_embeddings(image_path=None, text=None):
bedrock_runtime = boto3.client(
service_name="bedrock-runtime",
region_name="us-east-1",
)
user_input = {}
if image_path is not None:
with open(image_path, "rb") as image_file:
image = base64.b64encode(image_file.read()).decode('utf-8')
user_input["inputImage"] = image
if text is not None:
user_input["inputText"] = text
if not user_input:
raise ValueError("One user input of an image or a text is required")
body = json.dumps(user_input)
response = bedrock_runtime.invoke_model(
body=body,
modelId="amazon.titan-embed-image-v1",
accept="application/json",
contentType="application/json"
)
response_body = json.loads(response.get("body").read())
embedding_error = response_body.get("message")
if embedding_error:
raise Exception(f"Embeddings generation error: {embedding_error}")
return response_body.get("embedding")
service = 'aoss'
region = 'us-east-1'
host = 'https:/<host>.us-east-1.aoss.amazonaws.com'
index = '<index>'
datatype = '_doc'
url = host + '/' + index + '/' + datatype
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)
image_path= "이미지 경로"
description="그림에 대한 설명"
embedding=generate_embeddings(image_path=image_path)
headers = {'Content-Type': 'application/json'}
document = {'embedding': embedding,'description': description}
response = requests.post(url, auth=awsauth, json=document, headers=headers)
print(response)
Frontend 구성 (Streamlit)
streamlit_front.py
사용자가 이미지를 업로드하면 프로젝트에 생성된 여러가지 모듈을 통해 AWS 서비스들을 활용하여 이미지를 분석하고 설명을 생성합니다. 업로드된 이미지는 저장되고, Rekognition으로 분석되거나 사용자 정의 검색 함수를 통해 설명이 생성됩니다. 그 후 Amazon Bedrock을 사용하여 더 자세한 설명을 생성하고, 최종적으로 Amazon Polly를 통해 이 설명을 음성으로 변환합니다. 생성된 설명은 텍스트로 표시되고 음성으로도 재생됩니다.
import streamlit as st
import boto3
from query_module import search
from rekognition_module import detect_labels_properties
from bedrock_module import get_bedrock
import os
polly_client = boto3.client('polly')
def save_uploaded_file(uploadedfile):
if not os.path.exists("src"):
os.makedirs("src")
with open(os.path.join("src", uploadedfile.name), "wb") as f:
f.write(uploadedfile.getbuffer())
return os.path.join("src", uploadedfile.name)
def text_to_speech(text):
response = polly_client.synthesize_speech(
Engine='standard',
LanguageCode='ko-KR',
Text=text,
OutputFormat='mp3',
VoiceId='Seoyeon'
)
return response['AudioStream'].read()
st.title('Bedrock Docent Demo')
uploaded_file = st.file_uploader("Jpg형식의 이미지를 업로드 해주세요....", type="jpg")
if uploaded_file is not None:
st.image(uploaded_file, caption='이미지가 업로드 되었습니다.', use_column_width=True)
with st.spinner('이미지를 분석하고 설명을 생성하는 중입니다...'):
saved_file_path = save_uploaded_file(uploaded_file)
saved_file_path = saved_file_path.replace("\\", "/")
image_path="<파이썬 파일이 포함한 위치>"+saved_file_path
description = search(image_path)
if description:
desc_prompt=description
else:
desc_prompt=detect_labels_properties(image_path)
final_description=get_bedrock(desc_prompt+"위의 내용을 큐레이터 처럼 500자 이내로 설명해주세요")
audio_bytes = text_to_speech(final_description)
st.audio(audio_bytes, format='audio/mp3')
st.write(final_description)
- Backend 구성이 완료되었다면 아래의 명령어를 실행시켜 서버를 실행시킵니다.
streamlit run <example.py>
Backend 구성 ( Bedrock, Opensearch Servereless, Rekognition, Polly)
query_module.py
이 코드는 AWS Bedrock과 OpenSearch Serverless를 활용하여 이미지 검색 및 설명 생성 기능을 구현합니다. 'get_embedding' 함수는 AWS Bedrock을 사용하여 이미지나 텍스트의 임베딩을 생성합니다. 'search' 함수는 이 임베딩을 사용하여 OpenSearch Serverless에서 유사한 이미지를 검색합니다. 검색 결과의 신뢰도 점수가 0.6을 초과하면 해당 이미지의 설명을 반환하고, 그렇지 않으면 False를 반환합니다.
import json
import boto3
import requests
from requests_aws4auth import AWS4Auth
import base64
def get_embedding(image_path=None,text=None):
bedrock_runtime = boto3.client(
service_name="bedrock-runtime",
region_name="us-east-1",
)
user_input = {}
if image_path is not None:
with open(image_path, "rb") as image_file:
image = base64.b64encode(image_file.read()).decode('utf-8')
user_input["inputImage"] = image
if text is not None:
user_input["inputText"] = text
if not user_input:
raise ValueError("One user input of an image or a text is required")
body = json.dumps(user_input)
response = bedrock_runtime.invoke_model(
body=body,
modelId="amazon.titan-embed-image-v1",
accept="application/json",
contentType="application/json"
)
response_body = json.loads(response.get("body").read())
embedding_error = response_body.get("message")
if embedding_error:
raise Exception(f"Embeddings generation error: {embedding_error}")
return response_body.get("embedding")
def search(image_path):
limit = 1
region = 'us-east-1'
host = '<Opensearch Serverless host>'
index = '<Index>'
embedding = get_embedding(image_path=image_path)
credentials = boto3.Session().get_credentials()
awsauth = AWS4Auth(
credentials.access_key,
credentials.secret_key,
region,
"aoss",
session_token=credentials.token
)
datatype = '_search'
url = host + '/' + index + '/' + datatype
headers = {'Content-Type': 'application/json'}
document = {
'size': limit,
'query': {
'knn': {
'embedding': {
'vector': embedding,
'k': limit
}
}
}
}
try:
response = requests.get(url, auth=awsauth, json=document, headers=headers)
response.raise_for_status()
result = response.json()
hits = result.get('hits', {}).get('hits', [])
if hits:
best_hit = max(hits, key=lambda x: x['_score'])
score = best_hit['_score']
if score > 0.6:
description = best_hit['_source']['description']
return description
return False
except requests.exceptions.RequestException as e:
print(f"Error during API request: {e}")
return False
except KeyError as e:
print(f"Unexpected response format: {e}")
return False
except Exception as e:
print(f"An unexpected error occurred: {e}")
return False
rekognition_module.py
해당 코드는 Amazon Rekognition을 사용하여 이미지 분석을 수행합니다. 'detect_labels_properties' 함수는 이미지 파일을 입력받아 레이블 감지와 이미지 속성 분석을 수행합니다. 90% 이상의 신뢰도를 가진 레이블만 추출하고, 'extract_color_and_percent' 함수를 통해 이미지의 전경과 배경 색상 정보를 추출합니다. 분석 결과는 키워드(레이블), 피사체 색상, 배경 색상을 포함한 문자열 형태로 반환됩니다. 이 코드는 예외 처리를 통해 에러 상황에 대응하며, boto3 라이브러리를 사용하여 AWS 서비스와 상호작용합니다.
import boto3
from botocore.exceptions import ClientError
def extract_color_and_percent(data):
result = []
for color in data['DominantColors']:
result.append({
'CSSColor': color['CSSColor'],
'PixelPercent': color['PixelPercent']
})
return result
def detect_labels_properties(photo):
client = boto3.client('rekognition')
labels=[]
try:
with open(photo, 'rb') as image_file:
response = client.detect_labels(
Image={'Bytes': image_file.read()},
MaxLabels=10,
MinConfidence=70,
Features=['GENERAL_LABELS', 'IMAGE_PROPERTIES']
)
for label in response['Labels']:
if label['Confidence']>90:
labels.append(label['Name'])
image_properties = response.get('ImageProperties', {})
foreground = image_properties.get('Foreground', {})
foreground=extract_color_and_percent(foreground)
background = image_properties.get('Background', {})
background=extract_color_and_percent(background)
result = '''
해당 그림에는 아래의 내용들을 포함하고 있습니다.
키워드 : {}
피사체 색상 : {}
배경 화면 색상: {}
'''.format(labels, foreground, background)
return result
except ClientError as e:
print(f"Error occurred: {e}")
return None, None
bedrock_module.py
Amazon Bedrock을 사용하여 Anthropic의 Claude 3 Haiku 모델에 접근하는 함수를 정의합니다. 'get_bedrock' 함수는 사용자로부터 받은 프롬프트를 입력으로 받아 AI 모델에 전달합니다. 함수는 boto3 클라이언트를 사용하여 Bedrock 런타임에 연결하고, 지정된 모델 ID와 파라미터(최대 토큰 수, 온도 등)를 사용하여 요청을 구성합니다. 모델의 응답은 JSON 형식으로 받아 처리되며, 텍스트 응답만 추출하여 반환합니다. 예외 처리를 통해 오류 상황에 대응하며, 오류 발생 시 프로그램을 종료합니다.
import boto3
import json
def get_bedrock(prompt):
client = boto3.client("bedrock-runtime", region_name="us-east-1")
model_id = "anthropic.claude-3-haiku-20240307-v1:0"
native_request = {
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 512,
"temperature": 0.5,
"messages": [
{
"role": "user",
"content": [{"type": "text", "text": prompt}],
}
],
}
request = json.dumps(native_request)
try:
response = client.invoke_model(modelId=model_id, body=request)
model_response = json.loads(response["body"].read())
response_text = model_response["content"][0]["text"]
return response_text
except Exception as e:
print(e)
exit(1)
결론 및 참고 자료
결론적으로, 본 시스템은 사용자 경험을 최적화하기 위해 두 가지 접근 방식을 채택하고 있습니다. 사용자가 업로드한 이미지는 임베딩되어 기존 데이터베이스에서 검색됩니다. 만약 일치하는 이미지가 발견되면, 미리 준비된 전문적인 설명과 함께 Bedrock을 이용하여 추가적인 설명을 제공합니다. 반면, 새로운 이미지의 경우 Amazon Rekognition을 통해 실시간으로 분석하여 설명을 생성합니다. 두 경우 모두 결과는 텍스트와 음성으로 사용자에게 전달되어, 다양한 상황과 사용자 요구에 맞는 유연하고 효과적인 이미지 설명 서비스를 제공합니다.
[+] PythonSDK용 코드 예제(Boto3)
https://docs.aws.amazon.com/ko_kr/code-library/latest/ug/python_3_code_examples.html
관련 콘텐츠
AWS 공식업데이트됨 2년 전- AWS 공식업데이트됨 10달 전
- AWS 공식업데이트됨 2달 전