단, 두개의 AWS Lambda 함수로 Amazon OpenSearch, Amazon Bedrock 기반 이미지 검색 애플리케이션 구축하기

생성형 AI의 등장과 이와 더불어 관련 검색 기술이 빠르게 발전하면서, 기존 텍스트 매칭에서 벡터 기반 검색으로의 전환이 크게 주목받고 있습니다. 단순한 키워드 일치 방식은 이제 더 이상 충분하지 않을 수 있습니다. 이미지나 문장 등 비정형 데이터에서 의미적 유사성을 찾는 것이 점점 더 효과적으로 사용되어지고 이에 따라 점점 중요해지고 있기 때문입니다. 벡터 기반 검색은 이러한 요구를 만족시킬 수 있는 새로운 검색 방법입니다.
벡터 기반 검색은 단어 및 문장이나 이미지 같은 비정형 데이터를 검색할 때, 단순히 단어 및 쿼리가 일치하는지 여부를 넘어 쿼리와 검색 대상 간 콘텐츠의 의미를 이해하고 유사성을 찾아내는 강력한 방법입니다.

예를 들어, “고양이 사진”을 검색한다고 가정해봅시다. 기존의 텍스트 매칭 방식에서는 “고양이”라는 단어가 포함되어있는 이미지의 메타데이터(제목, 설명 등)와 매칭하여 찾을 수 있습니다. 하지만 벡터 기반 검색은 이보다 더 똑똑합니다. 예를 들어, “고양이가 창가에 앉아있는 사진”을 검색했을 때, 정확히 같은 설명이 없어도 쿼리와 저장되어 있는 이미지 자체의 의미를 이해하여 창가에 앉아 있는 고양이의 이미지를 찾아낼 수 있습니다. 이를 위해 벡터는 이미지나 문장의 의미를 숫자로 표현해, 이 숫자들 간의 거리나 유사성을 비교함으로써 비슷한 대상을 찾아내는 것입니다.

이렇게 복잡한 기술을 구현하는 것은 당연히 쉽지 않다고 생각할 수 있습니다. 그러나 Amazon OpenSearch Serverless와 Amazon Bedrock을 활용하면, 이러한 고급 검색 기능을 단 두 개의 AWS Lambda 함수로 간단하게 구현할 수 있습니다. 이 글에서는 복잡한 과정을 최대한 피하고, 어떻게 AWS를 기반으로 손쉽게 벡터 기반 이미지 검색 애플리케이션을 구축할 수 있는지 안내해 드립니다. 기술적 어려움을 최대한 덜고, 더 나은 검색 경험을 제공할 수 있는 길을 안내하겠습니다.

이미지 검색 애플리케이션 소개

<블로그에서 구현하려는 이미지 검색 애플리케이션 데모 영상>

해당 포스팅에서 구축하는 애플리케이션은 사용자가 업로드한 이미지나 입력한 텍스트를 기반으로 유사한 이미지를 검색할 수 있는 기능을 제공합니다. 이 애플리케이션은 벡터 기반 검색 방식을 활용하여 이미지 간의 의미적 유사성을 파악합니다. 영상의 애플리케이션은 Python의 Streamlit 프레임워크를 이용하여 로컬에서 동작하도록 구성되어있으며 필요에 따라 다양하게 커스텀 또는 자체 프론트 애플리케이션을 만들어 활용하실 수 있습니다.

구성에 필요한 모든 가이드를 제공하지만 필요한 이미지는 별도로 준비가 필요합니다. 저의 경우, 한국지능정보사회진흥원에서 운영하는 AI Hub의 이미지 샘플 데이터를 데모에 사용했습니다.

이미지 검색 애플리케이션 과정과 아키텍처

<블로그에서 구현 하려는 이미지 검색 애플리케이션 아키텍처>

위 아키텍처는 아래에 설명된 두개의 람다 함수를 기준으로 동작합니다. 미리 생성하여야 하는 리소스 구성 후, 두개의 람다를 이용하여 이미지 검색 기능에 대한 백엔드를 구현하게 됩니다.

🟩 EmbeddingImageAndSaveToOpensearch

Amazon S3에 객체(이미지)가 저장되면 람다 함수가 호출되고 이를 가져와 Base64 String으로 변환합니다.
Amazon Bedrock을 통해 Titan Multimodal Embdeddings G1 모델을 호출하여 이미지를 임베딩합니다.
임베딩을 통해 생성된 벡터를 Amazon OpenSearch Serverless에 저장합니다.

🟦 EmbeddingQueryAndQueryToOpensearch

프론트 애플리케이션에서 검색을 위해 이미지를 Amazon S3에 업로드합니다.
요청된 검색 유형이 ‘Image’인 경우, Amazon S3에서 가져와 Base64 String으로 변환합니다. (검색 유형이 ‘Text’인 경우 1,2를 생략합니다.)
검색 쿼리인 이미지 또는 텍스트를 Amazon Bedrock을 통해 Titan Multimodal Embdeddings G1 모델을 호출하여 임베딩합니다. (검색 유형이 ‘Text’인 경우, Amazon Translate를 통해 텍스트를 영어로 번역하여 임베딩합니다. 이는 옵션이며 필요하지 않은 경우, 제거될 수 있습니다.)
임베딩을 통해 생성된 벡터를 Amazon OpenSearch Serverless에 쿼리하여 요청된 벡터와 벡터 간 거리가 가까운 즉, 요청 쿼리와 의미적으로 유사한 결과를 전달받아 리턴합니다.

본 포스팅에서 이미지 검색 애플리케이션 구축을 위한 리전은 버지니아 북부(us-east-1)을 사용합니다.

사전 준비

단, 두개의 AWS Lambda 함수를 구성하기 전, 필요한 리소스를 사전에 준비합니다. 이를 기반으로 다양하게 변주하여 검색 애플리케이션을 고도화 할 수 있지만 가장 기본이 되고 쉽고 빠른 구성으로 가이드합니다.

Amazon Bedrock에서 모델 액세스: 해당 포스팅에서는 Amazon Bedrock을 통해 Titan Multimodal Embeddings G1 모델을 사용합니다. 이를 위해 모델 액세스 권한을 요청합니다.
1. Bedrock 콘솔에서 모델 액세스 권한 요청 합니다.
2. Enable all models 또는 Enable specific models를 클릭하여 Titan Multimodal Embeddings G1에 체크되어 있음을 확인하고 권한 요청을 제출합니다.
Amazon S3 버킷 생성 및 Amazon CloudFront 배포 설정: 업로드된 이미지를 저장할 Private S3 버킷을 생성하고 이를 Orgin으로 하는 CloudFront 배포를 구성하여 프론트 애플리케이션에서 이미지를 불러올 수 있게합니다. 만약, 별도의 CDN을 사용하거나 S3를 퍼블릭으로 사용하는 경우 구성하지 않아도 되지만 이는 권장되지 않습니다.
1. S3 콘솔에서 버킷을 생성합니다.
2. 고유한 버킷 이름을 입력하고 나머지 설정은 그대로 두고 버킷을 생성하고 버킷 명을 기록합니다.
3. CloudFront 콘솔에서 배포를 생성합니다.
  1. Origin domain은 생성한 버킷을 선택합니다.
  2. 원본 액세스 항목의Legacy access identities을 선택합니다.
  3. 새 OAI 생성을 클릭하여 CloudFront에서 S3의 객체를 가져오기 위한 Identity를 생성합니다.
  4. 버킷 정책의 예, 버킷 정책 업데이트를 선택하여 OAI에 대한 정책을 S3 버킷에 업데이트 되도록 합니다.
  5. 아래 기본 캐시 동작 - 뷰어 프로토콜 정책에서 HTTPS only를 선택하고 배포를 생성합니다.
  6. (선택사항) 배포에 대한 모니터링, 방화벽 구성 등이 필요한 경우, 웹 애플리케이션 방화벽을 활성화 합니다. 해당 포스팅에서는 비활성화 후, 진행합니다.
  7. 생성된 배포 도메인 이름을 기록합니다.

IAM Role 생성: Lambda 함수가 사용하는 Amazon Bedrock과 S3에 대하여 필요한 권한을 가질 수 있도록 역할을 생성합니다.

IAM 콘솔에서 정책을 생성합니다.

정책 편집기 유형을 JSON으로 선택하여 아래 정책을 복사하여 붙여 넣습니다. (Amazon Translate에 대한 정책의 경우, 텍스트 유형의 검색에서 영어가 아닌 언어로 검색하지 않는 경우에는 사용하지 않아도 됩니다.)

YOUR_BUCKET_NAME에는 생성한 버킷 명, YOUT_ACCOUNT_ID에는 AWS 계정 ID를 ‘-’없이 입력합니다.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "PolicyForS3",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject"
            ],
            "Resource": [
                "arn:aws:s3:::YOUR_BUCKET_NAME/*"
            ]
        },
        {
            "Sid": "PolicyForBedrock",
            "Effect": "Allow",
            "Action": [
                "bedrock:InvokeModel"
            ],
            "Resource": [
                "arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-image-v1"
            ]
        },
        {
            "Sid": "PolicyForAOSS",
            "Effect": "Allow",
            "Action": [
                "aoss:APIAccessAll"
            ],
            "Resource": "arn:aws:aoss:us-east-1:YOUT_ACCOUNT_ID:collection/*"
        },
        {
            "Sid": "PolicyForTranslate",
            "Effect": "Allow",
            "Action": [
                "translate:TranslateText",
                "comprehend:DetectDominantLanguage"
            ],
            "Resource": [
                "*"
            ]
        }
    ]
}

다음으로 넘어가 정책의 이름을 입력하고 생성합니다.
이제, AWS Lambda 함수를 위한 IAM 콘솔에서 역할을 생성합니다.
신뢰할 수 있는 엔터티 유형의 AWS 서비스를 선택하고 사용 사례에서 Lambda를 선택합니다.
앞에서 생성한 정책의 이름을 검색하여 선택합니다.
역할 이름(e.g. imageSearchLambdaRole)을 입력하고 역할을 생성합니다.

Amazon OpenSearch Serverless Collection 생성: 이미지 검색에 사용할 인덱스를 저장할 OpenSearch Serverless Collection을 생성합니다.
1. Opensearch Service 콘솔에서 서버리스: 컬렉션을 생성합니다.
  1. 수집 이름(e.g. image-search-collection)을 입력하고 수집 유형은 벡터 검색을 선택합니다.
  2. 보안은 손쉬운 생성으로 선택하고 검토 후, 생성을 완료합니다. (Amazon OpenSearch Serverless는 세밀한 네트워크 및 데이터 접근 제어를 제공합니다. 프로덕션으로 안전한 사용을 원하는 경우, 추가로 구성하여주세요.)
  3. 손쉬운 생성으로 생성하게된다면 네트워크 정책, 데이터 액세스 정책 각각에 easy-YOUR_COLLECTION_NAME으로 기본 정책이 생성됩니다.
  4. 생성한 수집의데이터 액세스부분의 연결된 정책을 클릭하고 편집을 시작합니다.
2. 생성한 수집의데이터 액세스부분의 연결된 정책을 클릭하고 편집을 시작합니다.
  1. 규칙의 보안 주체 선택 → 보안 주체 추가 → IAM 사용자 및 역할을 자체로 선택합니다.
  2. 검색창의 속성 → 역할을 선택하고 앞에서 생성한 AWS Lambda의 역할 이름을 입력하여 검색 후, 선택하고 저장합니다.
  3. (선택사항 – 네트워크 액세스를 퍼블릭이 아닌 AWS Lambda 접근 만을 허용할 경우) 생성된 네트워크 액세스 정책의네트워크 액세스 정책 편집에서 다음에서 수집에 액세스 - Private 선택 후, AWS service private access를 선택하고 Service = lambda.amazonaws.com을 입력하고 추가합니다. (이 구성은 OpenSearch 수집에 AWS Lambda의 네트워크 접근만 허용함을 의미합니다. 추가로 VPC 등에서의 제한된 네트워크 접근이 필요한 경우, 추가하여야 합니다.)
3. 앞에서 생성한 수집(컬렉션)으로 돌아와 생성이 완료되기까지 대기합니다. (약, 5분 소요)
4. 생성이 완료되면 인덱스탭이 생깁니다. 이를 선택하고 벡터 인덱스 생성을 클릭합니다.
  1. 저장되어질 이미지의 특성 또는 목적에 맞는 벡터 인덱스 이름(e.g. product-index)을 입력합니다.
  2. 벡터 필드 추가를 클릭하고 벡터 필드 이름: vector, Engine: faiss, 치수: 1024, 거리 지표: 유클리드로 구성 후,확인을 눌러 추가합니다.
  3. 아래 메타 데이터 관리에서 두 개의 메타데이터를 추가하고 벡터 인덱스를 생성합니다. (검색 대상 및 결과를 받아오는 인덱스입니다. 추가 메타 데이터가 필요한 경우, 추가하여 데이터를 저장하고 결과를 받아오도록 할 수 있습니다.)
    - 매핑 필드: s3_key, 데이터 유형: 문자열, 필터링 가능: False
    - 매핑 필드: s3_bucket, 데이터 유형: 문자열, 필터링 가능: True
5. 끝으로 수집(컬렉션)의 개요탭에서 OpenSearch 호스트 주소를 기록합니다. 이는 OpenSearch 엔드포인트값에서 앞의 ‘https://’ 부분을 제외한 나머지입니다. (e.g. xxxxxxxxxxxxxxxxx.us-east-1.aoss.amazonaws.com)
AWS Lambda를 위한 계층 준비: Amazon OpenSearch Service와의 상호작용을 위한 라이브러리인 opensearch-py와 이때의 AWS API 요청에 대한 인증을 처리하기 위한 라이브러리인 requests-aws4auth의 AWS Lambda에서의 사용을 위해 계층을 생성합니다.
1. Python 3.8 이상 버전이 설치되어있는 로컬 환경을 준비합니다.
2. 터미널에서 아래 주석을 제외한 명령어를 차례로 입력하여 계층을 위한 패키지 압축 파일을 생성합니다.
  - ```
  # 1. 계층을 위한 디렉토리 생성
  mkdir -p opensearch-layer/python
  
  # 2. Python 가상환경 생성 및 활성화
  python3 -m venv opensearch-layer/venv
  source opensearch-layer/venv/bin/activate
  
  # 3. 필요한 라이브러리 설치
  pip install opensearch-py requests-aws4auth
   
  # 4. site-packages 폴더를 python 디렉토리로 복사
  cp -r opensearch-layer/venv/lib/python3.*/site-packages/* opensearch-layer/python/
  
  # 5. 계층 패키징
  cd opensearch-layer
  zip -r opensearch-layer.zip python
  
  # 6. 압축 파일 생성 후, 가상환경 비활성화
  deactivate
```
- 명령어를 실행한 폴더 기준으로 opensearch-layer 폴더 안의 opensearch-layer.zip 파일이 정상적으로 생성 되었는지 확인 합니다.
3. Lambda 콘솔에서 계층을 생성합니다.
4. 계층 이름(e.g. opensearch_layer)을 입력하고 .zip 파일 업로드 선택 후, 업로드 버튼을 클릭하여 앞에서 생성한 압축 파일을 선택합니다.
5. 호환 런타임에서 Python 3.12를 추가하고 생성을 클릭 합니다.

이제, 모든 사전 준비가 끝났습니다. 위에서 생성한 리소스를 기반으로 동작하는 단, 두개의 AWS Lambda 함수를 구성하여 이미지 검색 기능 구현을 완료하겠습니다.

🟩 EmbeddingImageAndSaveToOpensearch 함수 구성

함수의 이름은 자유롭게 변경 가능합니다. 필요에 따라 OpenSearch 인덱스 별 함수 생성 또는 요청 파라미터로 인덱스를 받는 형태로 코드를 수정할 수도 있습니다.

EmbeddingImageAndSaveToOpensearch 함수는 Amazon S3 버킷에 이미지가 업로드되면 자동으로 호출되어 임베딩하고 OpenSearch에 저장하는 로직을 수행합니다.

Lambda 콘솔에서 함수를 생성합니다.
함수 이름(EmbeddingImageAndSaveToOpensearch)을 입력하고 런타임은 Python 3.12으로 설정합니다.
기본 실행 역할 변경 토글을 활성화 하고 기존 역할 사용을 선택하여 사전 준비에서 만든 Lambda 역할(e.g. imageSearchLambdaRole)을 선택 후, 함수 생성을 클릭 합니다.
함수 개요의 다이어그램 왼쪽의 트리거 추가를 선택합니다.
1. 소스 선택에서 S3를 선택하고 버킷은 사전 준비에서 생성한 버킷 이름을 입력하여 선택합니다.
2. 이벤트 유형은 모든 객체 생성 이벤트를 선택합니다.
3. 접두사에는 사전 준비에서 생성한 벡터 인덱스 명/ 형태로 입력합니다. 예를 들어, ‘product-index’라는 이름의 벡터 인덱스를 만들었다면 product-index/으로 입력합니다.
4. 재귀 호출 부분의 안내 사항을 확인하고 체크한 후, 추가를 클릭 합니다.

코드 편집기에서 아래 코드를 복사하여 붙여넣고 저장 후, Deploy를 클릭하여 함수를 배포합니다.

import os
import boto3
import json
import base64
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth

def lambda_handler(event, context):
    opensearch_region = os.getenv('OPENSEARCH_REGION')
    opensearch_host = os.getenv('OPENSEARCH_HOST')
    opensearch_index = os.getenv('OPENSEARCH_INDEX')
    
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']
    
    s3 = boto3.client('s3')
    response = s3.get_object(Bucket=bucket, Key=key)
    image_content = response['Body'].read()
    
    image_base64 = base64.b64encode(image_content).decode('utf-8')
    
    bedrock = boto3.client('bedrock-runtime')
    response = bedrock.invoke_model(
        modelId="amazon.titan-embed-image-v1",
        contentType="application/json",
        accept="application/json",
        body=json.dumps({
            "inputImage": image_base64
        })
    )
    
    embedding = json.loads(response['body'].read())['embedding']
    
    service = 'aoss'
    region = opensearch_region
    credentials = boto3.Session().get_credentials()
    awsauth = AWS4Auth(credentials.access_key, credentials.secret_key,
                       region, service, session_token=credentials.token)
    
    opensearch = OpenSearch(
        hosts = [{'host': opensearch_host, 'port': 443}],
        http_auth = awsauth,
        use_ssl = True,
        verify_certs = True,
        connection_class = RequestsHttpConnection
    )
    
    document = {
        'vector': embedding,
        's3_key': f"{key}",
        's3_bucket': f"{bucket}"
    }
    
    opensearch.index(index=opensearch_index, body=document)
    
    return f"s3://{bucket}/{key} has been successfully saved in the index"

아래에 계층탭 우측의 Add a layer를 선택하여 사전 준비에서 만든 계층을 추가합니다.
1. 계층 소스는 사용자 지정 계층을 선택합니다.
2. 사전 준비에서 만든 계층을 선택하고 버전은 가장 최신을 선택하고 추가를 클릭 합니다.
함수의 구성탭으로 이동합니다.
1. 좌측 일반 구성 탭에서 편집을 클릭하여 제한 시간을 1분으로 설정 후, 저장합니다.
2. 좌측 환경 변수 탭에서 편집을 클릭하여 아래 환경 변수를 추가합니다.
  - 키: OPENSEARCH_REGION, 값: us-east-1
  - 키: OPENSEARCH_HOST, 값: 사전 준비에서 생성한 호스트 주소(e.g. xxxxxxxxxxxxxxxxx.us-east-1.aoss.amazonaws.com)
  - 키: OPENSEARCH_INDEX, 값: 사전 준비에서 생성한 인덱스 명(e.g. product-index)

이제, 사전 준비에서 생성한 S3 버킷에서 폴더 만들기를 선택하여 OpenSearch 인덱스 명으로 폴더를 생성하고 검색 대상 이미지를 업로드하면 각 이미지별로 위에서 생성한 함수가 실행되어 임베딩 후, OpenSearch 인덱스에 저장하여 검색 준비가 완료됩니다!

🟦 EmbeddingQueryAndQueryToOpensearch 함수 구성

함수의 이름은 자유롭게 변경 가능합니다. 필요에 따라 OpenSearch 인덱스 별 함수 생성 또는 요청 파라미터로 인덱스를 받는 형태로 코드를 수정할 수도 있습니다.

EmbeddingQueryAndQueryToOpensearch 함수는 프론트 애플리케이션으로부터 받은 쿼리 요청을 처리합니다. 검색 유형은 ‘Image’, ‘Text’ 두가지가 있으며 모두 임베딩을 하여 OpenSearch에 쿼리하는 것은 동일합니다. ‘Text’의 경우, Amazon Translate을 통해 영어로 자동 번역하며 이는 옵션으로 필요하지 않은 경우, 해당 부분 코드를 제거해주세요.

Lambda 콘솔에서 함수를 생성합니다.
함수 이름(EmbeddingQueryAndQueryToOpensearch)을 입력하고 런타임은 Python 3.12으로 설정합니다.
기본 실행 역할 변경 토글을 활성화 하고 기존 역할 사용을 선택하여 사전 준비에서 만든 Lambda 역할(e.g. imageSearchLambdaRole)을 선택 후, 함수 생성을 클릭합니다.

코드 편집기에서 아래 코드를 복사하여 붙여넣고 저장 후, Deploy를 클릭하여 함수를 배포합니다.

import os
import boto3
import json
import base64
from opensearchpy import OpenSearch, RequestsHttpConnection
from requests_aws4auth import AWS4Auth

def lambda_handler(event, context):
    opensearch_region = os.getenv('OPENSEARCH_REGION')
    opensearch_host = os.getenv('OPENSEARCH_HOST')
    opensearch_index = os.getenv('OPENSEARCH_INDEX')
    
    query_type = event['type']
    query_text = event.get('text', '')
    query_image_bucket = event.get('s3_bucket', '')
    query_image_key = event.get('s3_key', '')
    
    bedrock = boto3.client('bedrock-runtime')
    
    if query_type == 'text':
        # 영어 이외의 언어 텍스트 검색이 필요하지 않은 경우, 아래 translated_text 까지의 코드는 생략 가능합니다.
        translate = boto3.client('translate')
        response = translate.translate_text(
            Text=query_text,
            SourceLanguageCode='auto',
            TargetLanguageCode='en'
        )
        translated_text = response['TranslatedText']
        
        request_body = json.dumps({
            "inputText": translated_text # 한국어 텍스트 검색이 필요하지 않은 경우, query_text 로 대체
        })
    elif query_type == 'image':
        print(query_image_bucket)
        print(query_image_key)
        s3 = boto3.client('s3')
        response = s3.get_object(Bucket=query_image_bucket, Key=query_image_key)
        image_content = response['Body'].read()
        image_base64 = base64.b64encode(image_content).decode('utf-8')
        request_body = json.dumps({
            "inputImage": image_base64
        })
    else:
        return {
            'statusCode': 400,
            'body': json.dumps('Invalid query type')
        }
    
    response = bedrock.invoke_model(
        modelId="amazon.titan-embed-image-v1",
        contentType="application/json",
        accept="application/json",
        body=request_body
    )
    
    query_vector = json.loads(response['body'].read())['embedding']
    
    region = opensearch_region
    service = 'aoss'
    credentials = boto3.Session().get_credentials()
    awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)

    opensearch = OpenSearch(
        hosts = [{'host': opensearch_host, 'port': 443}],
        http_auth = awsauth,
        use_ssl = True,
        verify_certs = True,
        connection_class = RequestsHttpConnection
    )

    query = {
        "size": 9,
        "query": {
            "knn": {
                "vector": {
                    "vector": query_vector,
                    "k": 9
                }
            }
        }
    }

    response = opensearch.search(index=opensearch_index, body=query)

    results = [
        {
            'score': hit['_score'],
            's3_key': hit['_source']['s3_key'],
            's3_bucket': hit['_source']['s3_bucket']
        }
        for hit in response['hits']['hits']
    ]

    return results

아래에 계층탭 우측의 Add a layer를 선택하여 사전 준비에서 만든 계층을 추가합니다.
1. 계층 소스는 사용자 지정 계층을 선택합니다.
2. 사전 준비에서 만든 계층을 선택하고 버전은 가장 최신을 선택하고 추가를 클릭 합니다.
함수의 구성탭으로 이동합니다.
1. 좌측 일반 구성 탭에서 편집을 클릭하여 제한 시간을 1분으로 설정 후, 저장합니다.
2. 좌측 환경 변수 탭에서 편집을 클릭하여 아래 환경 변수를 추가합니다.
  - 키: OPENSEARCH_REGION, 값: us-east-1
  - 키: OPENSEARCH_HOST, 값: 사전 준비에서 생성한 호스트 주소(e.g. xxxxxxxxxxxxxxxxx.us-east-1.aoss.amazonaws.com)
  - 키: OPENSEARCH_INDEX, 값: 사전 준비에서 생성한 인덱스 명(e.g. product-index)
상단 함수 개요의 우측 부분의 함수 ARN을 기록합니다.

이제, 프론트 애플리케이션으로부터 요청 받아 쿼리를 처리하는 AWS Lambda 함수 생성이 완료되었습니다!

선택사항. Streamlit 기반 프론트 애플리케이션으로 테스트하기

마지막으로 🟦 EmbeddingQueryAndQueryToOpensearch 의 동작 확인을 위해 Streamlit을 사용하여 간단한 프론트 애플리케이션을 만듭니다. Streamlit은 간단한 파이썬 코드 작성으로 프론트 웹 애플리케이션을 만들 수 있는 프레임워크입니다. 물론 다른 라이브러리 및 프론트 애플리케이션을 통해 원하는 뷰 및 기능을 갖춘 애플리케이션을 만들 수도 있습니다!

로컬 환경에서 Python AWS SDK인 boto3를 사용하기 위해 설치와 구성을 하여야 합니다. 프론트 애플리케이션에는 Amazon S3에 쿼리 이미지를 업로드하기 위한 ‘Amazon S3: PutObject’, 🟦 EmbeddingQueryAndQueryToOpensearch 함수 실행을 위한 ‘AWS Lambda: InvokeFunction’ 에 대한 권한이 필요합니다.

IAM 콘솔에서 정책 생성
1. 정책 편집기에서 JSON 토글을 클릭하고 아래 정책을 붙여넣습니다.
  - YOUR_BUCKET_NAME에는 생성한 버킷 명, YOUR_EmbeddingQueryAndQueryToOpensearch_ARN에는 EmbeddingQueryAndQueryToOpensearch 함수의 ARN을 입력합니다.
  - ```
  {
      "Version": "2012-10-17",
      "Statement": [
          {
              "Sid": "PolicyForS3",
              "Effect": "Allow",
              "Action": [
                  "s3:PutObject"
              ],
              "Resource": [
                  "arn:aws:s3:::YOUR_BUCKET_NAME/input-images/*"
              ]
          },
          {
              "Sid": "PolicyForLambda",
              "Effect": "Allow",
              "Action": [
                  "lambda:InvokeFunction"
              ],
              "Resource": ["YOUR_EmbeddingQueryAndQueryToOpensearch_ARN"]
          }
      ]
  }
```
- 다음을 클릭 후, 이름(e.g. policyForImageSearchApp)을 입력하고 정책을 생성합니다.
2. IAM 콘솔에서 사용자 생성합니다. 이미 있는 경우, 해당 과정을 스킵하고 기존 사용자에 위에서 만든 정책을 추가합니다.
  1. 임의의 사용자 이름 입력 후, 직접 정책 연결을 선택하고 위에서 만든 정책을 추가합니다.
  2. 보안 자격 증명탭에세 액세스 키를 생성하고 해당 가이드를 따라 boto3 설치 및 로컬 환경에서 자격증명 구성을 진행합니다.
로컬 환경에 Streamlit을 설치합니다.

Streamlit 및 boto3 설치가 되어 있는 환경에서 아래 코드를 파이썬 파일(e.g. app.py)을 만들어 붙여넣습니다.

YOUR_CF_DISTRIBUTION_DOMAIN_NAME에는 사전 준비에서 만든 CloudFront 배포 도메인 이름, YOUR_BUCKET_NAME에는 S3 버킷 명, YOUR_EmbeddingQueryAndQueryToOpensearch_ARN에는 EmbeddingQueryAndQueryToOpensearch 함수의 ARN을 입력합니다.

import json
import boto3
import streamlit as st
from PIL import Image
import io

# AWS Setting
s3 = boto3.client('s3')
lambda_client = boto3.client('lambda')
cloudfront_url = 'https://YOUR_CF_DISTRIBUTION_DOMAIN_NAME/'
BUCKET_NAME = 'YOUR_BUCKET_NAME'
LAMBDA_ARN = 'YOUR_EmbeddingQueryAndQueryToOpensearch_ARN'

def upload_to_s3(file):
    try:
        s3_key = 'input-images/' + file.name
        file_contents = file.getvalue()
        s3.upload_fileobj(io.BytesIO(file_contents), BUCKET_NAME, s3_key)
        return s3_key
    except Exception as e:
        st.error(f"Error during S3 upload: {e}")
        return None


def invoke_lambda(payload):
    try:
        response = lambda_client.invoke(
            FunctionName=LAMBDA_ARN,
            InvocationType='RequestResponse',
            Payload=json.dumps(payload)
        )
        return json.loads(response['Payload'].read())
    except Exception as e:
        st.error(f"Error Calling a Lambda Function: {e}")
        return None


def search_images(query, search_type):
    if search_type == "Image":
        payload = {
            "type": "image",
            "s3_bucket": BUCKET_NAME,
            "s3_key": query
        }
    else:  # Text search
        payload = {
            "type": "text",
            "text": query
        }

    return invoke_lambda(payload)

def main():
    st.title("Image Search Application")
    search_type = st.radio("Select search type:", ("Image", "Text"))
    if search_type == "Image":
        uploaded_file = st.file_uploader("Upload an image to search for.", type=["jpg", "png", "jpeg"])
        if uploaded_file is not None:
            image = Image.open(uploaded_file)
            st.image(image, caption='Upload Image', width=300)
            if st.button("Search"):
                s3_key = upload_to_s3(uploaded_file)
                if s3_key:
                    st.success(f"Image upload successfully.")
                    results = search_images(s3_key, search_type)
                    if results:
                        display_results(results)
    else:
        search_query = st.text_input("Type search text")
        if st.button("Search"):
            results = search_images(search_query,search_type)
            if results:
                display_results(results)

def display_results(results):
    for i in range(0, len(results), 3):
        cols = st.columns(3)
        for j in range(3):
            if i + j < len(results):
                item = results[i + j]
                with cols[j]:
                    image_url = cloudfront_url + item['s3_key']
                    st.write(f"Score: {item['score']}")
                    st.image(image_url, caption=item['s3_key'], use_column_width=True)
        st.write("---")

if __name__ == "__main__":
    main()

터미널에서 streamlit run app.py 을 입력하여 애플리케이션을 실행합니다.
실행되는 앱에서 이미지 업로드 또는 텍스트를 입력하여 검색을 테스트합니다.<구현한 이미지 검색 애플리케이션 검색 예시>

실습 리소스 정리

본 포스팅에서 구현해본 이미지 검색 애플리케이션은 그대로 사용할 수 있지만 인덱스 구성(메타데이터 등) 변경 또는 프론트 애플리케이션의 변경 등 다양하게 변주하여 실제 프로덕션에서 활용할 수 있습니다.

실습 후, 사용하지 않는 경우 비용이 발생할 수 있으므로 아래 리스트를 확인하시어 생성하신 리소스를 제거해주시기 바랍니다.

Amazon CloudFront 배포
Amazon S3 버킷
Amazon OpenSearch Serverless 컬렉션
선택사항, AWS Lambda 함수: 호출되지 않는 경우, 비용이 발생하지 않습니다.

마무리

이번 글에서 소개한 벡터 기반 이미지 검색 애플리케이션은 Amazon OpenSearch Serverless와 Amazon Bedrock을 활용해 복잡한 기술적 난관을 단, 두 개의 Lambda 함수로 해결할 수 있음을 보여줍니다. 실습을 진행하지 않았더라도 코드를 확인 하시어 어떤 방식으로 이미지 검색 애플리케이션을 간단하게 구현할 수 있는지 확인하실 수 있습니다.

이미지 기반 검색 애플리케이션은 다양한 분야에서 큰 가치를 발휘할 수 있습니다. 예를 들어, 전자 상거래(e-commerce) 에서는 고객이 상품의 이미지를 업로드하면 유사한 제품을 추천할 수 있고, 제조업에서는 부품이나 기계의 이미지를 통해 비슷한 구성품을 쉽게 찾아낼 수 있습니다. 또한, 미디어 및 콘텐츠 관리 분야에서는 이미지나 비디오 라이브러리에서 유사한 콘텐츠를 효율적으로 검색할 수 있습니다.

AWS의 서비스들을 활용하면 이러한 벡터 기반 고급 검색 애플리케이션을 손쉽게 구축할 수 있습니다. 이에 필요한 복잡한 벡터 검색 엔진의 인프라 관리나 임베딩 모델 학습에 얽매이지 않고, 제공되는 서비스를 활용하여 아이디어를 실현하는 데에만 집중하실 수 있습니다. 이 글이 여러분이 손쉽게 강력한 이미지 검색 솔루션을 구축하고, 이를 다양한 비즈니스에 응용하는 데 도움이 되길 바랍니다.

AWS 기술 블로그