亚马逊AWS官方博客
HAQM Bedrock 数据自动化现已正式推出
许多应用程序需要与通过不同模式提供的内容进行交互。其中一些应用程序处理复杂的文件，例如保险索赔和医疗账单。移动应用程序需要分析用户生成的媒体。企业需要在其数字资产（包括文档、图像、音频和视频文件）的基础上建立语义索引。但是，从非结构化多模态内容中获取见解并不容易：您必须针对不同的数据格式实施处理管道，并通过多个步骤来获取所需的信息。这通常意味着要有多个模型投入生产，您必须对这些模型进行成本优化（通过微调和提示工程）、保护措施（例如，防范幻觉）、与目标应用程序的集成（包括数据格式）和模型更新。
为简化这一过程，我们在 AWS re:Invent 期间推出了 HAQM Bedrock 数据自动化的预览版，这是 HAQM Bedrock 的一项功能，可简化从非结构化、多模态内容（例如文档、图像、音频和视频）生成有价值的见解的过程。借助 Bedrock Data Automation，您可以减少构建智能文档处理、媒体分析和其他以数据为中心的多模态自动化解决方案的开发时间和精力。
您可以将 Bedrock 数据自动化用作独立功能，也可以用作 HAQM Bedrock 知识库的解析器，对来自多模态内容的见解进行索引，并为检索增强生成（RAG）提供更相关的响应。
如今，Bedrock Data Automation 现已正式推出，支持跨区域推理端点，可在更多 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 数据自动化时，您现在需要指定要使用的项目或蓝图。如果输入蓝图，将获得自定义输出。要继续获取标准默认输出，请将参数 DataAutomationProjectArn 配置为使用 arn:aws:bedrock:<REGION>:aws:data-automation-project/public-default。
顾名思义，InvokeDataAutomationAsync 操作是异步的。您只需传递输入和输出配置，结果就会按照输出配置的指定写入 HAQM Simple Storage Service（HAQM S3）的存储桶中。您可以使用 notificationConfiguration 参数从 Bedrock 数据自动化接收 HAQM EventBridge 通知。
借助 Bedrock 数据自动化，您可以通过两种方式配置输出：
- 标准输出可提供与数据类型相关的预定义见解，如文档语义、视频章节摘要和音频转录。使用标准输出，您只需几个步骤即可获取所需的见解。
- 自定义输出让您可使用蓝图指定提取需求，从而获得更有针对性的见解。
为了查看新功能的实际运行情况，我创建了一个项目并自定义了标准输出设置。对于文档，我选择纯文本而不是标记符。请注意，您可以使用 Bedrock 数据自动化 API 自动执行这些配置步骤。
对于视频，我需要完整的音频转录和整个视频的摘要。我还要求提供各章的摘要。
要配置蓝图，我在 HAQM Bedrock 控制台导航窗格的数据自动化部分选择自定义输出设置。在那里，我搜索了美国驾驶执照示例蓝图。您可以浏览其他示例蓝图，以获取更多示例和想法。
示例蓝图无法编辑，因此我使用操作菜单复制蓝图，并将其添加到我的项目中。在那里，我可以通过修改蓝图和添加自定义字段来微调要提取的数据，这些字段可以使用生成式人工智能，以我需要的格式提取或计算数据。
我在 S3 存储桶中上传了美国驾照的图片。然后，我使用这个示例 Python 脚本，通过适用于 Python 的 HAQM SDK（Boto3）使用 Bedrock 数据自动化从图像中提取文本信息：
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'
# Fields to display
BLUEPRINT_FIELDS = [
'NAME_DETAILS/FIRST_NAME',
'NAME_DETAILS/MIDDLE_NAME',
'NAME_DETAILS/LAST_NAME',
'DATE_OF_BIRTH',
'DATE_OF_ISSUE',
'EXPIRATION_DATE'
]
# 适用于 Python 的 HAQM SDK（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
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)
# Custom output
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:
# For fields part of a 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 数据自动化获取自定义输出的项目 ID，以及要在输出中显示的蓝图字段。
我运行传递输入文件名称的脚本。在输出中，我看到了 Bedrock 数据自动化提取的信息。美国驾照与之匹配，并在输出中打印出驾照上的姓名和日期。
不出所料，我在输出中看到了从与 Bedrock 数据自动化项目相关的蓝图中选择的信息。
同样，我在同事 Mike Chambers 提供的视频文件上运行了相同的脚本。为了保持较小的输出量，我不会打印完整的音频脚本或视频中显示的文本。
注意事项
HAQM Bedrock 数据自动化现在可通过跨区域推理，在以下两个 AWS 区域使用：美国东部（弗吉尼亚州北部）和美国西部（俄勒冈州）。使用这些区域的 Bedrock 数据自动化时，可以在以下四个区域中使用跨区域推理来处理数据：美国东部（俄亥俄州、弗吉尼亚州北部）和美国西部（加利福尼亚北部、俄勒冈州）。所有这些区域都位于美国，因此数据都是在相同的地理区域内处理的。我们正在努力在 2025 年晚些时候增加对欧洲和亚洲更多地区的支持。
与预览版和使用跨区域推理相比，定价没有变化。有关更多信息，请访问 HAQM Bedrock 定价。
Bedrock 数据自动化现在还包括许多与安全、治理和可管理性相关的功能，如 AWS Key Management Service（AWS KMS） 客户管理密钥支持，可实现细粒度加密控制；AWS PrivateLink 可直接连接到虚拟私有云（VPC）中的 Bedrock 数据自动化 API，而无需通过互联网连接；对 Bedrock 数据自动化资源和作业进行标记，以跟踪成本并在 AWS Identity and Access Management（IAM）中执行基于标记的访问策略。
我在这篇博客文章中使用了 Python，但任何 AWS SDK 都支持 Bedrock 数据自动化。例如，您可以将 Java、.NET 或 Rust 用于后端文档处理应用程序；将 JavaScript 用于处理图像、视频或音频文件的 Web 应用程序；将 Swift 用于处理最终用户提供的内容的本机移动应用程序。从多模态数据中获得见解从未如此简单。
以下是一些阅读建议，以了解更多信息（包括代码示例）：
– Danilo
*前述特定亚马逊云科技生成式人工智能相关的服务仅在亚马逊云科技海外区域可用，亚马逊云科技中国仅为帮助您了解行业前沿技术和发展海外业务选择推介该服务。