利用 Amazon Textract 创建文档湖

关键要点

AWS 的客户在医疗、金融、公共部门及其他行业中将数十亿个文档存储为图像或 PDF 文件在 Amazon Simple Storage ServiceAmazon S3中。然而，他们无法从这些文档中获得有价值的洞察，例如将封闭的信息用于大型语言模型LLMs或搜索，直到提取文本、表单、表格等其他结构化数据。通过使用 AWS 智能文档处理IDP和 AI 服务如 Amazon Textract，用户能够利用顶尖的机器学习ML技术快速准确地处理来自 PDF 或文档图像TIFF、JPEG、PNG的数据。一旦从文档中提取出文本，用户可以使用这些数据进行模型微调或发送到数据库。

在本篇文章中，我们专注于将大量文档处理为原始文本文件并存储到 Amazon S3。我们提供了两种不同的解决方案：第一种是从任意服务器或实例包括 Jupyter Notebook运行 Python 脚本，这是最快的入门方式；第二种是使用 AWS Cloud Development KitAWS CDK构建各种基础设施组件，这种方式可以实现更为全面的功能，处理文档并构建完整的 IDP 流程。

这两种解决方案均可快速处理数百万页文档。建议在大规模运行前先测试一小部分文档，以确保结果符合预期。接下来的部分，我们将首先介绍脚本解决方案，然后是 AWS CDK 解决方案。

解决方案一：使用 Python 脚本

该解决方案通过 Amazon Textract 快速处理文档以获得原始文本，并假设如果脚本出现故障，流程将从停下的地方继续。此解决方案涉及三个服务：Amazon S3、Amazon DynamoDB 和 Amazon Textract。以下图展示了脚本执行的事件顺序。脚本结束时，完成状态和所用时间将返回至 SageMaker Studio 控制台。

我们已将此解决方案打包成 ipynb 脚本 和 py 脚本。您可以根据需求使用任一可部署的解决方案。

前提条件

要从 Jupyter Notebook 运行此脚本，分配给笔记本的 AWS 身份和访问管理IAM角色必须具有与 DynamoDB、Amazon S3 和 Amazon Textract 交互的权限。一般建议为每项服务提供最小权限以遵循安全原则。有关详细信息，请参考 使用 AWS 管理策略并朝向最小权限移动。

替代方案是可以在其他环境中运行此脚本，例如通过您管理的 Amazon Elastic Compute CloudAmazon EC2实例或容器，只要安装有 Python、Pip3 和 AWS SDK for Python (Boto3)，同样需要应用允许脚本与管理服务交互的 IAM 策略。

实施步骤

首先，您需要克隆 GitHub 存储库。

在运行脚本之前，您需要设置以下变量：

脚本如下：

pythontrackingtable = TableNameforstorings3ObjectNamesinputbucket = yourfilesarehereoutputbucket = Amazon TextractwritesJSONcontainingrawtexttohere

运行脚本时，将创建以下 DynamoDB 表架构：

textTable TableNameforstorings3ObjectNamesPartition Key objectName (String) bucketName (String) createdDate (Decimal) outputbucketName (String) txJobId (String)

脚本第一次运行时，将检查 DynamoDB 表是否存在，如有需要会自动创建。创建表后，需要将希望处理的 Amazon S3 中的文档对象引用列表填充到表中。脚本会遍历指定的 inputbucket 中的对象，并自动填充表中的名称。填充一百万个文档名称大约需要 10 分钟。如果您在某个存储桶中有数百万个对象，可以利用 Amazon S3 的清单功能生成名称的 CSV 文件，随后自行脚本执行将该 CSV 列表填充至 DynamoDB 表，避免使用 fetchAllObjectsInBucketandStoreName 函数。

如前所述，提供了一个 Notebook 版本和 Python 脚本版本。Notebook 是最直接的入门方式，直接运行每个单元即可。

如果您选择从 CLI 运行 Python 脚本，建议使用终端复用器如 tmux，以防脚本因 SSH 会话结束而停止。例如：tmux new d python3 textractFeederpy。

以下是 脚本的入口点，您可以在此注释掉不必要的方法：

pythonMain entry point into script Start Hereif name == main now = timeperfcounter() print(started)

在脚本填充 DynamoDB 表时，将设置以下字段：

如果您选择使用 S3 清单报告中的 CSV 文件，必须填充这两个字段。

现在表已创建并填充文档对象引用，脚本准备调用 Amazon Textract 的 StartDocumentTextDetection API。Amazon Textract 具有对 API 调用的每秒调用限制。若有需要，您可以从 Amazon Textract 控制台请求配额增加。代码旨在同时使用多线程以最大化与服务的吞吐量。您可以通过修改 threadCountforTextractAPICall 变量在代码中更改此限制。默认设置为 20 线程。脚本最初从 DynamoDB 表读取 200 行，并将其存储在一个包裹于线程安全类的内存列表中。每个调用线程然后启动并在自己的 游道 中运行。基本上，Amazon Textract 调用线程将从内存列表中检索包含对象引用的项目。随后调用异步的 startdocumenttextdetection API 并等待带有作业 ID 的确认。作业 ID 随后回传至相应的 DynamoDB 行，并且线程将重复此过程，检索列表中的下一个项目。

以下是主协调代码的 脚本：

pythonwhile len(results) gt 0 for record in results # 将记录放入线程安全列表中 fileListappend(record) 创建处理 Amazon Textract 的线程 threadsforTextractAPI = threadingThread(name=Thread str(i) target=procestTextractFunction args=(fileList))

调用线程将持续运行，直到列表中没有项目；一旦所有 游道 中的线程停止，则从 DynamoDB 检索下一组 200 行并启动新的 20 个线程，整个过程将不断重复，直到所有不含作业 ID 的行从 DynamoDB 检索并更新为止。

如果脚本因某些意外问题崩溃，您可以再次从 orchestrate() 方法运行该脚本，以确保线程继续处理包含为空作业 ID 的行。请注意，当在脚本停止后重新运行 orchestrate() 方法时，可能会有部分文档重新发送至 Amazon Textract。这个数字将等于或少于崩溃时正在运行的线程数量。

当 DynamoDB 表中不再有包含空作业 ID 的行时，脚本将停止。所有来自 Amazon Textract 的 JSON 输出将默认存储在 outputbucket 的 textractoutput 文件夹下。每个 textractoutput 文件夹将以相应作业 ID 命名，文件夹中包含的 JSON 文件则是从 1 开始的数字命名，并可能扩展为额外的 JSON 文件例如：2、3 等。扩展 JSON 文件的原因在于密集或多页文档中提取的内容超出了 Amazon Textract 默认的 JSON 大小限制1000 个区块。有关块的更多信息，请参考 Block。

您可以在 GitHub 中找到此解决方案的 Python Notebook 版本和脚本。

清理

当 Python 脚本完成后，您可以关闭或停止启动的 Amazon SageMaker Studio 笔记本或容器以节省成本。

接下来是我们的第二个解决方案，用于大规模文档处理。

解决方案二：使用无服务器 AWS CDK 构造

此解决方案使用 AWS Step Functions 和 Lambda 函数来协调 IDP 流程。我们使用 IDP AWS CDK 构造，这使得与 Amazon Textract 大规模配合工作变得简单。此外，我们使用 Step Functions 分布式映射 来遍历 S3 存储桶中的所有文件并启动处理。第一个 Lambda 函数确定您的文档包含多少页，从而使管道能够自动选择同步对于单页文档或异步对于多页文档API。使用异步 API 时，将调用另一个 Lambda 函数，将 Amazon Textract 为所有页面生成的 JSON 文件整合为一个 JSON 文件，方便下游应用程序处理信息。

该解决方案还包含两个附加的 Lambda 函数。第一个函数将 JSON 中的文本解析并保存为 Amazon S3 中的文本文件。第二个函数分析 JSON 并存储以用于工作负载度量。

以下图展示了 Step Functions 的工作流。

前提条件

此代码库使用 AWS CDK，并要求安装 Docker。您可以在已配置 AWS CDK 和 Docker 的 AWS Cloud9 实例中进行部署。

实施步骤

要实现此解决方案，首先需要克隆 存储库。

克隆存储库后，安装依赖项：

bashpip install r requirementstxt

然后使用以下代码部署 AWS CDK 堆栈：

bashcdk bootstrapcdk deploy parameters SourceBucket=ltSource Bucketgt SourcePrefix=ltSource Prefixgt

您必须提供源存储桶和源前缀要处理的文件位置。

部署完成后，导航至 Step Functions 控制台，您将看到状态机 ServerlessIDPArchivePipeline。

打开状态机详细信息页，在 Executions 选项卡上，选择 Start execution。

再次选择 Start execution 以运行状态机。

启动状态机后，您可以通过查看映射运行监控管道。在 Item processing status 部分，您会看到该进程的运行情况，明确哪些评估为成功，哪些为失败。此过程将持续，直到所有文档都被处理。

通过该解决方案，您应该能够在 AWS 账户中处理数百万个文件，而无需担心如何正确确定哪些文件发送至哪个 API 或因文件损坏导致工作流失败。通过 Step Functions 控制台，您可以实时监控和跟踪文件。

清理

当管道停止运行后，要清理资源，可以返回项目并输入以下命令：

bashcdk destroy

这将删除该项目为之部署的所有服务。

结论

在本文中，我们展示了一个将文档图像和 PDF 转换为文本文件的解决方案。这是将文档用于生成 AI 和搜索的关键前提。要了解更多关于如何使用文本训练或微调基础模型的信息，请参考 在 Amazon SageMaker JumpStart 上微调 Llama 2 进行文本生成。有关搜索的使用，请参考 使用 Amazon Textract 和 Amazon OpenSearch 实现智能文档搜索索引。要进一步了解 AWS AI 服务提供的高级文档处理能力，请参考 AWS 上智能文档处理的指南。

作者简介

Tim Condello 是 Amazon Web Services (AWS) 的高级人工智能 (AI) 和机器学习 (ML) 专家解决方案架构师。他专注于自然语言处理和计算机视觉，乐于将客户的想法转化为可扩展的解决方案。

David Girling 是一名高级 AI/ML 解决方案架构师，拥有超过 20 年的企业系统设计和开发经验。他是一个专门团队的一员，专注于帮助客户学习、创新，并利用这些高效的服务处理数据。