AWS AI 서비스를 활용한 지능형 제품 관리 및 검색 시스템 구현

6분 분량

콘텐츠 수준: 중급

이 기사는 AWS Rekognition을 사용하여 이미지를 자동으로 분석하고 라벨링하는 과정, 그 결과를 OpenSearch에 저장하여 효율적인 유사도 기반 검색을 구현하는 방법, 그리고 AWS Bedrock의 AI 모델을 활용하여 검색 결과를 개선하고 풍부하게 만드는 통합 시스템의 구현 과정을 설명합니다

사용사례

대규모 온라인 쇼핑몰과 패션 트렌드 분석 플랫폼에서 이 시스템을 활용할 수 있습니다. 온라인 쇼핑몰에서는 방대한 양의 제품 이미지를 자동으로 분류하고 태깅하여 관리 효율성을 높이고, 고객들이 원하는 스타일의 제품을 쉽게 찾을 수 있도록 지원합니다. 또한, AI 기술을 활용하여 일관된 품질의 제품 설명을 자동으로 생성함으로써 상품 정보의 질을 향상시킵니다.

사용되는 서비스

Amazon S3
Amazon OpenSearch Service
Amazon Rekognition
Amazon Bedrock

전제조건

본 개발은 Amazon Bedrock을 지원하는 리전인 us-east-1(버지니아 북부)에서 진행되었습니다.
- 해당 기사에서는 Bedrock의 Claude 3 haiku모델을 사용하며 권한을 부여하여야합니다.
이미지 저장을 위한 Amazon S3 버킷이 사전에 생성되어 있어야 합니다.
본 예제에서는 개발 편의를 위해 로컬 환경에서 AdministratorAccess 권한을 사용하였습니다. 실제 운영 환경에서는 보안을 위해 각 서비스에 대한 최소 권한 정책을 구성하여 사용하는 것을 강력히 권장합니다.

1. Opensearch 생성 및 인덱스 생성

[+] Amazon OpenSearch Service 서비스 도메인 생성

https://docs.aws.amazon.com/ko_kr/opensearch-service/latest/developerguide/gsgcreate-domain.html

Amazon OpenSearch Service 선택하여 [도메인 생성(Create domain)]을 선택합니다.
도메인의 이름을 입력한뒤에 도메인 생성 방법으로 [표준 생성]을 선택합니다.
템플릿의 경우 개발/테스트를 선택합니다.
배포 옵션으로는 대기 모드가 있는 도메인을 선택합니다.
버전(Version)에서 최신 버전을 선택합니다.
지금은 데이터 노드, 웜 및 콜드 데이터 스토리지, 전용 마스터 노드, 스냅샷 구성, 사용자 엔드포인트 섹션을 무시하세요.
간단한 설명을 위해 퍼블릭 액세스 도메인을 사용합니다. [네트워크(Network)]에서 [퍼블릭 액세스(Public access)]를 선택합니다.
세분화된 액세스 제어 설정에서 세분화된 액세스 제어 활성화 확인란을 선택한 상태로 유지합니다. 마스터 사용자 생성을 선택하고 사용자 이름과 암호를 입력합니다.
[액세스 정책]에서 [도메인 수준 액세스 정책 구성]을 선택하여 아래와 같은 정책을 참고하여 수정합니다.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "*"
      },
      "Action": "es:*",
      "Resource":"<opensearch_arn>/*"
    }
  ]
}

이후 대시보드에 접속하여 [Dev tools]에 접속하여 아래의 Command 를 입력하여 인덱스를 생성합니다.

PUT <index_name>

2. Python 프로젝트 생성 및 환경구성

pip install streamlit
pip install boto3
pip install requests-aws4auth
pip install requests
pip install Pillow

3. 코드 작성

Python IDE를 실행시키고 아래의 필요한 라이브러리를 설치합니다.

pip install streamlit
pip install boto3
pip install requests-aws4auth
pip install requests
pip install Pillow

페이지는 main.py, input_page.py,display_page.py로 나누어서 생성합니다.

main.py

상품 관리 웹 애플리케이션을 구현하는 메인 페이지 입니다.
사이드바에서 "상품 입력"과 "상품 검색" 두 페이지 중 하나를 선택할 수 있습니다.

import streamlit as st  
import pandas as pd  
  
from input_page import product_input_page  
from display_page import product_search_page  
  
if 'products' not in st.session_state:  
    st.session_state.products = pd.DataFrame(columns=['Name', 'Description', 'Image'])  
  
def main():  
    st.sidebar.title("네비게이션")  
    page = st.sidebar.radio("페이지 선택", ["상품 입력", "상품 검색"])  
    if page == "상품 입력":  
        product_input_page()  
    elif page == "상품 검색":  
        product_search_page()  
  
  
if __name__ == "__main__":  
    main()

input_page.py

상품 등록을 위한 페이지로, 사용자로부터 상품 이미지, 이름, 설명을 입력받습니다.
업로드된 이미지는 AWS S3에 저장되고, AWS Rekognition을 사용하여 이미지에서 자동으로 라벨(태그)을 추출합니다.
입력된 상품 정보(이름, 설명, 이미지 경로, 라벨)는 OpenSearch에 저장되어 검색 가능한 형태로 관리됩니다.

import streamlit as st  
import boto3  
import requests  
from requests_aws4auth import AWS4Auth  
  
host = '<opensearch_host>'  
region = '<region>'  
service = 'es'  
  
AWS_ACCESS_KEY_ID = "<AWS_ACCESS_KEY_ID>"  
AWS_SECRET_ACCESS_KEY = "<AWS_SECRET_ACCESS_KEY>"  
  
s3 = boto3.client('s3')  
  
bucket_name="<bucket_name>"  
  
rekognition = boto3.client(  
    'rekognition',  
    aws_access_key_id=AWS_ACCESS_KEY_ID,  
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY,  
    region_name=region  
)  
  
def send_opensearch(index, datatype, document):  
    credentials = boto3.Session().get_credentials()  
    awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)  
    url = host + '/' + index + '/' + datatype  
    headers = {"Content-Type": "application/json"}  
    r = requests.post(url, auth=awsauth, json=document, headers=headers)  
    print(f"Status Code: {r.status_code}")  
    print(f"Response: {r.text}")  
  
  
def analyze_image(bucket, image_name):  
    try:  
        response = rekognition.detect_labels(  
            Image={  
                'S3Object': {  
                    'Bucket': bucket,  
                    'Name': image_name  
                }  
            },  
            MaxLabels=5,
            MinConfidence=70 )  
  
        labels = [label['Name'] for label in response['Labels']]  
        return labels  
    except Exception as e:  
        st.error(f"이미지 분석 중 오류 발생: {str(e)}")  
        return []  
  
  
def upload_to_s3(file, bucket_name):  
    try:  
        s3.upload_fileobj(file, bucket_name, file.name)  
        st.success(f"파일 '{file.name}'이(가) '{bucket_name}' 버킷에 업로드되었습니다.")  
    except Exception as e:  
        raise Exception(f"파일 '{file.name}' 업로드 중 오류 발생: {e}")  
  
  
  
def product_input_page():  
    st.title("상품 입력")  
  
    uploaded_file = st.file_uploader("업로드할 파일 선택", type=['png', 'jpg', 'jpeg'])  
  
    name = st.text_input("상품 이름")  
  
    description = st.text_area("상품 설명")  
  
    if st.button("상품 등록"):  
        if name and description and uploaded_file:  
            try:  
                upload_to_s3(uploaded_file, bucket_name)  
  
                labels = analyze_image(bucket_name, uploaded_file.name)  
  
                index, datatype = 'product', '_doc'  
                docs = {  
                    "name": name,  
                    "description": description,  
                    "file_path": f"{bucket_name}/{uploaded_file.name}",  
                    "labels": labels  
                }  
                send_opensearch(index, datatype, docs)  
                st.success("상품이 성공적으로 등록되었습니다.")  
  
            except Exception as e:  
                st.error(f"상품 등록 중 오류 발생: {str(e)}")  
        else:  
            st.warning("상품 이름, 설명, 그리고 파일을 모두 입력해주세요.")

display_page.py

OpenSearch를 통해 상품을 검색하고, 검색 결과에 대해 AWS Bedrock의 Claude AI 모델을 사용하여 상품 설명을 생성합니다.
검색된 상품의 이미지는 AWS S3에서 가져와서 표시하며, 각 상품의 이름, 설명, 이미지, 라벨 정보를 보여줍니다.
AWS 서비스들(OpenSearch, S3, Bedrock)과의 연동을 위해 boto3 클라이언트를 사용하며, 인증은 AWS4Auth를 통해 처리합니다.

import streamlit as st  
import boto3  
from requests_aws4auth import AWS4Auth  
import requests  
import json  
from PIL import Image  
import io  
from botocore.exceptions import ClientError  
  
host = '<opensearch_host>'  
region = '<region>' 
index = "<index>"  
service = 'es'  
  
def search_opensearch(keyword):  
    session = boto3.Session()  
    credentials = session.get_credentials()  
    awsauth = AWS4Auth(credentials.access_key,  
                       credentials.secret_key,  
                       region,  
                       'es',  
                       session_token=credentials.token)  
    url = f"{host}/{index}/_search"  
    query = {  
        "query": {  
            "multi_match": {  
                "query": keyword,  
                "fields": ["name", "labels"],  
                "type": "best_fields",  
                "tie_breaker": 0.3  
            }  
        }  
    }  
  
    headers = {"Content-Type": "application/json"}  
  
    response = requests.get(url, auth=awsauth, headers=headers, data=json.dumps(query))  
  
    if response.status_code == 200:  
        data = response.json()  
        parsed_results = []  
        for hit in data['hits']['hits']:  
            source = hit['_source']  
            parsed_item = {  
                'name': source.get('name', ''),  
                'description': source.get('description', ''),  
                'file_path': source.get('file_path', ''),  
                'labels': source.get('labels', [])  
            }  
            parsed_results.append(parsed_item)  
        return parsed_results  
  
    else:  
        return {"error": f"Search request failed: {response.status_code}", "details": response.text}  
  
  
def get_image(bucket_name,image_key):  
    s3 = boto3.client('s3')  
    if bucket_name and image_key:  
        try:  
  
            response = s3.get_object(Bucket=bucket_name, Key=image_key)  
  
            image_content = response['Body'].read()  
  
            image = Image.open(io.BytesIO(image_content))  
            return image  
  
        except Exception as e:  
            st.error(f"에러 발생: {str(e)}")  
    else:  
        st.info('버킷 이름과 이미지 경로를 입력하세요.')  
  
def invoke_claude(description):  
    try:  
        client = boto3.client("bedrock-runtime", region_name="us-east-1")  
        model_id="anthropic.claude-3-haiku-20240307-v1:0"  
        persona="This is an explanation of this product. Please refer to this description for additional explanation, but it must be within 4 lines.."  
        native_request = {  
            "anthropic_version": "bedrock-2023-05-31",  
            "max_tokens": 512,  
            "temperature": 0.5,  
            "messages": [  
                {  
                    "role": "user",  
                    "content": [{"type": "text", "text": description+persona}],  
                }  
            ],  
        }  
  
        request = json.dumps(native_request)  
  
        response = client.invoke_model(modelId=model_id, body=request)  
  
        model_response = json.loads(response["body"].read())  
  
        return model_response["content"][0]["text"]  
  
    except (ClientError, Exception) as e:  
        print(f"ERROR: Can't invoke '{model_id}'. Reason: {e}")  
        return None  
  
  
  
def product_search_page():  
    st.title("상품 검색")  
    search_term = st.text_input("검색어를 입력하세요")  
  
  
    parsed_results = search_opensearch(search_term)  
    if parsed_results:  
  
        bedrock_desc=invoke_claude(parsed_results[0]['description']+''.join(parsed_results[0]['labels']))  
        container = st.container(border=True)  
        container.write(bedrock_desc)  
  
        if parsed_results and search_term:  
            for item in parsed_results:  
                with st.container():  
                    st.subheader(item['name'])  
  
                    st.write(item['description'])  
                    bucket_name,image_key=item['file_path'].split("/")  
                    image=get_image(bucket_name, image_key)  
                    st.image(image, caption='S3에서 가져온 이미지', use_container_width=True)  
  
                    labels="#"+'#'.join(item['labels'])  
                    st.write(labels)  
    else:  
        st.error("Error")