Cloud Storage 注入流水线

本文档介绍了 Cloud Storage 注入流水线的功能以及如何运行该流水线。

Cloud Storage 注入流水线有何作用？

您的用户可以将文档从 Cloud Storage 转移到 Document AI Warehouse。用户可以在其中选择处理这些数据，以用于搜索、文档管理工作流等任务，或者只是测试 Document AI 输出。

为什么要使用此流水线？

如果必须提取大量文档（无论是否需要处理），此流水线都能提供可靠的工作流。它还有助于用户加快 Document AI Warehouse 客户的概念验证或生产工作负载的准备时间。

流水线功能

一般来说，Cloud Storage 注入流水线支持以下操作：

将原始文档注入 Document AI Warehouse。

{
  name: "projects/PROJECT_NUMBER/locations/LOCATION_ID",
  gcs_ingest_pipeline: {
    input_path: "gs://BUCKET_NAME/FOLDER_NAME",
    schema_name: "projects/PROJECT_NUMBER/locations/LOCATION_ID/documentSchemas/DOCUMENT_SCHEMA_ID"
  }
}

将 Document AI 处理的文档注入到 Document AI Warehouse 中。

{
  name: "projects/PROJECT_NUMBER/locations/LOCATION_ID",
  gcs_ingest_pipeline: {
    input_path: "gs://BUCKET_NAME/FOLDER_NAME",
    schema_name: "projects/PROJECT_NUMBER/locations/LOCATION_ID/documentSchemas/DOCUMENT_SCHEMA_ID",
    processor_type: "PROCESS_TYPE"
  }
}

processor_type 是必需的，用于表明输入文件已由 Document AI 处理。processor_type（OCR_PROCESSOR、INVOICE_PROCESSOR、FORM_PARSER_PROCESSOR）可在此处的在 API 中输入字段下找到。

在同一请求中提取 Document AI 处理的文档和相应的原始文档。

{
  name: "projects/PROJECT_NUMBER/locations/LOCATION_ID",
  gcs_ingest_pipeline: {
    input_path: "gs://BUCKET_NAME/FOLDER_NAME",
    schema_name: "projects/PROJECT_NUMBER/locations/LOCATION_ID/documentSchemas/DOCUMENT_SCHEMA_ID",
    processor_type: "PROCESS_TYPE"
  }
}

input_path 位置必须包含 Document AI 处理的文档。流水线会查找该 URI 路径下每个 Doc AI 处理的文档对应的原始文档。

提取一种类型的原始文档，并在流水线中触发 Document AI 处理。

{
  name: "projects/PROJECT_NUMBER/locations/LOCATION_ID",
  gcs_ingest_with_doc_ai_processors_pipeline: {
    input_path: "gs://BUCKET_NAME/FOLDER_NAME",
    extract_processor_infos: {
      processor_name: "projects/PROJECT_NUMBER/locations/LOCATION_ID/processors/PROCESSOR_ID",
      schema_name: "projects/PROJECT_NUMBER/locations/LOCATION_ID/documentSchemas/DOCUMENT_SCHEMA_ID"
    },
    processor_results_folder_path: "gs://OUTPUT_BUCKET_NAME/OUTPUT_BUCKET_FOLDER_NAME"
  }
}

extract_processor_infos 字段只能包含一个提取处理器。input_path 字段下的所有文档都被视为一种文档类型，并由同一提取处理器进行处理。

提取多种类型的原始文档，并在流水线中触发 Document AI 处理。

{
  name: "projects/PROJECT_NUMBER/locations/LOCATION_ID",
  gcs_ingest_with_doc_ai_processors_pipeline: {
    input_path: "gs://BUCKET_NAME/FOLDER_NAME",
    split_classify_processor_info: {
      processor_name: "projects/PROJECT_NUMBER/locations/LOCATION_ID/processors/PROCESSOR_ID"
    },
    extract_processor_infos: [
      {
        processor_name: "projects/PROJECT_NUMBER/locations/LOCATION_ID/processors/PROCESSOR_ID",
        document_type: "DOCUMENT_TYPE",
        schema_name: "projects/PROJECT_NUMBER/locations/LOCATION_ID/documentSchemas/DOCUMENT_SCHEMA_ID"
      }
    ],
    processor_results_folder_path: "gs://OUTPUT_BUCKET_NAME/OUTPUT_BUCKET_FOLDER_NAME"
  }
}

split_classify_processor_info 字段按类型对文档进行分类。extract_processor_infos 字段（以数组形式传递）必须包含与分类结果中每种文档类型对应的不同提取处理器。

提取配置

Cloud Storage 注入流水线支持在文档注入期间进行可选的自定义，如下列选项所述：

使用 skip_ingested_documents 标志可在流水线多次在同一 Cloud Storage 文件夹上触发时跳过已提取的文档。如果 Cloud Storage 中的文档包含 status=ingested 元数据（管道的状态跟踪功能），则启用此标志后，系统不会重新提取文档。
使用 document_acl_policy 标志可在创建文档期间提供额外的文档级 ACL 政策。此标志适用于在 Document AI Warehouse 中启用了文档级 ACL 的项目。
使用 enable_document_text_extraction 标志可让 Document AI Warehouse 在原始文档包含内容时提取其中的文本。此标志不同于 Document AI 处理，并且文档提取仅支持 Document AI Warehouse 中的以下文档文件类型。
- RAW_DOCUMENT_FILE_TYPE_TEXT
- RAW_DOCUMENT_FILE_TYPE_DOCX
使用 folder 字段指定提取的文档的目标文件夹。所有提取的文档都将链接到指定的父文件夹下。
使用 cloud_function 字段可在文档 proto 文件进入 Document AI Warehouse 之前进一步自定义其中的字段。cloud_function 值必须是 Document AI Warehouse 服务账号可访问的有效网址。每次通话时长不得超过 5 分钟。请求和响应采用 JSON 格式。您可能会在请求载荷中找到以下键：
- display_name
- properties
- plain_text 或 cloud_ai_document.text
- reference_id
- document_schema_name
- raw_document_path
- raw_document_file_type
响应载荷中的键会作为 proto 的一部分被提取到 Document AI Warehouse 中。如果修改了某个键或向响应中添加了某个键，则会覆盖原始值。系统会舍弃响应中的额外键。如果相应值无效，则文档提取会失败。
- display_name
- properties
- plain_text 或 cloud_ai_document.text
- reference_id
- document_acl_policy
- folder

使用 pipeline_config 实现提取流水线自定义。

{
  name: "projects/PROJECT_NUMBER/locations/LOCATION_ID",
  gcs_ingest_pipeline: {
    input_path: "gs://BUCKET_NAME/FOLDER_NAME",
    schema_name: "projects/PROJECT_NUMBER/locations/LOCATION_ID/documentSchemas/DOCUMENT_SCHEMA_ID",
    skip_ingested_documents: "true",
    pipeline_config: {
      enable_document_text_extraction: "true"
      folder: "projects/PROJECT_NUMBER/locations/LOCATION_ID/documents/FOLDER_DOCUMENT_ID",
      cloud_function: "https://REGION-PROJECT_ID.cloudfunctions.net/CLOUD_FUNCTION_NAME"
    }
  }
}

状态跟踪

状态跟踪功能可显示整个流水线和每个文档的提取结果进度。

您可以通过检查 LRO 是否已完成来查看流水线是否已完成。
您可以通过查看 Cloud Storage 元数据来检查流水线中每个文档的状态。Cloud Storage 会在每个文档上创建以下键值对。
- 键：status；值：status=queued 或 status=processed 或 status=ingested 或 status=failed。
- 键：error；值：the error message。

支持的文档类型

流水线中支持的文档类型与 Document AI Warehouse 支持的文档类型相同：文本、PDF、图片（扫描的 PDF、TIFF 文件、JPEG 文件）、Microsoft Office 文件（DOCX、PPTX、XSLX）。

来自 Cloud Storage 元数据的文档属性

您可以使用流水线中的 Cloud Storage 元数据在 Document AI Warehouse 中创建自定义属性。如果 Cloud Storage 文档的元数据值的键与架构中的属性名称匹配，则该元数据值将自动复制到相应的 Document AI Warehouse 文档属性。

运行流水线

不同的参数用于触发 Cloud Storage 注入流水线中的不同功能。如需了解详情，请参阅方法：projects.locations.runPipeline。

以下部分提供了两个触发 Cloud Storage 注入流水线的示例。

运行不含 Document AI 处理器的 Cloud Storage 注入流水线。

REST

curl --location --request POST 'https://contentwarehouse.googleapis.com/v1/projects/PROJECT_NUMBER/locations/LOCATION_ID:runPipeline' \
--header 'Content-Type: application/json' \
--header "Authorization: Bearer ${AUTH_TOKEN}" \
--data '{
        "name": "projects/PROJECT_NUMBER/locations/LOCATION_ID",
        "gcs_ingest_pipeline": {
            "input_path": "gs://BUCKET_NAME/FOLDER_NAME/",
"schema_name": "projects/PROJECT_NUMBER/locations/
LOCATION_ID/documentSchemas/DOCUMENT_SCHEMA_ID",
            "skip_ingested_documents": "true"
        },
        "request_metadata": {
            "user_info": {
                "id": "user:USER EMAIL ADDRESS"
            }
        }
}'

使用 Document AI 处理器运行 Cloud Storage 提取流水线。

REST

curl --location --request POST 'https://contentwarehouse.googleapis.com/v1/projects/PROJECT_NUMBER/locations/LOCATION_ID:runPipeline' \
--header 'Content-Type: application/json' \
--header "Authorization: Bearer ${AUTH_TOKEN}" \
--data '{
        "name": "projects/PROJECT_NUMBER/locations/LOCATION_ID",
        "gcs_ingest_with_doc_ai_processors_pipeline": {
            "input_path": "gs://BUCKET_NAME/FOLDER_NAME/",
            "split_classify_processor_info": {
              "processor_name": "projects/PROJECT_NUMBER/locations/LOCATION_ID/processors/PROCESSOR_ID"
            },
            "extract_processor_infos": [
              {
                "processor_name": "projects/PROJECT_NUMBER/locations/LOCATION_ID/processors/PROCESSOR_ID",
                "document_type": "DOCUMENT_TYPE",
                "schema_name": "projects/PROJECT_NUMBER/locations/LOCATION_ID/documentSchemas/DOCUMENT_SCHEMA_ID"
              }
            ],
            "processor_results_folder_path": "gs://OUTPUT_BUCKET_NAME/OUTPUT_BUCKET_FOLDER_NAME/"
            "skip_ingested_documents": "true"
        },
        "request_metadata": {
            "user_info": {
                "id": "user:USER EMAIL ADDRESS"
            }
        }
}'

此命令会返回长时间运行的操作的资源名称。有了此资源名称，您就可以按照下一步骤跟踪流水线的进度。

获取长时间运行的操作的结果

REST

curl --location --request GET 'https://contentwarehouse.googleapis.com/v1/projects/PROJECT_NUMBER/locations/LOCATION/operations/OPERATION' \
--header "Authorization: Bearer ${AUTH_TOKEN}"

后续步骤

如需查看已提取的文档，请前往 Document AI Warehouse 的网络应用。

Cloud Storage 注入流水线 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。