按需和批量流水线动态提取数据

许多公司拥有大量纸质或电子文档，其中包含未开发的商业智能。随着生成式AI的发展，各种大型语言模型可从这些文档中准确提取相关数据。本文介绍了一种智能文档处理流水线，它在Amazon Bedrock上同时提供按需推理和批量推理选项，从而灵活控制文档处理时间和成本。对于时间敏感的请求，可以使用按需推理选项；而批量推理选项则最具成本效益。文章还解释了如何在文档级别动态指定大语言模型和提示词，使您能够通过同一流水线从多种类型的文档中提取数据。

解决方案概述

如果您像我们的某位客户一样，积压了数亿份扫描PDF格式的租地文档（仅包含图像无可编辑文本的PDF，例如扫描保存的租地文件），且新文档每天仍在增加，那么这是一个可用于有效提取数据的解决方案。如下图所示，该解决方案构建了按需和批量两个推理流水线，并配有动态调用机制。通过使用Amazon Bedrock Prompt Management中有效设计的提示词，可以从格式和惯例各异的扫描PDF或文本文件中提取并标准化数据。

左侧是按需流水线，逐一处理文档并在数秒内返回结果，适用于时间敏感的请求。右侧是批量推理流水线，在单个Amazon Bedrock批量推理作业中处理多个文档请求，模型调用将异步进行。用户可以在两个流水线的请求中指定提示词ID和版本，相应的提示词文本将从Amazon Bedrock Prompt Management中检索。

1. 按需推理流水线

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

1.1 AWS SQS FIFO队列

使用FIFO队列是为了保证可靠的消息传递（每条消息传递一次）、严格的消息顺序处理以及消息分组功能。消息可以通过AWS CLI或AWS SDK API外部创建。例如：aws sqs send-message --queue-url ... --message-group-id "1" --message-body "msg 1" --message-attributes file://message_txt.txt。Lambda函数在收到提取数据后将删除队列消息。

1.2 Lambda函数处理与推理

Lambda函数首先使用s3_location属性从S3下载文档，如果是扫描PDF则转换为图像。由于Claude 4 Sonnet模型每次多模态调用最多允许20张图像，因此超过20页的文档需拆分为每20页一个块。文档ID、块总数和块ID与提取结果和模型性能指标一起存储在DynamoDB表中。

接着，从Amazon Bedrock Prompt Management检索提示词。租地文档格式各异，使用定制提示词可提高提取准确性。提示词有唯一ID并区分版本，SQS消息中需指定这些信息。注意，每个区域最多50个提示词，每个提示词最多10个版本。

Lambda函数然后组合提示词和图像，通过Converse API向Amazon Bedrock发送请求。LLM返回JSON字符串格式的提取数据，最终解析并存储到DynamoDB表，并删除SQS消息。

2. 批量推理流水线

批量推理流水线使用标准SQS队列以获得高吞吐量。主要组件包括：Amazon EventBridge调度器、批量推理Lambda函数（预处理扫描PDF、创建JSONL文件并提交批量推理作业）、Amazon EventBridge规则和后处理Lambda函数。

2.1 EventBridge调度器

按计划启动批量推理Lambda函数。

2.2 批量推理Lambda函数

函数首先检查队列中是否有足够消息（至少100条，因为Amazon Bedrock批量推理作业有最低记录数要求）。然后循环处理消息，提取文档ID和模型信息。与按需流水线类似，检索文档、转换为图像并拆分大文件。由于标准队列不保证恰好一次传递，函数需忽略重复消息。

不同文档格式需要不同用户提示词，通过SQS消息指定提示词ID和版本，函数从Amazon Bedrock Prompt Management检索提示词。接着创建元数据文件和JSONL文件，上传到批量推理数据S3存储桶，并删除SQS消息。最后，使用JSONL文件创建Amazon Bedrock批量推理作业。注意，一个批量作业只能使用一个模型ID，因此函数会选择出现次数最多的模型ID。

2.3 Amazon Bedrock批量推理作业

作业开始后，检索JSONL文件，完成推理后将输出存储到指定S3存储桶，并通过Amazon EventBridge发送状态变更事件。

2.4 EventBridge规则触发后处理Lambda函数

后处理Lambda函数从S3获取输出JSONL，解析并保存提取的租地属性到DynamoDB表。

先决条件

如果要尝试此示例，请确保拥有AWS账户及适当的IAM权限以创建和管理CloudFormation堆栈。

部署CloudFormation堆栈

部署按需流水线：点击启动堆栈链接进入AWS CloudFormation，按照步骤创建堆栈。接受IAM资源创建确认后提交。同样步骤部署批量流水线。

测试流水线

按需流水线测试：下载示例数据（三个租地文档），使用上传到S3的文档创建SQS消息。稍后可查看DynamoDB中的提取结果。批量流水线需至少100个文档方可测试。