Cloud Function中Document AI批量处理超时问题及触发后续函数方案咨询

针对Cloud Function（xyz）因等待Document AI批量处理超时的问题，推荐采用异步通知+事件驱动的架构，确保批量处理独立完成后自动触发Cloud Function（abc），同时传递处理结果信息。以下是具体实现方案：

核心思路 #

1. Cloud Function（xyz）仅负责启动Document AI批量处理任务，无需等待任务完成，避免超时。
2. 配置Document AI批量处理任务，在完成后通过Cloud Pub/Sub发送通知。
3. 将Cloud Function（abc）设置为Pub/Sub主题的订阅触发器，接收完成通知后处理结果。

步骤1：修改Cloud Function（xyz）代码，添加完成通知配置 #

移除原代码中同步等待的operation.result()，新增Pub/Sub通知配置，让Document AI在批量处理完成后自动发送消息到指定主题：

from typing import Optional
from google.api_core.client_options import ClientOptions
from google.cloud import documentai

def batch_process_documents(
    gcs_input_uri: str,
):
    project_id = "tally-ai-394207"
    location = "us"
    processor_id = "d5efc50f9221c006"  # invoice parser
    gcs_output_uri = "gs://document-ai-output-bucket-haritha/ManishoOutput/"  # Must end with a '/'
    pubsub_topic_id = "document-ai-batch-completion-topic"  # 替换为你的Pub/Sub主题ID
    
    # Set the API endpoint based on the location
    api_endpoint = f"{location}-documentai.googleapis.com"
    client_options = ClientOptions(api_endpoint=api_endpoint)
    client = documentai.DocumentProcessorServiceClient(client_options=client_options)

    # Specify a GCS URI Prefix to process an entire directory
    gcs_prefix = documentai.GcsPrefix(gcs_uri_prefix=gcs_input_uri)
    input_config = documentai.BatchDocumentsInputConfig(gcs_prefix=gcs_prefix)

    # Cloud Storage URI for the Output Directory
    gcs_output_config = documentai.DocumentOutputConfig.GcsOutputConfig(
        gcs_uri=gcs_output_uri
    )

    # Where to write results
    output_config = documentai.DocumentOutputConfig(gcs_output_config=gcs_output_config)

    # 配置批量处理完成后的Pub/Sub通知
    notification_config = documentai.NotificationConfig(
        pubsub_topic=f"projects/{project_id}/topics/{pubsub_topic_id}"
    )

    name = client.processor_path(project_id, location, processor_id)

    request = documentai.BatchProcessRequest(
        name=name,
        input_documents=input_config,
        document_output_config=output_config,
        notification_config=notification_config  # 添加通知配置
    )

    # 仅启动批量处理任务，不等待完成
    client.batch_process_documents(request)
    
    # 直接返回，Cloud Function（xyz）执行结束
    return "Document AI batch processing started successfully"

步骤2：创建Cloud Pub/Sub主题并配置权限 #

1. 在Google Cloud控制台创建Pub/Sub主题（即代码中的document-ai-batch-completion-topic）。
2. 给Document AI的服务账号授予Pub/Sub Publisher权限：
   • Document AI服务账号格式：service-${PROJECT_NUMBER}@gcp-sa-documentai.iam.gserviceaccount.com
   • 在Pub/Sub主题的权限设置中，添加该账号，分配角色roles/pubsub.publisher。

步骤3：配置Cloud Function（abc）作为Pub/Sub订阅触发器 #

1. 创建Cloud Function（abc），选择Pub/Sub触发器，关联步骤2中创建的主题。
2. 在abc中解析Pub/Sub消息，获取批量处理的结果信息（如输出GCS路径、LRO详情等），然后进行后续处理：

import base64
import json
from google.cloud import storage

def process_batch_completion(event, context):
    # 解析Pub/Sub消息
    pubsub_message = base64.b64decode(event['data']).decode('utf-8')
    message_data = json.loads(pubsub_message)
    
    # 从消息中提取关键信息
    lro_name = message_data['name']
    gcs_output_uri = message_data['metadata']['documentOutputConfig']['gcsOutputConfig']['gcsUri']
    
    # 示例：遍历GCS输出目录，处理解析后的文档
    storage_client = storage.Client()
    bucket_name, prefix = gcs_output_uri.replace("gs://", "").split("/", 1)
    bucket = storage_client.bucket(bucket_name)
    blobs = bucket.list_blobs(prefix=prefix)
    
    for blob in blobs:
        if blob.name.endswith('.json'):
            # 读取解析结果
            parsed_document = json.loads(blob.download_as_text())
            # 在这里添加你的业务逻辑处理
            print(f"Processed document: {blob.name}")
    
    return "Batch completion processed successfully"

替代方案：用Cloud Workflows编排流程 #

如果需要更可控的流程编排，可使用Cloud Workflows替代Cloud Function（xyz），它支持长时间等待LRO完成，再触发Cloud Function（abc），适合需要同步监控任务状态的场景：

1. 创建Workflow定义，包含启动Document AI批量处理、等待任务完成、调用abc三个步骤。
2. 将Workflow触发方式设置为Cloud Storage上传事件（替代原xyz的触发器）。

关键注意事项 #

• 权限配置：确保abc拥有读取Document AI输出桶的Storage Object Viewer权限，以及访问Document AI LRO的Document AI Viewer权限。
• 消息可靠性：Pub/Sub订阅配置为重试策略，避免因abc临时故障导致通知丢失。
• 结果验证：在abc中可通过LRO名称调用Document AI API查询任务状态，确保处理成功后再进行后续操作。

内容的提问来源于stack exchange，提问作者Manish gupta