HAQM Web Services ブログ
HAQM Bedrock のデータオートメーションを利用してマルチモーダルコンテンツからインサイトを取得する (一般提供が開始されました)
多くのアプリケーションは、さまざまなモダリティを通じて利用できるコンテンツとインタラクションする必要があります。これらのアプリケーションの一部は、保険金請求や医療費請求書などの複雑なドキュメントを処理します。モバイルアプリは、ユーザーが作成したメディアを分析する必要があります。組織は、ドキュメント、画像、音声、動画ファイルなどのデジタルアセットの上にセマンティックインデックスを構築する必要があります。しかし、非構造化マルチモーダルコンテンツからインサイトを取得するようセットアップするのは簡単ではありません。さまざまなデータ形式用の処理パイプラインを実装し、必要な情報を取得するために複数のステップを実行する必要があります。これは通常、複数のモデルを本番で使用し、(ファインチューニングやプロンプトエンジニアリングを通じた) そのコストの最適化、安全策 (例: ハルシネーションへの対策)、ターゲットアプリケーションとの統合 (データ形式を含む)、モデルの更新に対応する必要があることを意味します。
このプロセスを容易にするために、HAQM Bedrock のデータオートメーションのプレビューを AWS re:Invent の期間中に導入しました。これは、ドキュメント、画像、音声、動画などの非構造化マルチモーダルコンテンツからの価値あるインサイトの生成を効率化する HAQM Bedrock の機能です。Bedrock のデータオートメーションを利用すると、インテリジェントなドキュメント処理、メディア分析、および他のデータ中心のマルチモーダルオートメーションソリューションを構築するための開発時間と労力を削減できます。
Bedrock のデータオートメーションをスタンドアロン機能として使用したり、HAQM Bedrock のナレッジベースのパーサーとして使用したりして、マルチモーダルコンテンツから得たインサイトをインデックス化し、検索拡張生成 (RAG) のためにより関連性の高い応答を提供できます。
本日、Bedrock のデータオートメーションは、クロスリージョン推論エンドポイントのサポートとともに一般提供が開始され、より多くの AWS リージョンで利用できるようになり、さまざまな場所でシームレスにコンピューティングを使用するようになりました。プレビュー期間中のフィードバックに基づいて、精度を高め、画像や動画のロゴ認識のサポートも追加しました。
実際にどのように機能するかを見てみましょう。
HAQM Bedrock のデータオートメーションとクロスリージョン推論エンドポイントの併用
Bedrock のデータオートメーションプレビューについての公開ブログ記事には、HAQM Bedrock コンソールでビジュアルデモを使用してドキュメントや動画から情報を抽出する方法が記載されています。この機能がどのように動作し、カスタマイズするために何ができるのかを理解するために、コンソールのデモエクスペリエンスをお試しいただくことをお勧めします。この記事では、コンソールでのいくつかのステップから始めて、コードサンプルに従いながら、アプリケーションでの Bedrock のデータオートメーションの動作に焦点を当てます。
HAQM Bedrock コンソールの [データオートメーション] セクションでは、初めてアクセスしたときに、クロスリージョンサポートを有効にするかどうかの確認が求められるようになりました。例:
API の観点から見ると、InvokeDataAutomationAsync オペレーションでは、使用するデータオートメーションプロファイルを指定するために、追加のパラメータ (dataAutomationProfileArn) が必要になりました。このパラメータの値は、リージョンと AWS アカウント ID によって異なります:
arn:aws:bedrock:<REGION>:<ACCOUNT_ID>:data-automation-profile/us.data-automation-v1
また、dataAutomationArn パラメータの名前が dataAutomationProjectArn に変更され、プロジェクトの HAQM リソースネーム (ARN) が含まれていることをより明確に示すようになりました。Bedrock のデータオートメーションを呼び出す際に、使用するプロジェクトまたはブループリントを指定しなければならなくなりました。ブループリントを渡すと、カスタム出力が得られます。引き続き標準のデフォルト出力を取得するには、arn:aws:bedrock:<REGION>:aws:data-automation-project/public-default を使用するようにパラメータ DataAutomationProjectArn を設定します。
名前が示すように、InvokeDataAutomationAsync オペレーションは非同期です。入力と出力の設定を渡した後、結果の準備ができたら、出力設定で指定された HAQM Simple Storage Service (HAQM S3) バケットに書き込まれます。notificationConfiguration パラメータを使用して、Bedrock のデータオートメーションから HAQM EventBridge の通知を受け取ることができます。
Bedrock のデータオートメーションでは、2 つの方法で出力を設定できます:
- [標準出力] は、ドキュメントのセマンティクス、動画の章の概要、音声トランスクリプトなど、データタイプに関連する事前定義済みのインサイトを提供します。標準出力を使用すると、わずか数ステップで必要なインサイトをセットアップできます。
- [カスタム出力] を使用すると、ブループリントを使用して抽出のニーズを指定し、よりカスタマイズされたインサイトを得ることができます。
新しい機能が実際に動作している様子を見るために、プロジェクトを作成し、標準出力設定をカスタマイズします。ドキュメントでは、マークダウンではなく、プレーンテキストを選択します。なお、これらの設定ステップは、Bedrock Data Automation API を使用して自動化できます。
動画では、完全な音声トランスクリプトと動画全体の要約が必要です。また、各章の要約も要求します。
ブループリントを設定するには、HAQM Bedrock コンソールのナビゲーションペインの [データオートメーション] セクションで [カスタム出力設定] を選択します。そこで、[US-Driver-License] サンプルブループリントを検索します。より多くの例やアイデアを得るために、他のサンプルブループリントを参照できます。
サンプルブループリントは編集できないため、[アクション] メニューを使用してブループリントを複製し、プロジェクトに追加します。そこで、ブループリントを変更し、生成 AI を使用して必要な形式でデータを抽出または計算できるカスタムフィールドを追加することで、抽出するデータをファインチューニングできます。
米国の運転免許証の画像を S3 バケットにアップロードします。その後、AWS SDK for Python (Boto3) を通じて Bedrock のデータオートメーションを利用するこのサンプル 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) clients
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}" # ファイル
output_s3_uri = f"s3://{BUCKET_NAME}/{OUTPUT_PATH}" # フォルダ
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 のデータオートメーションからカスタム出力を取得するために使用するプロジェクト ID、および出力に表示するブループリントフィールドが含まれます。
入力ファイルの名前を渡してスクリプトを実行します。出力には、Bedrock のデータオートメーションによって抽出された情報が表示されます。[US-Driver-License] が一致し、運転免許証の名前と日付が出力に表示されます。
想定どおり、出力には、Bedrock のデータオートメーションプロジェクトに関連付けられたブループリントから選択した情報が表示されます。
同様に、私の同僚である Mike Chambers の動画ファイルに対して同じスクリプトを実行します。出力サイズを小さくするために、完全な音声トランスクリプトや動画に表示されるテキストは表示しません。
知っておくべきこと
HAQM Bedrock のデータオートメーションは現在、米国東部 (バージニア北部) と米国西部 (オレゴン) の 2 つの AWS リージョンでクロスリージョン推論を介してご利用いただけます。これらのリージョンから Bedrock のデータオートメーションを利用する場合、米国東部 (オハイオ、バージニア北部) と米国西部 (北カリフォルニア、オレゴン) の 4 つのリージョンのいずれかでクロスリージョン推論を使用してデータを処理できます。これらのリージョンはすべて米国内にあるため、データは同じ地域内で処理されます。2025 年後半には、欧州とアジアのさらに多くのリージョンのサポートを追加する予定です。
プレビュー、およびクロスリージョン推論を使用する場合と比較して、料金に変更はありません。詳細については、「HAQM Bedrock の料金」にアクセスしてください。
また、Bedrock のデータオートメーションには、きめ細かな暗号化コントロールのための AWS Key Management Service (AWS KMS) カスタマーマネージドキーのサポート、インターネット経由ではなく、仮想プライベートクラウド (VPC) 内の Bedrock Data Automation API に直接接続するための AWS PrivateLink、コストを追跡し、AWS Identity and Access Management (IAM) でタグベースのアクセスポリシーを強制適用するための Bedrock のデータオートメーションのリソースとジョブのタグ付けなど、セキュリティ、ガバナンス、管理性に関連する多数の機能も含まれるようになりました。
このブログ記事では Python を利用しましたが、Bedrock のデータオートメーションはどの AWS SDK でもご利用いただけます。例えば、バックエンドのドキュメント処理アプリケーションには Java、.NET、または Rust、画像、動画、または音声ファイルを処理するウェブアプリケーションには JavaScript、エンドユーザーが提供するコンテンツを処理するネイティブモバイルアプリには Swift を、それぞれご利用いただけます。マルチモーダルデータからインサイトを取得するのがこれまでになく簡単になりました。
詳細 (コードサンプルを含む) については、次の記事をお読みください:
- Simplify multimodal generative AI with HAQM Bedrock Data Automation
- HAQM Bedrock のデータオートメーションを使用したマルチモーダルデータ処理に関するガイダンス
- HAQM Bedrock のデータオートメーションのユーザーガイド
– Danilo
私たちの取り組みについて、どのように感じましたか? この 1 分間のアンケートにぜひご協力ください!
原文はこちらです。