Extract Data with On-demand and Batch Pipelines Dynamically

动态使用按需和批量流水线提取数据 | 人工智能

许多公司拥有大量纸质或电子文档，其中包含未开发的商业智能。随着生成式人工智能的发展，各种大型语言模型可以用于从这些文档中准确提取相关信息。本文演示了一种智能文档处理流水线，该流水线在 Amazon Bedrock 上同时包含按需推理和批量推理选项，从而实现文档处理时间和成本的灵活性。对于时间敏感的请求，可以使用按需推理选项，而批量推理选项则在成本优化方面表现最佳。它还解释了如何在文档级别动态指定大型语言模型和提示，使您能够使用相同的流水线从多种类型的文档中提取数据。

解决方案概述

如果您像我们的客户之一一样，拥有数亿份以扫描 PDF 格式（仅包含图像而不含可编辑文本的 PDF，例如在此情况下，扫描的地产租赁合同保存为 PDF）形式存储的地产租赁文档，并且每天仍有新文档不断堆积，那么您可以使用此解决方案从这些文档中有效提取数据。如下面的图表所示，该解决方案构建了两个推理流水线：按需和批量，并通过一种机制动态调用它们。通过使用在 Amazon Bedrock 提示管理中有效设计的提示，可以从格式和惯例各不相同的扫描 PDF 或文本文件中提取并标准化数据。

左侧的流水线是按需流水线，它逐个从文档中提取数据，并在几秒钟内返回结果。这使其适用于时间敏感的请求。

右侧的流水线是批量推理流水线，它通过单个 Amazon Bedrock 批量推理作业处理多个文档请求，您的模型调用将被异步处理。在两个流水线中，用户都可以在请求中指定提示 ID 和版本，相应的提示文本将从 Amazon Bedrock 提示管理中检索。

以下部分将详细描述这两个流水线。

1. 按需推理流水线

在按需推理流水线中创建了一个 AWS SQS 先进先出（FIFO）队列。当包含文档 ID、LLM 模型 ID、提示 ID/版本和系统提示 ID/版本的队列消息到达时，会触发一个 AWS Lambda 函数。该函数从指定的 Amazon S3 存储桶中检索 PDF 文档，将 PDF 页面转换为 PNG 图像，从 Amazon Bedrock 提示管理中检索相关提示，构建消息以调用 LLM，并将结果保存到 Amazon DynamoDB 表中。

1.1. AWS SQS FIFO 队列

当单个文档到达时，使用 AWS SQS FIFO 队列来触发 Amazon Bedrock 推理。使用 FIFO 队列的主要原因包括：

• 可靠的消息传递 – 确保每条消息恰好传递一次。

• 先进先出（FIFO）处理 – 保持严格的顺序，提供更好的处理可预测性。

• 消息分组 – 消息组 ID 属性确保每个组内的消息按顺序处理。每个生产者可以使用唯一的消息组 ID 来维护相关消息的顺序。

#### 队列消息是如何创建的？

队列消息可以通过 AWS CLI 或 AWS SDK API 从外部创建。以下是一个 AWS CLI 命令示例：

aws sqs send-message --queue-url https://sqs.us-east-1.amazonaws.com/1111111111/ondemand-data-pipeline-queue.fifo --message-group-id "1" --message-body "msg 1" --message-attributes file://message_txt.txt

在这个示例中，message_txt.txt 文件是一个包含应用程序所需消息属性的 JSON 文件。有关详细信息，请参阅下面“测试管道”部分。

在 Amazon Bedrock 返回提取数据后，Lambda 函数将删除队列消息。

1.2. Lambda 函数 – 队列消息处理和推理

#### 1.2.1 获取文档、转换为图像和拆分大文件

Lambda 函数使用队列消息中的 s3_location 属性下载文档。如果文档是扫描的 PDF，它将被转换为图像，以便多模态模型理解。

截至目前，Claude 4 Sonnet 模型每次多模态调用最多允许 20 张图像。因此，如果文档包含超过 20 页的图像，必须将其拆分为 20 页的块。doc_id、chunk_count 和 chunk_id 与提取结果和模型性能指标一起存储在 Amazon DynamoDB 表中。

• doc_id：文档的标识符

• chunk_count：该文档的块总数

• chunk_id：文档每个块的标识符

#### 1.2.2 从 Amazon Bedrock 提示管理中获取提示

土地租赁文件的格式各不相同 – 有些以编号列表的形式呈现土地地块属性，有些以表格形式呈现，还有一些甚至以土地图纸的形式呈现。因此，使用针对每种文档格式定制的不同提示可以提高提取的准确性。

在 LLM 调用中使用的提示存储在 Amazon Bedrock 提示管理中。每个提示都有一个唯一的 ID 并且有版本。SQS 消息必须指定相关的提示 ID 和版本，这些信息在 Lambda 执行期间用于检索提示内容。

注意：每个区域最多有 50 个提示，每个提示最多有 10 个版本。

#### 1.2.3 为 LLM 调用构建消息并处理响应

Lambda 函数继续执行以下步骤：

• 通过将提示正文和图像连接起来，构建 LLM 的消息。

• 使用 Converse API 向 Amazon Bedrock 发送请求。

LLM 将以 JSON 字符串的形式返回提取的数据，您可以在 DynamoDB 表中查看结果，如下面“测试管道”部分所示。

#### 1.2.4 保存结果

最后，Lambda 函数通过以下步骤完成处理过程：

• 解析 JSON 并将土地地块属性存储到 DynamoDB 表中。

• 如果文档已成功处理并保存了结果，SQS 消息将从队列中删除。

2. 批处理推理管道

由于其高吞吐量，标准的 AWS SQS 队列用于批处理推理管道。队列消息的创建方式与按需管道类似，只是不需要 message-group-id 属性。

批处理推理管道的主要组件包括：

• Amazon EventBridge Scheduler。

• 用于预处理扫描 PDF、创建 JSONL 文件并提交批处理推理任务的批处理推理 AWS Lambda 函数。

• Amazon EventBridge 规则。

• 后处理 AWS Lambda 函数。

以下部分描述了批量推理管道的详细信息。

2.1. Amazon EventBridge 调度器

Amazon EventBridge 调度器按照预定的时间安排启动批量推理 Lambda 函数。

2.2. 批量推理 Lambda 函数

该函数首先检查队列中是否有足够的消息，然后再继续执行。在撰写本文时，Amazon Bedrock 批量推理任务的记录数最低要求为 100 条。

#### 2.2.1 接收队列消息

Lambda 函数遍历队列中的消息，并提取文档 ID、LLM 模型 ID、提示 ID/版本和系统提示 ID/版本。

#### 2.2.2 获取无重复文档、转换为图像和拆分大文件

Lambda 函数随后获取文档，如果文档是扫描的 PDF，则将其转换为图像，并在必要时拆分大文件——这与按需管道中的处理方式相同。由于标准 SQS 队列无法保证消息的精确一次传递，该函数还会确保忽略重复的消息。

#### 2.2.3 允许在批量推理任务中使用不同的提示

与按需管道类似，不同的文档格式需要不同的用户提示以实现更有效的数据提取。

每个文档的目标提示 ID 和版本在 SQS 消息中指定。在 Lambda 执行期间，该函数从 Amazon Bedrock 提示管理中检索提示内容。

#### 2.2.4 为批量推理任务创建 JSONL 工件

Lambda 函数随后执行以下任务：

• 在 Batch Inference Data S3 存储桶中创建 metadata.json 文件，用于存储消息属性，包括 SQS 消息 ID、doc_id、提示 ID/版本、系统提示 ID/版本和其他项目相关属性。该文件之后将被后处理 Lambda 用于填充 DynamoDB 表。

• 处理文档以创建 Amazon Bedrock 批量推理任务所需的 JSONL 文件。该过程使用 Python 的 multiprocessing 模块进行并行处理以提高效率。JSONL 文件上传到 Batch Inference Data S3 存储桶。

• 在文档准备并上传到 S3 存储桶后，删除 SQS 消息。这需要为队列设置一个较长的可见性超时。

#### 2.2.5 组装消息并提交批量推理任务

最后，批量推理 Lambda 函数使用上一步的 JSONL 工件创建 Amazon Bedrock 批量推理任务。请注意，每个批量任务只能使用一个模型处理文档，这意味着同一批量任务中的 SQS 消息必须指定相同的模型 ID。如果传入的消息中指定了多个模型 ID，Lambda 函数将使用轮询机制选择最常指定的模型 ID 来使用。

2.3. Amazon Bedrock 批量推理任务

当 Amazon Bedrock 接收到批量推理任务时，会将其放入队列中。一旦任务开始，它将执行以下步骤。

#### 2.3.1 获取批量推理任务的 JSONL 工件

Amazon Bedrock 获取在任务创建期间指定的 JSONL 工件。

#### 2.3.2 存储批量推理输出

任务完成后，Amazon Bedrock 将输出存储到 Batch Inference Data S3 存储桶中，该存储桶也在任务创建时指定。

#### 2.3.3 通知 Amazon EventBridge

任务完成后，Amazon Bedrock 会向 Amazon EventBridge 发送一个作业状态更改事件，该事件会被 EventBridge 规则捕获。

2.4. EventBridge 规则触发后推理 Lambda 函数

EventBridge 规则会触发后处理 Lambda 函数，以处理进一步的模型输出处理。

2.5. 后处理 Lambda 函数

#### 2.5.1 获取输出 JSONL

Lambda 函数从批量推理数据的 S3 存储桶中获取推理输出 JSONL。

#### 2.5.2 保存推理输出

该函数解析 JSONL 文件，并将提取出的地块属性保存到 DynamoDB 表中。

先决条件

如果你想亲自尝试这个示例，请确保满足以下先决条件：

• 一个可以访问 AWS 管理控制台的 AWS 账户

• 适当的 IAM 权限以创建和管理 CloudFormation 堆栈，通常包括：cloudformation:CreateStack cloudformation:DescribeStacks cloudformation:UpdateStack cloudformation:DeleteStack

部署 CloudFormation 堆栈

部署按需管道：

当你选择“Launch Stack”链接时，你将被带到 AWS CloudFormation 以启动 CloudFormation 堆栈：

• 在“Create stack”页面上，选择“Next”

• 在“Specify stack details”页面上，选择“Next”

• 在“Configure stack options”页面上，选择“Next”

• 在“Review and create”页面上，选择“I acknowledge that AWS CloudFormation might create IAM resources”

• 选择“Submit”

提交后，你可以查看堆栈的一些信息，例如 Stack info、Events、Resource 等。以下截图是 Events 的参考示例：

你也可以按照相同的步骤部署批量管道。

测试管道

以下步骤指导你测试按需管道。如果你至少有 100 个文档，也可以按照类似步骤测试批量管道。

• 将数据下载到你的本地环境。有三个来自 Winkler 县、Andrews 县和 Sutton 县的土地文档，这些文档是从 Texas Land Records 和 County Records 网站购买的。

• 将下载的 PDF 文件上传到 CloudFormation 堆栈创建的 S3 存储桶 ondemand-data-pipeline-bucket-${account_id}。

• 使用以下示例创建一个文本文件 message_txt.json，替换从你的 CloudFormation 堆栈创建的提示 ID、系统提示 ID 和 S3 存储桶。

{ 
  "application": {
    "DataType": "String",
    "StringValue": "bedrock-example"
  },
  "id": {
    "DataType": "String",
    "StringValue": "Winkler_2024-06-05_N_C42758_V_OPR"
  },
  "model_id": {
    "DataType": "String",
    "StringValue": "anthropic.claude-sonnet-4-20250514-v1:0"
  },
  "prompt_id": {
    "DataType": "String",
    "StringValue": "6CT88W3MWT"
  },
  "prompt_version": {
    "DataType": "String",
    "StringValue": "1"
  },
  "s3_location": {
    "DataType": "String",
    "StringValue": "s3://ondemand-data-pipeline-bucket-111111111/Winkler_2024-06-05_N_C42758_V_OPR.pdf"
  },
  "system_prompt_id": {
    "DataType": "String",
    "StringValue": "R2NFLXFXOJ"
  },
  "system_prompt_version": {
    "DataType": "String",
    "StringValue": "1"  
  }
}

• 使用上面的 AWS CLI 示例创建一个 shell 脚本 send2queue.sh，替换队列名称并执行它。你将在 SQS 队列 ondemand-data-pipeline-queue.fifo 中看到一条消息。

• 队列消息将触发 Lambda 函数 ondemand-data-pipeline-queue-processor。

• 在 Amazon CloudWatch 中查看 Lambda 日志，日志组为 /aws/lambda/ondemand-data-pipeline-queue-processor。

• 在 DynamoDB 表 ondemand-data-pipeline-table 中查看 Amazon Bedrock 推理输出。对于 Winkler County 示例，model_response 列中的 JSON 结果应如下所示：

[
  {
    "tract": 1,
    "state": "Texas",
    "county": "Winkler",
    "abstract": "A-1239",
    "survey": "PSL Survey",
    "section": "8",
    "range_block": "B2",
    "quarter": "N/2 of N/2"
  },
  {
    "tract": 2,
    "state": "Texas",
    "county": "Winkler",
    "abstract": "A-1239",
    "survey": "PSL Survey",
    "section": "8",
    "range_block": "B2",
    "quarter": "N/2 of S/2"
  },
  {
    "tract": 3,
    "state": "Texas",
    "county": "Winkler",
    "abstract": "A-1240",
    "survey": "PSL Survey",
    "section": "9",
    "range_block": "B2",
    "quarter": "S/2 of N/2"
  },
  {
    "tract": 4,
    "state": "Texas",
    "county": "Winkler",
    "abstract": "A-1240",
    "survey": "PSL Survey",
    "section": "9",
    "range_block": "B2",
    "quarter": "S/2 of S/2"
  }
]

清理

要清理资源：

• 登录 AWS 管理控制台

• 导航到 CloudFormation 服务

• 在 CloudFormation 仪表板中，找到并选择要删除的堆栈

• 在页面顶部选择“删除”按钮

• 在提示时确认删除

CloudFormation 将按照正确的顺序自动删除堆栈中创建的资源，并适当处理依赖关系。

删除 CloudFormation 堆栈不会删除 S3 存储桶和 DynamoDB，因为它们的删除策略设置为保留，以帮助防止数据丢失。要删除这些资源，请前往 AWS 管理控制台中每个服务的页面并手动删除它们。

结论

本文介绍的按需和批量 Amazon Bedrock 推理管道说明了如何根据时间敏感性和数据量动态处理文档。在选择使用哪种管道时，还应考虑成本因素。根据我们的测试，批量管道的 Amazon Bedrock 成本比按需管道低 50%。

该解决方案的另一个关键功能是能够在单个文档级别指定大型语言模型（用于按需管道）和提示，使这些管道能够支持各种类型的智能文档处理。

通过启用并行处理，使用 Python 的 multiprocessing 模块，批量推理管道的两个 Lambda 函数可以在 15 分钟内处理 1,000 个文档。

行动号召

Amazon Bedrock 可以帮助您构建许多生成式 AI 应用程序。我们建议您遵循以下 GitHub 仓库中的快速入门指南，熟悉构建生成式 AI 应用程序。对于高级读者，可以进一步研究如何扩展该解决方案。一个想法是将 Lambda 代码在 AWS Batch 中运行，从而允许在单个 Amazon Bedrock 批量推理作业中处理数万个文档。

作者简介

'"`