# Build a document processing pipeline with Workflows

If you are using Document AI to build workflows that involve several steps to process documents, you might want to use serverless orchestration to manage your workflow.

[Workflows](https://docs.cloud.google.com/workflows) links a series of tasks together in an order you define. It requires no infrastructure management and scales seamlessly on demand, including scaling down to zero. With its pay-per-use pricing model, you only pay for execution time.

Workflows provides a built-in [connector for the Document AI API](https://docs.cloud.google.com/workflows/docs/reference/googleapis/documentai/Overview). Connectors don't require additional code aside from defining the steps using a config file. Connectors handle the formatting of requests, hiding the details of the API. Connectors also provide built-in behavior for authentication, handling retries and [long-running operations](https://docs.cloud.google.com/workflows/docs/connectors#lro).

For more information, see [Understand connectors](https://docs.cloud.google.com/workflows/docs/connectors).

This guide processes a set of sample invoices using the [Invoice Parser](https://docs.cloud.google.com/document-ai/docs/processors-list#processor_invoice-processor) processor with [batch processing](https://docs.cloud.google.com/document-ai/docs/send-request#batch-process)

*** ** * ** ***

To follow step-by-step guidance for this task directly in the
Google Cloud console, click **Guide me**:

[Guide me](https://console.cloud.google.com/ai/document-ai?tutorial=document-ai--documentai_workflows)

*** ** * ** ***

## Before you begin

<br />

### Required roles


To get the permissions that
you need to create a custom classifier,

ask your administrator to grant you the
following IAM roles on your project:

- [Document AI Administrator](https://docs.cloud.google.com/iam/docs/roles-permissions/clouddocumentai#documentai.admin) (`roles/documentai.admin`)
- [Project IAM Admin](https://docs.cloud.google.com/iam/docs/roles-permissions/resourcemanager#resourcemanager.projectIamAdmin) (`roles/resourcemanager.projectIamAdmin`)
- [Service Account User](https://docs.cloud.google.com/iam/docs/roles-permissions/iam#iam.serviceAccountUser) (`roles/iam.serviceAccountUser`)
- [Storage Admin](https://docs.cloud.google.com/iam/docs/roles-permissions/storage#storage.admin) (`roles/storage.admin`)
- [Workflows Editor](https://docs.cloud.google.com/iam/docs/roles-permissions/workflows#workflows.editor) (`roles/workflows.editor`)
- Create a new service account: [Create Service Accounts](https://docs.cloud.google.com/iam/docs/roles-permissions/iam#iam.serviceAccountCreator) (`roles/iam.serviceAccountCreator`)


For more information about granting roles, see [Manage access to projects, folders, and organizations](https://docs.cloud.google.com/iam/docs/granting-changing-revoking-access).


You might also be able to get
the required permissions through [custom
roles](https://docs.cloud.google.com/iam/docs/creating-custom-roles) or other [predefined
roles](https://docs.cloud.google.com/iam/docs/roles-overview#predefined).

> [!NOTE]
> **Note:** If you want to access input files stored in a different project, then you need to grant additional roles to the Document AI service agent. For more information, see [Cross project file access
> setup](https://docs.cloud.google.com/document-ai/docs/setup#cross_project_file_access_setup).

## Create an Invoice Parser

Use the Google Cloud console to create a new Invoice Parser. Refer to the [creating and managing processors](https://docs.cloud.google.com/document-ai/docs/create-processor) page for more information.

1. In the Google Cloud console navigation menu, click **Document AI** and
   select **Processor Gallery**.

   [Processor
   Gallery](https://console.cloud.google.com/ai/document-ai/processor-library)
2. In the **Processor Gallery** ,
   search for
   **Invoice Parser** and select **Create**.

   ![Create invoice
   processor](https://docs.cloud.google.com/static/document-ai/docs/images/create-invoice-parser.png)
3. Enter a **Processor name** , such as `workflows-invoice-processor`.

4. Select the region (`location` in the Document AI API) to host the processor.

5. Click **Create** . The **Processor Details** tab appears.

6. Copy your **Processor ID**. You must use this in your code later.


   ![Processor ID](https://docs.cloud.google.com/static/document-ai/docs/images/workflows/GetProcessorID.png)

## Configure output bucket in Cloud Storage

When using [batch (asynchronous) processing](https://docs.cloud.google.com/document-ai/docs/send-request#batch-process), the input and output files must be stored in Cloud Storage. The following bucket contains sample files to use as input for this guide:

    gs://cloud-samples-data/documentai/workflows

Configure a bucket to store the output files in your project.

1. Follow the [Create Buckets](https://docs.cloud.google.com/storage/docs/creating-buckets) guide in the Cloud Storage documentation to create a bucket, or use a bucket that already exists.

2. Create a [folder](https://docs.cloud.google.com/storage/docs/objects#simulated-folders) in the bucket (for example, `docai-workflows-output`).

## Create workflow

1. Open the **Workflows dashboard** in the Google Cloud console.


   [Go to the Workflows dashboard](https://console.cloud.google.com/workflows)
2. Click **Create** to create your first workflow. The **Create Workflow** page appears.


   ![Workflows Create](https://docs.cloud.google.com/static/document-ai/docs/images/workflows/WorkflowsCreate.png)
3. Rename the workflow to `docai-workflows-quickstart`.

4. Add a description, and select a service account if you have one for this project.

   1. If you don't have a service account, select **Create New Service Account** .
      - Name your service account, and select **Create and Continue** . ![CreateServiceAccount](https://docs.cloud.google.com/static/document-ai/docs/images/workflows/CreateServiceAccount.png)
      - Give it the following roles:
      - `Document AI API User`
      - `Workflows Invoker`
      - `Logs Writer`
      - `Service Usage Consumer`
      - ![ServiceAccountRoles](https://docs.cloud.google.com/static/document-ai/docs/images/workflows/ServiceAccountRoles.png)
5. On the **Create Workflow** page, click **Next**.

## Define workflow

1. In the **Workflows editor**, clear the default YAML and paste in the code below:

       main:
         params: [input]
         steps:
           - start:
               call: sys.log
               args:
                 text: ${input}
           - vars:
               assign:
                 - input_gcs_bucket: ${input.inputBucket}
                 - output_gcs_bucket: ${input.outputBucket}
                 - processor_id: ${input.processorId}
                 - location: ${input.location}
                 - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
           - batch_process:
               call: googleapis.documentai.v1.projects.locations.processors.batchProcess
               args:
                 name: ${"projects/" + project_id + "/locations/" + location + "/processors/" + processor_id}
                 location: ${location}
                 body:
                   inputDocuments:
                     gcsPrefix:
                       gcsUriPrefix: ${input_gcs_bucket}
                   documentOutputConfig:
                     gcsOutputConfig:
                       gcsUri: ${output_gcs_bucket}
                   skipHumanReview: true
               result: batch_process_resp
           - return:
               return: ${batch_process_resp}

![WorkflowsEditor](https://docs.cloud.google.com/static/document-ai/docs/images/workflows/WorkflowsEditor.png)

1. Select **Deploy**.

## Test workflow

Now that the workflow has been deployed, you should be on the **Workflow details** page.

1. Click **Execute** at the top of the page to go to the **Execute workflow** page. This page allows you to test your workflow in the console.

2. Replace the placeholder variables in the following input JSON:

   - <var translate="no">OUTPUT_BUCKET_PATH</var>: The full path to output bucket and folder created earlier.
   - <var translate="no">PROCESSOR_ID</var>: The processor ID for the processor created earlier.
   - <var translate="no">LOCATION</var>: The region for the processor created earlier.

       {
         "inputBucket": "gs://cloud-samples-data/documentai/workflows",
         "outputBucket": "OUTPUT_BUCKET_PATH",
         "processorId": "PROCESSOR_ID",
         "location": "LOCATION"
       }

3. Copy/Paste the JSON into the **input textbox**.

4. Click **Execute** at the bottom of the workflow to test your input and wait for a moment for the process to complete.

5. Once the workflow finishes, the output JSON should look something like this.

       {
         "done": true,
         "metadata": {
           "@type": "type.googleapis.com/google.cloud.documentai.v1.BatchProcessMetadata",
           "createTime": "2023-02-24T18:31:18.521764Z",
           "individualProcessStatuses": [
             {
               "humanReviewStatus": {
                 "state": "SKIPPED"
               },
               "inputGcsSource": "gs://cloud-samples-data/documentai/workflows/baking_technologies_invoice.pdf",
               "outputGcsDestination": "OUTPUT_BUCKET_PATH/11119195367705871803/0",
               "status": {}
             },
             {
               "humanReviewStatus": {
                 "state": "SKIPPED"
               },
               "inputGcsSource": "gs://cloud-samples-data/documentai/workflows/generic_invoice.pdf",
               "outputGcsDestination": "OUTPUT_BUCKET_PATH/11119195367705871803/1",
               "status": {}
             },
             {
               "humanReviewStatus": {
                 "state": "SKIPPED"
               },
               "inputGcsSource": "gs://cloud-samples-data/documentai/workflows/google_invoice.pdf",
               "outputGcsDestination": "OUTPUT_BUCKET_PATH/11119195367705871803/2",
               "status": {}
             }
           ],
           "state": "SUCCEEDED",
           "updateTime": "2023-02-24T18:31:52.543163Z"
         },
         "name": "projects/PROJECT_ID/locations/LOCATION/operations/11119195367705871803",
         "response": {
           "@type": "type.googleapis.com/google.cloud.documentai.v1.BatchProcessResponse"
         }
       }

6. Open your output storage bucket, you will see the output for each invoice processed in a folder named after the Operation ID.

   ![BatchGCSOutput](https://docs.cloud.google.com/static/document-ai/docs/images/workflows/BatchGCSOutput.png)

7. Within this folder there will be subfolders, one for each document processed from the input batch containing [`Document`](https://docs.cloud.google.com/document-ai/docs/reference/rest/v1/Document) JSON file(s).

8. You may inspect each JSON file or write code to use the relevant fields for your use case. Refer to the [handle the processing response guide](https://docs.cloud.google.com/document-ai/docs/handle-response) for more information.

9. You can now link additional steps in the workflow to further process the output, such as inserting the extracted entities into a database for analytics.

10. Refer to the [Workflows Documentation](https://docs.cloud.google.com/workflows/docs/overview) for more information on how to trigger workflows.

## Clean up


To avoid incurring charges to your Google Cloud account for
the resources used on this page, follow these steps.

To avoid unnecessary Google Cloud charges, use the
[Google Cloud console](https://console.cloud.google.com/) to delete your processor, bucket, workflow, and project if you do not need
them.

## What's next

- Learn more about [Workflows](https://docs.cloud.google.com/workflows/docs/overview)
- Learn more about the [Document AI Workflows Connector](https://docs.cloud.google.com/workflows/docs/reference/googleapis/documentai/Overview)
- See a list and descriptions of currently [available processors](https://docs.cloud.google.com/document-ai/docs/processor-overview).
- Learn how to [handle the processing response](https://docs.cloud.google.com/document-ai/docs/handle-response).