HAQM Web Services 한국 블로그
HAQM Bedrock Data Automation 정식 출시 – 멀티모달 콘텐츠로 부터 인사이트 도출하기
대부분의 애플리케이션은 다양한 형식을 통해 가용 콘텐츠와 상호 작용해야 합니다. 이러한 애플리케이션 중에는 보험 청구 및 의료비 청구서와 같은 복잡한 문서를 처리하는 애플리케이션도 있습니다. 모바일 앱은 사용자가 생성한 미디어를 분석해야 합니다. 기업들은 문서, 이미지, 오디오, 비디오 파일 등의 디지털 자산을 기반으로 의미 체계 인덱스를 구축해야 합니다. 하지만 비정형 멀티모달 콘텐츠에서 인사이트를 도출하는 기능은 설정하기가 쉽지 않습니다. 다양한 데이터 형식에 대한 처리 파이프라인을 구현하고, 필요한 정보를 얻기 위해 여러 단계를 거쳐야 하기 때문입니다. 이는 일반적으로 프로덕션 환경에 여러 모델을 배포해야 하고, 그러한 각 모델마다 (미세 조정 및 프롬프트 엔지니어링을 통한) 비용 최적화, 안전 조치(예: 할루시네이션 방지), 대상 애플리케이션과의 통합(데이터 형식 포함), 모델 업데이트를 수행해야 함을 의미합니다.
이 프로세스를 간소화하기 위해 AWS re:Invent 기간 중에, 문서, 이미지, 오디오, 동영상 등 비정형 멀티모달 콘텐츠에서 귀중한 인사이트를 간편하게 생성할 수 있는 HAQM Bedrock 기능인 HAQM Bedrock Data Automation을 평가판으로 발표했습니다. Bedrock Data Automation을 사용하면 지능형 문서 처리, 미디어 분석 및 기타 멀티모달 데이터 중심 자동화 솔루션을 구축하는 데 소요되는 개발 시간과 작업량을 줄일 수 있습니다.
Bedrock Data Automation을 독립형 기능으로 사용하거나 HAQM Bedrock Knowledge Bases의 파서로 사용하여, 멀티모달 콘텐츠에서 인사이트를 인덱싱하고 검색 증강 생성(RAG)을 위한 보다 관련성 높은 답변을 제공할 수 있습니다.
오늘 Bedrock Data Automation이 교차 리전 추론 엔드포인트 지원 기능과 함께 정식 출시되어 더 많은 AWS 리전에서 제공할 수 있게 되었으며, 여러 로케이션의 컴퓨팅 리소스도 원활하게 활용할 수 있게 되었습니다. 평가판 단계에서 보내주신 피드백을 바탕으로 정확도를 개선하고 이미지 및 비디오의 로고 인식을 위한 지원 기능도 추가했습니다.
이 기능이 실제로 어떻게 작동하는지 살펴보겠습니다.
교차 리전 추론 엔드포인트와 함께 HAQM Bedrock Data Automation 사용
Bedrock Data Automation 평가판에 대한 블로그 게시물에서는 HAQM Bedrock 콘솔의 시각적 데모를 사용하여 문서 및 비디오에서 정보를 추출하는 방법을 보여줍니다. 콘솔 데모 환경을 통해 이 기능의 작동 원리와 맞춤화하기 위해 수행할 수 있는 작업을 알아보시기 바랍니다. 이 게시물에서는 콘솔에서 수행하는 몇 단계를 설명한 후 코드 샘플을 소개하면서 Bedrock Data Automation이 애플리케이션에서 어떻게 작동하는지 자세히 살펴봅니다.
이제 HAQM Bedrock 콘솔의 데이터 자동화 섹션에 처음 액세스하면 교차 리전 지원을 활성화할지 여부를 확인하는 메시지가 나타납니다. 예를 들면 다음과 같습니다.
API 관점에서 보면, 이제 사용할 데이터 자동화 프로필을 지정하기 위한 추가 파라미터(dataAutomationProfileArn)가 InvokeDataAutomationAsync 작업에 필요합니다. 이 파라미터의 값은 리전 및 AWS 계정 ID에 따라 달라집니다.
arn:aws:bedrock:<REGION>:<ACCOUNT_ID>:data-automation-profile/us.data-automation-v1
또한 프로젝트 HAQM 리소스 이름(ARN)이 포함되어 있다는 점을 더 잘 반영하기 위해 dataAutomationArn 파라미터의 이름이 dataAutomationProjectArn으로 변경되었습니다. 이제 Bedrock Data Automation을 간접적으로 호출할 때 사용할 프로젝트 또는 블루프린트를 지정해야 합니다. 블루프린트를 전달하면 사용자 지정 출력이 생성됩니다. 표준 기본 출력을 계속 가져오려면 arn:aws:bedrock:<REGION>:aws:data-automation-project/public-default를 사용하도록 DataAutomationProjectArn 파라미터를 구성합니다.
이름에서 알 수 있듯이 InvokeDataAutomationAsync 작업은 비동기식입니다. 입력 및 출력 구성을 전달하고 결과가 준비되면, 출력 구성에 지정된 대로 HAQM Simple Storage Service(HAQM S3) 버킷에 결과가 기록됩니다. notificationConfiguration 파라미터를 사용하여 Bedrock Data Automation으로부터 HAQM EventBridge 알림을 받을 수 있습니다.
Bedrock Data Automation에서는 다음 두 가지 방법으로 출력을 구성할 수 있습니다.
- 표준 출력은 문서 의미 체계, 비디오 챕터 요약, 오디오 트랜스크립트 등의 데이터 유형과 관련하여 사전 정의된 인사이트를 제공합니다. 표준 출력을 사용할 경우 몇 단계만 거치면 원하는 인사이트를 설정할 수 있습니다.
- 사용자 지정 출력은 블루프린트를 사용하여 추출 요구 사항을 지정함으로써 보다 맞춤화된 인사이트를 얻을 수 있게 해줍니다.
새 기능이 어떻게 작동하는지 직접 볼 수 있도록, 프로젝트를 만들고 표준 출력 설정을 사용자 지정해 보겠습니다. 문서의 경우 마크다운 대신 일반 텍스트를 선택합니다. 참고로 Bedrock Data Automation API를 사용하여 이러한 구성 단계를 자동화할 수 있습니다.
비디오의 경우 전체 오디오 트랜스크립션과 전체 비디오의 요약이 필요합니다. 또한 각 챕터의 요약을 요청합니다.
블루프린트를 구성하려면 HAQM Bedrock 콘솔 탐색 창의 데이터 자동화 섹션에서 사용자 지정 출력 설정을 선택합니다. 여기에서 US-Driver-License 샘플 블루프린트를 검색하겠습니다. 다른 샘플 블루프린트에서 더 많은 예와 아이디어를 찾아볼 수 있습니다.
샘플 블루프린트는 편집할 수 없으므로 작업 메뉴를 사용하여 블루프린트를 복제하고 프로젝트에 추가합니다. 여기에서 블루프린트를 수정하고, 생성형 AI를 사용하여 필요한 형식으로 데이터를 추출하거나 계산할 수 있는 사용자 지정 필드를 추가하는 방법으로, 추출할 데이터를 미세 조정할 수 있습니다.
미국 운전 면허증 이미지를 S3 버킷에 업로드하겠습니다. 그런 다음 AWS SDK for Python(Boto3)을 통해 Bedrock Data Automation을 사용하여 이미지에서 텍스트 정보를 추출하는 이 샘플 Python 스크립트를 사용하겠습니다.
import json
import sys
import time
import boto3
DEBUG = False
AWS_REGION = '<REGION>'
BUCKET_NAME = '<BUCKET>'
INPUT_PATH = 'BDA/Input'
OUTPUT_PATH = 'BDA/Output'
PROJECT_ID = '<PROJECT_ID>'
BLUEPRINT_NAME = 'US-Driver-License-demo'
# 표시할 필드
BLUEPRINT_FIELDS = [
'NAME_DETAILS/FIRST_NAME',
'NAME_DETAILS/MIDDLE_NAME',
'NAME_DETAILS/LAST_NAME',
'DATE_OF_BIRTH',
'DATE_OF_ISSUE',
'EXPIRATION_DATE'
]
# AWS SDK for Python(Boto3) 클라이언트
bda = boto3.client('bedrock-data-automation-runtime', region_name=AWS_REGION)
s3 = boto3.client('s3', region_name=AWS_REGION)
sts = boto3.client('sts')
def log(data):
if DEBUG:
if type(data) is dict:
text = json.dumps(data, indent=4)
else:
text = str(data)
print(text)
def get_aws_account_id() -> str:
return sts.get_caller_identity().get('Account')
def get_json_object_from_s3_uri(s3_uri) -> dict:
s3_uri_split = s3_uri.split('/')
bucket = s3_uri_split[2]
key = '/'.join(s3_uri_split[3:])
object_content = s3.get_object(Bucket=bucket, Key=key)['Body'].read()
return json.loads(object_content)
def invoke_data_automation(input_s3_uri, output_s3_uri, data_automation_arn, aws_account_id) -> dict:
params = {
'inputConfiguration': {
's3Uri': input_s3_uri
},
'outputConfiguration': {
's3Uri': output_s3_uri
},
'dataAutomationConfiguration': {
'dataAutomationProjectArn': data_automation_arn
},
'dataAutomationProfileArn': f"arn:aws:bedrock:{AWS_REGION}:{aws_account_id}:data-automation-profile/us.data-automation-v1"
}
response = bda.invoke_data_automation_async(**params)
log(response)
return response
def wait_for_data_automation_to_complete(invocation_arn, loop_time_in_seconds=1) -> dict:
while True:
response = bda.get_data_automation_status(
invocationArn=invocation_arn
)
status = response['status']
if status not in ['Created', 'InProgress']:
print(f" {status}")
return response
print(".", end='', flush=True)
time.sleep(loop_time_in_seconds)
def print_document_results(standard_output_result):
print(f"Number of pages: {standard_output_result['metadata']['number_of_pages']}")
for page in standard_output_result['pages']:
print(f"- Page {page['page_index']}")
if 'text' in page['representation']:
print(f"{page['representation']['text']}")
if 'markdown' in page['representation']:
print(f"{page['representation']['markdown']}")
def print_video_results(standard_output_result):
print(f"Duration: {standard_output_result['metadata']['duration_millis']} ms")
print(f"Summary: {standard_output_result['video']['summary']}")
statistics = standard_output_result['statistics']
print("Statistics:")
print(f"- Speaket count: {statistics['speaker_count']}")
print(f"- Chapter count: {statistics['chapter_count']}")
print(f"- Shot count: {statistics['shot_count']}")
for chapter in standard_output_result['chapters']:
print(f"Chapter {chapter['chapter_index']} {chapter['start_timecode_smpte']}-{chapter['end_timecode_smpte']} ({chapter['duration_millis']} ms)")
if 'summary' in chapter:
print(f"- Chapter summary: {chapter['summary']}")
def print_custom_results(custom_output_result):
matched_blueprint_name = custom_output_result['matched_blueprint']['name']
log(custom_output_result)
print('\n- Custom output')
print(f"Matched blueprint: {matched_blueprint_name} Confidence: {custom_output_result['matched_blueprint']['confidence']}")
print(f"Document class: {custom_output_result['document_class']['type']}")
if matched_blueprint_name == BLUEPRINT_NAME:
print('\n- Fields')
for field_with_group in BLUEPRINT_FIELDS:
print_field(field_with_group, custom_output_result)
def print_results(job_metadata_s3_uri) -> None:
job_metadata = get_json_object_from_s3_uri(job_metadata_s3_uri)
log(job_metadata)
for segment in job_metadata['output_metadata']:
asset_id = segment['asset_id']
print(f'\nAsset ID: {asset_id}')
for segment_metadata in segment['segment_metadata']:
# 표준 출력
standard_output_path = segment_metadata['standard_output_path']
standard_output_result = get_json_object_from_s3_uri(standard_output_path)
log(standard_output_result)
print('\n- Standard output')
semantic_modality = standard_output_result['metadata']['semantic_modality']
print(f"Semantic modality: {semantic_modality}")
match semantic_modality:
case 'DOCUMENT':
print_document_results(standard_output_result)
case 'VIDEO':
print_video_results(standard_output_result)
# 사용자 지정 출력
if 'custom_output_status' in segment_metadata and segment_metadata['custom_output_status'] == 'MATCH':
custom_output_path = segment_metadata['custom_output_path']
custom_output_result = get_json_object_from_s3_uri(custom_output_path)
print_custom_results(custom_output_result)
def print_field(field_with_group, custom_output_result) -> None:
inference_result = custom_output_result['inference_result']
explainability_info = custom_output_result['explainability_info'][0]
if '/' in field_with_group:
# 그룹의 필드 부분에 대한 설정
(group, field) = field_with_group.split('/')
inference_result = inference_result[group]
explainability_info = explainability_info[group]
else:
field = field_with_group
value = inference_result[field]
confidence = explainability_info[field]['confidence']
print(f'{field}: {value or '<EMPTY>'} Confidence: {confidence}')
def main() -> None:
if len(sys.argv) < 2:
print("Please provide a filename as command line argument")
sys.exit(1)
file_name = sys.argv[1]
aws_account_id = get_aws_account_id()
input_s3_uri = f"s3://{BUCKET_NAME}/{INPUT_PATH}/{file_name}" # File
output_s3_uri = f"s3://{BUCKET_NAME}/{OUTPUT_PATH}" # Folder
data_automation_arn = f"arn:aws:bedrock:{AWS_REGION}:{aws_account_id}:data-automation-project/{PROJECT_ID}"
print(f"Invoking Bedrock Data Automation for '{file_name}'", end='', flush=True)
data_automation_response = invoke_data_automation(input_s3_uri, output_s3_uri, data_automation_arn, aws_account_id)
data_automation_status = wait_for_data_automation_to_complete(data_automation_response['invocationArn'])
if data_automation_status['status'] == 'Success':
job_metadata_s3_uri = data_automation_status['outputConfiguration']['s3Uri']
print_results(job_metadata_s3_uri)
if __name__ == "__main__":
main()스크립트의 초기 구성에는 입력 및 출력에 사용할 S3 버킷의 이름, 버킷의 입력 파일 위치, 결과의 출력 경로, Bedrock Data Automation에서 사용자 지정 출력을 가져오는 데 사용할 프로젝트 ID, 출력에 표시할 블루프린트 필드가 포함됩니다.
입력 파일의 이름을 전달하는 스크립트를 실행하겠습니다. 출력에서 Bedrock Data Automation이 추출한 정보를 볼 수 있습니다. US-Driver-License가 일치하며 운전 면허증의 이름과 날짜가 출력에 인쇄됩니다.
예상대로 Bedrock Data Automation 프로젝트와 관련한 블루프린트에서 선택한 정보를 출력에서 볼 수 있습니다.
마찬가지로, 제 동료 Mike Chambers의 비디오 파일에서 동일한 스크립트를 실행합니다. 출력을 작게 유지하기 위해 전체 오디오 트랜스크립트나 비디오에 표시되는 텍스트는 인쇄하지 않습니다.
알아야 할 사항
이제 미국 동부(버지니아 북부)와 미국 서부(오리건)의 두 AWS 리전에서 교차 리전 추론을 통해 HAQM Bedrock Data Automation을 사용할 수 있습니다. 해당 리전의 Bedrock Data Automation을 사용하면 미국 동부(오하이오, 버지니아 북부)와 미국 서부(캘리포니아 북부, 오리건) 등 4개 리전 중 하나에서 교차 리전 추론을 사용하여 데이터를 처리할 수 있습니다. 이들 리전은 모두 미국에 있으므로, 데이터가 동일한 지역 내에서 처리됩니다. AWS는 현재 2025년 후반에 유럽과 아시아의 더 많은 리전에 대한 지원을 추가하기 위해 노력하고 있습니다.
평가판과 비교하여, 그리고 교차 리전 추론을 사용할 때와 비교하여 요금에는 변화가 없습니다. 자세한 내용은 HAQM Bedrock 요금 페이지를 참조하세요.
이제 Bedrock Data Automation에는 세분화된 암호화 제어를 위한 AWS Key Management Service(AWS KMS) 고객 관리형 키 지원, 인터넷을 통해 연결하는 것이 아니라 가상 프라이빗 클라우드(VPC)에서 Bedrock Data Automation API에 직접 연결할 수 있는 AWS PrivateLink, 비용을 추적하고 AWS Identity and Access Management(AWS IAM)의 태그 기반 액세스 정책을 적용하기 위한 Bedrock Data Automation 리소스 및 작업에 대한 태깅 등 다양한 보안, 거버넌스 및 관리 관련 기능도 추가되었습니다.
이 블로그 게시물에서는 Python을 사용했지만 Bedrock Data Automation은 모든 AWS SDK와 함께 사용할 수 있습니다. 예를 들어 백엔드 문서 처리 애플리케이션에는 Java, .NET 또는 Rust를, 이미지, 비디오 또는 오디오 파일을 처리하는 웹 앱에는 JavaScript를, 최종 사용자가 제공한 콘텐츠를 처리하는 네이티브 모바일 앱에는 Swift를 각각 사용할 수 있습니다. 그 어느 때보다 손쉽게 멀티모달 데이터에서 인사이트를 도출할 수 있게 되었습니다.
다음은 자세한 내용을 참조할 수 있는 코드 샘플을 비롯한 몇 가지 참고 자료입니다.
- Simplify multimodal generative AI with HAQM Bedrock Data Automation
- HAQM Bedrock Data Automation을 사용한 멀티모달 데이터 처리 지침
- HAQM Bedrock Data Automation 사용 설명서
– Danilo