GSP927

概览

Document AI API 是一种文档理解解决方案，可以提取文档、电子邮件等非结构化数据，从而让用户更容易理解、分析和使用这些数据。

在本实验中，您将构建一个文档处理流水线，用于自动分析上传到 Cloud Storage 的文档。该流水线使用 Cloud Run 函数和 Document AI 表单处理器来提取数据，并将其存储在 BigQuery 中。如果表单包含地址字段，则会将地址数据发送到 Pub/Sub 主题。这会触发第二个 Cloud Run 函数，该函数使用 Geocoding API 添加坐标，并将结果写入 BigQuery。

这个简单的流水线使用通用表单处理器来检测基本表单数据，例如带标签的地址字段。对于更复杂的文档，Document AI 提供了专门的解析器（超出了本实验的范围），即使没有明确的标签，它也能提取详细信息。例如，账单解析器可以通过理解常见的账单布局，从未加标签的账单中识别地址和供应商详细信息。

您将创建的整体架构如下所示：

1. 将包含地址数据的表单上传到 Cloud Storage。
2. 上传操作会触发 Cloud Run 函数调用来处理表单。
3. 从 Cloud Run 函数调用 Document AI。
4. Document AI JSON 数据保存回 Cloud Storage。
5. Cloud Run 函数写入 BigQuery 的表单数据。
6. Cloud Run 函数将地址发送到 Pub/Sub 主题。
7. Pub/Sub 消息触发 Cloud Run 函数来进行地理编码处理。
8. 从 Cloud Run 函数调用 Geocoding API。
9. 对 Cloud Run 函数写入 BigQuery 的数据进行地理编码。

此示例架构使用 Cloud Run functions 实施一个简单的流水线，但建议不要在生产环境中使用 Cloud Run functions，因为 Document AI API 调用可能会超出 Cloud Run functions 支持的超时限制。建议使用 Cloud Tasks 来获取更强大的无服务器解决方案。

学习内容

在本实验中，您将学习如何完成以下操作：

• 启用 Document AI API。
• 部署使用 Document AI、BigQuery、Cloud Storage 和 Pub/Sub API 的 Cloud Run functions。

您将配置一个 Cloud Run 函数，以执行以下操作：

• 在将文档上传到 Cloud Storage 时触发。
• 使用 Python 版 Document AI 客户端库。
• 在创建 Pub/Sub 消息时触发。

设置和要求

点击“开始实验”按钮前的注意事项

请阅读以下说明。实验是计时的，并且您无法暂停实验。计时器在您点击开始实验后即开始计时，显示 Google Cloud 资源可供您使用多长时间。

此实操实验可让您在真实的云环境中开展实验活动，免受模拟或演示环境的局限。为此，我们会向您提供新的临时凭据，您可以在该实验的规定时间内通过此凭据登录和访问 Google Cloud。

为完成此实验，您需要：

• 能够使用标准的互联网浏览器（建议使用 Chrome 浏览器）。

注意：请使用无痕模式（推荐）或无痕浏览器窗口运行此实验。这可以避免您的个人账号与学生账号之间发生冲突，这种冲突可能导致您的个人账号产生额外费用。

• 完成实验的时间 - 请注意，实验开始后无法暂停。

注意：请仅使用学生账号完成本实验。如果您使用其他 Google Cloud 账号，则可能会向该账号收取费用。

激活 Cloud Shell

Cloud Shell 是一种装有开发者工具的虚拟机。它提供了一个永久性的 5GB 主目录，并且在 Google Cloud 上运行。Cloud Shell 提供可用于访问您的 Google Cloud 资源的命令行工具。

1. 点击 Google Cloud 控制台顶部的激活 Cloud Shell 。

2. 在弹出的窗口中执行以下操作：

   • 继续完成 Cloud Shell 信息窗口中的设置。
   • 授权 Cloud Shell 使用您的凭据进行 Google Cloud API 调用。

如果您连接成功，即表示您已通过身份验证，且项目 ID 会被设为您的 Project_ID 。输出内容中有一行说明了此会话的 Project_ID：

Your Cloud Platform project in this session is set to {{{project_0.project_id | "PROJECT_ID"}}}

gcloud 是 Google Cloud 的命令行工具。它已预先安装在 Cloud Shell 上，且支持 Tab 自动补全功能。

3. （可选）您可以通过此命令列出活跃账号名称：

gcloud auth list

4. 点击授权。

输出：

ACTIVE: * ACCOUNT: {{{user_0.username | "ACCOUNT"}}} To set the active account, run: $ gcloud config set account `ACCOUNT`

5. （可选）您可以通过此命令列出项目 ID：

gcloud config list project

输出：

[core] project = {{{project_0.project_id | "PROJECT_ID"}}} 注意：如需查看在 Google Cloud 中使用 gcloud 的完整文档，请参阅 gcloud CLI 概览指南。

任务 1. 启用 API 并创建 API 密钥

您必须为此实验启用 Document AI、Cloud Run functions、Cloud Build 和 Geocoding API，然后创建 Geocoding Cloud Run 函数所需的 API 密钥。

1. 点击 Google Cloud 控制台顶部的激活 Cloud Shell 。

2. 在 Cloud Shell 中，输入以下命令以启用实验所需的 API：

gcloud services enable documentai.googleapis.com gcloud services enable cloudfunctions.googleapis.com gcloud services enable cloudbuild.googleapis.com gcloud services enable geocoding-backend.googleapis.com

3. 在控制台的导航菜单 () 中，依次点击 API 和服务 > 凭证。

4. 选择创建凭证，然后从下拉菜单中选择 API 密钥。

“API 密钥已创建”对话框中会显示您新建的密钥。API 密钥是一个长字符串，包含大小写字母、数字和短划线。例如，a4db08b757294ea94c08f2df493465a1。

5. 点击操作下的三点状图标，然后在对话框中点击修改 API 密钥。

6. 在 API 限制部分中选择限制密钥，为新 API 密钥添加 API 限制。

7. 点击选择 API 下拉菜单，然后在过滤条件框中输入 Geocoding API。

8. 选择 Geocoding API，然后点击确定。

9. 点击保存按钮。

注意： 如果您在限制密钥下拉列表中找不到 Geocoding API，请刷新页面，以便更新可用 API 列表。 检查是否已启用所有必需的 API。

任务 2. 下载实验源代码

在此任务中，您会将源文件复制到 Cloud Shell 中。这些文件包括 Cloud Run 函数的源代码，以及您将在实验中创建的 BigQuery 表的架构。

1. 在 Cloud Shell 中，输入以下命令来下载本实验的源代码：

mkdir ./documentai-pipeline-demo gcloud storage cp -r \ gs://spls/gsp927/documentai-pipeline-demo/* \ ~/documentai-pipeline-demo/

任务 3. 创建表单处理器

使用专业的 Document AI 表单解析器，在 Document AI Platform 上创建一个通用表单处理器实例。通用表单处理器可处理各种类型的文档并提取文档中所有可识别的文本内容。除了打印的文本，它还可以处理手写的和沿任意方向排列的文本，并支持多种语言，理解表单数据元素之间的关联，因此可以帮助您从包含文本标签的表单字段中提取键值对。

1. 在 Google Cloud 控制台的搜索栏中，输入 Document AI，然后点击产品和页面中的相应结果。

2. 点击探索处理器，然后点击 Form Parser（表单解析器）以创建新处理器。

3. 将处理器的名称指定为 ocr-processor 并从地区列表中选择 US（美国）。

4. 点击创建以创建处理器。

在本实验的后面部分，您将使用此处理器的 ID 和位置配置 Cloud Run 函数，以便 Cloud Run 函数使用此特定处理器来处理示例账单。

任务 4. 创建 Cloud Storage 存储桶和 BigQuery 数据集

在本部分，您将创建文档处理流水线所需的 Google Cloud 资源，从而准备好环境。

创建输入、输出和归档 Cloud Storage 存储桶

为文档处理流水线创建输入、输出和归档 Cloud Storage 存储桶。

1. 在 Cloud Shell 中，输入以下命令，为实验创建 Cloud Storage 存储桶：

export PROJECT_ID=$(gcloud config get-value core/project) export BUCKET_LOCATION="{{{my_primary_project.default_region|REGION}}}" gsutil mb -c standard -l ${BUCKET_LOCATION} -b on \ gs://${PROJECT_ID}-input-invoices gsutil mb -c standard -l ${BUCKET_LOCATION} -b on \ gs://${PROJECT_ID}-output-invoices gsutil mb -c standard -l ${BUCKET_LOCATION} -b on \ gs://${PROJECT_ID}-archived-invoices

创建 BigQuery 数据集和表。

创建 BigQuery 数据集和数据处理流水线所需的三个输出表。

1. 在 Cloud Shell 中，输入以下命令，为实验创建 BigQuery 表：

bq --location="US" mk -d \ --description "Form Parser Results" \ ${PROJECT_ID}:invoice_parser_results cd ~/documentai-pipeline-demo/scripts/table-schema/ bq mk --table \ invoice_parser_results.doc_ai_extracted_entities \ doc_ai_extracted_entities.json bq mk --table \ invoice_parser_results.geocode_details \ geocode_details.json

您可以在 Cloud 控制台中前往 BigQuery，并使用 BigQuery SQL 工作区检查 invoice_parser_results 数据集中表的架构。

创建 Pub/Sub 主题

初始化 Pub/Sub 主题，该主题用于触发处理流水线中的 Geocoding API 数据丰富化操作。

1. 在 Cloud Shell 中，输入以下命令，为实验创建 Pub/Sub 主题：

export GEO_CODE_REQUEST_PUBSUB_TOPIC=geocode_request gcloud pubsub topics \ create ${GEO_CODE_REQUEST_PUBSUB_TOPIC} 检查 BigQuery 数据集、Cloud Storage 存储桶和 Pub/Sub 主题是否已创建。

任务 5. 创建 Cloud Run 函数

创建两个 Cloud Run 函数，以供数据处理流水线用于处理上传到 Cloud Storage 的账单。这些函数使用 Document AI API 从原始文档中提取表单数据，然后使用 GeoCode API 检索从文档中提取的地址信息的地理定位数据。

您可以使用代码编辑器或您选择的任何其他编辑器来检查两个 Cloud Run 函数的源代码。Cloud Run 函数存储在 Cloud Shell 中的以下文件夹中：

• Process Invoices - scripts/cloud-functions/process-invoices
• Geocode Addresses - scripts/cloud-functions/geocode-addresses

当文件上传到您之前创建的输入文件存储桶时，主 Cloud Run 函数 process-invoices 会被触发。

函数文件夹 scripts/cloud-functions/process-invoices 包含用于创建 process-invoices Cloud Run 函数的两个文件。

requirements.txt 文件指定函数所需的 Python 库。这包括 Document AI 客户端库，以及 Python 代码所需的其他 Google Cloud 库，这些库可用于从 Cloud Storage 读取文件，将数据保存到 BigQuery，以及向 Pub/Sub 写入消息，进而触发解决方案流水线中的其余函数。

main.py Python 文件包含 Cloud Run 函数代码，它用于创建 Document AI、BigQuery 和 Pub/Sub API 客户端，以及以下用于处理文档的内部函数：

• write_to_bq - 将字典对象写入 BigQuery 表。请注意，在调用此函数之前，您必须确保架构有效。
• get_text - 将表单名称和值文本锚点映射到文档中的扫描文本。这样，该函数就能识别表单中的特定元素（例如供应商名称和地址），并提取相关值。一个专业的 Document AI 处理器可以直接在实体属性中提供这种上下文信息。
• process_invoice - 使用异步 Document-AI 客户端 API 从 Cloud Storage 读取和处理文件，如下所述：
  • 创建异步请求，以处理触发 Cloud Run 函数调用的文件。
  • 处理表单数据以提取账单字段，仅将预定义架构中的特定字段存储在字典中。
  • 发布 Pub/Sub 消息，以使用从文档中提取的地址表单数据来触发地理编码 Cloud Run 函数。
  • 将表单数据写入 BigQuery 表。
  • 删除异步 Document AI API 调用生成的中间（输出）文件。
  • 将输入文件复制到归档存储桶中。
  • 删除已处理的输入文件。

process_invoices Cloud Run 函数只会处理检测到具有以下表单字段名称的表单数据：

• input_file_name
• address
• supplier
• invoice_number
• purchase_order
• date
• due_date
• subtotal
• tax
• total

另一个 Cloud Run 函数 geocode-addresses 在 Pub/Sub 主题上收到新消息时触发，并从 Pub/Sub 消息中提取其参数数据。

创建 Cloud Run 函数来处理上传到 Cloud Storage 的文档

创建一个 Cloud Run 函数，该函数使用 Document AI 表单处理器来解析已上传到 Cloud Storage 存储桶的表单文档。

1. 运行以下命令，获取项目的 Cloud Storage 服务代理的电子邮件地址：

gcloud storage service-agent --project=$PROJECT_ID

2. 运行以下命令，向 Cloud Storage 服务账号授予所需的权限：

PROJECT_NUMBER=$(gcloud projects describe $PROJECT_ID --format="value(projectNumber)") gcloud iam service-accounts create "service-$PROJECT_NUMBER" \ --display-name "Cloud Storage Service Account" || true gcloud projects add-iam-policy-binding $PROJECT_ID \ --member="serviceAccount:service-$PROJECT_NUMBER@gs-project-accounts.iam.gserviceaccount.com" \ --role="roles/pubsub.publisher" gcloud projects add-iam-policy-binding $PROJECT_ID \ --member="serviceAccount:service-$PROJECT_NUMBER@gs-project-accounts.iam.gserviceaccount.com" \ --role="roles/iam.serviceAccountTokenCreator" 注意：如果 Cloud Storage 服务账号已存在，您可以忽略错误。

3. 创建 Invoice Processor Cloud Run 函数：

cd ~/documentai-pipeline-demo/scripts export CLOUD_FUNCTION_LOCATION="{{{my_primary_project.default_region|REGION}}}" gcloud functions deploy process-invoices \ --no-gen2 \ --region=${CLOUD_FUNCTION_LOCATION} \ --entry-point=process_invoice \ --runtime=python39 \ --source=cloud-functions/process-invoices \ --timeout=400 \ --env-vars-file=cloud-functions/process-invoices/.env.yaml \ --trigger-resource=gs://${PROJECT_ID}-input-invoices \ --trigger-event=google.storage.object.finalize 注意：如果此命令因权限错误而失败，请稍等片刻，然后重试。

创建 Cloud Run 函数，以根据地址查找地理编码数据

创建 Cloud Run 函数，该函数接受来自 Pub/Sub 消息的地址数据，并使用 Geocoding API 精确定位地址。

1. 创建地理编码 Cloud Run 函数：

cd ~/documentai-pipeline-demo/scripts gcloud functions deploy geocode-addresses \ --no-gen2 \ --region=${CLOUD_FUNCTION_LOCATION} \ --entry-point=process_address \ --runtime=python39 \ --source=cloud-functions/geocode-addresses \ --timeout=60 \ --env-vars-file=cloud-functions/geocode-addresses/.env.yaml \ --trigger-topic=${GEO_CODE_REQUEST_PUBSUB_TOPIC}

任务 6. 修改 Cloud Run 函数的环境变量

在此任务中，您将通过 Cloud 控制台修改每个函数的环境变量，以最终确定配置，使其反映您要在实验中使用的特定参数。

修改 process-invoices Cloud Run 函数的环境变量

为 process-invoices 函数设置 Cloud Run 函数环境变量。

1. 在 Cloud 控制台的搜索栏中，输入 Cloud Run functions，然后点击产品和页面中的相应结果。

系统会重定向到 Cloud Run 控制台，点击前往 Cloud Run functions（第 1 代）函数列表页面，查看已部署的函数 process-invoices 和 geocode-addresses。

注意：如果您没有看到前往 Cloud Run functions（第 1 代）函数列表页面的链接，请刷新 Cloud Run 控制台。

2. 点击 Cloud Run 函数 process-invoices，打开其管理页面。
3. 点击修改。
4. 点击运行时、构建、连接和安全设置以展开该部分。
5. 在运行时环境变量下，添加 GCP_PROJECT 变量，并添加与您的项目 ID 匹配的值。
6. 在运行时环境变量下，更新 PROCESSOR_ID 值，使其与您之前创建的账单处理器 ID 相匹配。
7. 在运行时环境变量下，更新 PARSER_LOCATION 值，使其与您之前创建的账单处理器的区域相匹配。此值将为 us 或 eu。此参数必须采用小写字母。
8. 点击下一步，选择 .env.yaml，然后再次更新 PROCESSOR_ID、PARSER_LOCATION 和 GCP_PROJECT 值，以用于您的账单处理器。

9. 点击部署。

部署 Process Invoices Cloud Run 函数

修改 geocode-addresses Cloud Run 函数的环境变量

为 GeoCode 数据丰富化函数设置 Cloud Run 函数环境变量。

1. 点击 Cloud Run 函数 geocode-addresses，打开其管理页面。
2. 点击修改。
3. 点击运行时、构建、连接和安全设置以展开该部分。
4. 在运行时环境变量下，更新 API_key 值，使其与任务 1 中创建的 API 密钥值相匹配。
5. 点击下一步，选择 .env.yaml，然后更新 API_key 值，使其与您在上一步中设置的 API 密钥值相匹配。
6. 点击部署。

部署 Geocode Addresses Cloud Run 函数

任务 7. 测试并验证端到端解决方案

将测试数据上传到 Cloud Storage，并随着文档被处理和提取的数据得到增强，监控整个流水线的进度。

1. 在 Cloud Shell 中，输入以下命令，将示例表单上传到 Cloud Storage 存储桶，这将触发 process-invoices Cloud Run 函数：

export PROJECT_ID=$(gcloud config get-value core/project) gsutil cp gs://spls/gsp927/documentai-pipeline-demo/sample-files/* gs://${PROJECT_ID}-input-invoices/

2. 在 Cloud 控制台的搜索栏中，输入 Cloud Run functions，然后点击产品和页面中的相应结果。
3. 点击 Cloud Run 函数 process-invoices，打开其管理页面。
4. 点击日志。

您将看到与创建函数相关的事件，以及为配置环境变量而进行的更新，接着是一些事件，它们会显示有关正在处理的文件和 Document AI 检测到的数据的详细信息。

继续查看事件，直到您看到表明函数已执行完成的最终事件，即函数已完成 LoadJob。如果报告了错误，请仔细检查在上一部分介绍的 .env.yaml 文件中设置的参数是否正确。特别是，您要确保处理器 ID、位置和项目 ID 有效。事件列表不会自动刷新。

处理完成后，如果 Document AI 处理器在上传的文档中检测到地址数据，则 BigQuery 表中将填充 Document AI 提取的实体以及 Geocoding API 提供的丰富化数据。

打开 BigQuery 控制台

1. 在 Google Cloud 控制台中，选择导航菜单 > BigQuery。

您会看到欢迎在 Cloud 控制台中使用 BigQuery 消息框，其中提供了指向快速入门指南和版本说明的链接。

2. 点击完成。

BigQuery 控制台即会打开。

3. 在探索器中展开您的项目 ID。

4. 展开 invoice_parser_results。

5. 选择 doc_ai_extracted_entities，然后点击预览。

您可以看到账单处理器从账单中提取的表单信息，以及检测到的地址信息和供应商名称。

6. 选择 geocode_details，然后点击预览。

您可以看到，对于每个已处理的账单，如果其中包含 Document AI 能够提取的地址数据，则会显示格式化的地址、纬度和经度。

检查端到端流水线是否已处理表单和地址数据。

恭喜！

您已成功使用 Document AI API 和其他 Google Cloud 服务构建了端到端账单处理流水线。在本实验中，您启用了 Document AI API，部署了使用 Document AI、BigQuery、Cloud Storage 和 Pub/Sub API 的 Cloud Run 函数，并配置了在将文档上传到 Cloud Storage 时触发的 Cloud Run 函数。您还配置了一个 Cloud Run 函数，以使用 Document AI Python 客户端库，并在接收到 Pub/Sub 消息时自动触发。

后续步骤/了解详情

• 如需详细了解此身份验证形式，请参阅指南。

Google Cloud 培训和认证

…可帮助您充分利用 Google Cloud 技术。我们的课程会讲解各项技能与最佳实践，可帮助您迅速上手使用并继续学习更深入的知识。我们提供从基础到高级的全方位培训，并有点播、直播和虚拟三种方式选择，让您可以按照自己的日程安排学习时间。各项认证可以帮助您核实并证明您在 Google Cloud 技术方面的技能与专业知识。

上次更新手册的时间：2025 年 8 月 26 日

上次测试实验的时间：2025 年 7 月 29 日

版权所有 2025 Google LLC 保留所有权利。Google 和 Google 徽标是 Google LLC 的商标。其他所有公司名和产品名可能是其各自相关公司的商标。