监控长时间运行的操作

本页介绍了如何在 AI 应用中管理长时间运行的操作 (LRO) 的生命周期。

如果对某个方法的调用可能需要很长时间才能完成，系统就会返回长时间运行的操作对象。例如，当您通过 API 或客户端库调用 documents.import 时，AI Applications API 会创建一个长时间运行的操作。该操作会跟踪处理作业的状态。

您可以使用 AI Applications API 提供的长时间运行的操作方法来检查操作的状态。您还可以列出或轮询操作。

在操作完成之后，操作的记录会保留大约 30 天，也就是说，在这段时间之后，您将无法再查看或列出操作。

列出长时间运行的操作

以下示例展示了如何列出 Google Cloud资源的操作。

REST

如需列出 Google Cloud 资源的长时间运行操作，请按以下步骤操作：

  • 调用 operations.list 方法

    curl -X GET \
    -H "Authorization: Bearer $(gcloud auth print-access-token)" \
    "https://discoveryengine.googleapis.com/v1/projects/PROJECT_ID/locations/global/collections/default_collection/dataStores/DATA_STORE_ID/operations"

    DATA_STORE_ID：使用引擎创建的 AI 应用数据存储区的 ID。在 Google Cloud 控制台网址中，数据存储区 ID 位于 engines/ 之后、/data 之前。

Python

如需了解详情，请参阅 AI 应用 Python API 参考文档。

如需向 AI 应用进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。 

from typing import Optional

from google.cloud import discoveryengine
from google.longrunning import operations_pb2

# TODO(developer): Uncomment these variables before running the sample.
# project_id = "YOUR_PROJECT_ID"
# location = "YOUR_PROCESSOR_LOCATION"  # Options: "global"
# search_engine_id = "YOUR_SEARCH_ENGINE_ID"

# Create filter in https://google.aip.dev/160 syntax
# operations_filter = "YOUR_FILTER"


def list_operations_sample(
    project_id: str,
    location: str,
    search_engine_id: str,
    operations_filter: Optional[str] = None,
) -> operations_pb2.ListOperationsResponse:
    # Create a client
    client = discoveryengine.DocumentServiceClient()

    # The full resource name of the search engine branch.
    name = f"projects/{project_id}/locations/{location}/collections/default_collection/dataStores/{search_engine_id}"

    # Make ListOperations request
    request = operations_pb2.ListOperationsRequest(
        name=name,
        filter=operations_filter,
    )

    # Make ListOperations request
    response = client.list_operations(request=request)

    # Print the Operation Information
    for operation in response.operations:
        print(operation)

    return response

获取长时间运行的操作的详细信息

以下示例展示了如何获取操作的详细信息。

REST

如需获取长时间运行的操作的状态并查看其详细信息，请按以下步骤操作：

  1. 您可以通过以下两种方式之一找到相应操作的名称：

    • 调用会返回长时间运行的操作的方法后，请查看响应。

      例如，如果您调用 documents.import，响应的开头将如下所示：

      {
  "operations": [
    {
      "name": "projects/12345/locations/global/collections/default_collection/dataStores/my-datastore_4321/branches/0/operations/import-documents-56789",
      "metadata": {
        "@type": "type.googleapis.com/google.cloud.discoveryengine.v1.ImportDocumentsMetadata",
      }
    }
  ]
}

      响应中的 name 值提供操作名称，可用于查询操作状态。复制操作名称时，请勿添加引号。

    • 通过列出长时间运行的操作来获取操作名称。

  2. 对创建操作的资源调用 operations.get 方法：

    curl -X GET \
    -H "Authorization: Bearer $(gcloud auth print-access-token)" \
    "https://discoveryengine.googleapis.com/v1/OPERATION_NAME"

    OPERATION_NAME：提供您需要了解的操作的名称。您可以通过列出长时间运行的操作来查找操作名称。

    GET 命令的响应前几行如下所示：

    {
      "name": "projects/12345/locations/global/collections/default_collection/dataStores/my-datastore_4321/branches/0/operations/import-documents-56789",
      "metadata": {
        "@type": "type.googleapis.com/google.cloud.discoveryengine.v1.ImportDocumentsMetadata"
      }
    }

Python

如需了解详情，请参阅 AI 应用 Python API 参考文档。

如需向 AI 应用进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。 

from google.cloud import discoveryengine
from google.longrunning import operations_pb2

# TODO(developer): Uncomment these variables before running the sample.
# Example: `projects/{project}/locations/{location}/collections/{default_collection}/dataStores/{search_engine_id}/branches/{0}/operations/{operation_id}`
# operation_name = "YOUR_OPERATION_NAME"


def get_operation_sample(operation_name: str) -> operations_pb2.Operation:
    # Create a client
    client = discoveryengine.DocumentServiceClient()

    # Make GetOperation request
    request = operations_pb2.GetOperationRequest(name=operation_name)
    operation = client.get_operation(request=request)

    # Print the Operation Information
    print(operation)

    return operation

轮询长时间运行的操作

以下示例展示了如何轮询操作的状态。

REST

如需轮询长时间运行的操作，直到其完成为止，请按以下步骤操作：

  1. 运行以下命令，该命令会重复调用 operations.get 方法，并在每次请求之间使用 10 秒的退避时间：

    while true; \
    do curl -X GET \
    -H "Authorization: Bearer $(gcloud auth print-access-token)" \
    "https://discoveryengine.googleapis.com/v1/OPERATION_NAME"; \
    sleep 10; \
    done

    OPERATION_NAME：提供您要轮询的操作的名称。您可以通过列出长时间运行的操作来查找操作名称。例如 projects/12345/locations/global/collections/default_collection/dataStores/my-datastore_4321/branches/0/operations/import-documents-56789。

  2. 在状态显示 "done": true 后停止轮询作业 (Control+Z)。

Python

如需了解详情，请参阅 AI 应用 Python API 参考文档。

如需向 AI 应用进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。 

from time import sleep

from google.cloud import discoveryengine
from google.longrunning import operations_pb2

# TODO(developer): Uncomment these variables before running the sample.
# Example: `projects/{project}/locations/{location}/collections/{default_collection}/dataStores/{search_engine_id}/branches/{0}/operations/{operation_id}`
# operation_name = "YOUR_OPERATION_NAME"


def poll_operation_sample(
    operation_name: str, limit: int = 10
) -> operations_pb2.Operation:
    # Create a client
    client = discoveryengine.DocumentServiceClient()

    # Make GetOperation request
    request = operations_pb2.GetOperationRequest(name=operation_name)

    for _ in range(limit):
        operation = client.get_operation(request=request)
        # Print the Operation Information
        print(operation)

        # Stop polling when Operation is no longer running
        if operation.done:
            break

        # Wait 10 seconds before polling again
        sleep(10)

    return operation

取消长时间运行的操作

下面介绍了如何取消操作：

REST

如需取消长时间运行的操作，请按以下步骤操作：

  • 调用 operations.cancel 方法

    curl -X post \
   -H "Authorization: Bearer $(gcloud auth print-access-token)" \
   "https://discoveryengine.googleapis.com/v1/OPERATION_NAME":cancel

    OPERATION_NAME：提供您要取消的操作的名称。您可以通过列出长时间运行的操作来查找操作名称。例如 projects/12345/locations/global/collections/default_collection/dataStores/my-datastore_4321/branches/0/operations/import-documents-56789。

    发出 API 调用后，服务器会尝试取消操作。您会看到以下结果，并可以执行以下操作：

    • 如果 "code": 400 和 "status": "FAILED_PRECONDITION" 出现错误，则表示无法取消请求。

    • 取消成功后，系统会返回一个空的 JSON 对象。如需验证取消操作，请执行以下操作：

      • 使用 operations.get 方法。

      • 如果操作成功取消，来自 operations.get 方法的响应将包含错误 "code": 1，表示 CANCELLED 状态代码。

        例如：

        {
  "name": "projects/12345/locations/global/collections/default_collection/dataStores/my-datastore_4321/branches/0/operations/import-documents-56789",
  "metadata": {
    "@type": "type.googleapis.com/google.cloud.discoveryengine.v1alpha.ImportDocumentsMetadata",
    "createTime": "2025-04-28T21:29:21.199190Z",
    "updateTime": "2025-04-28T21:31:29.076865Z"
  },
  "done": true,
  "error": {
    "code": 1,
    "message": "Operation projects/12345/locations/global/collections/default_collection/dataStores/my-datastore_4321/branches/0/operations/import-documents-56789 is cancelled."
  }
}

Python

如需了解详情，请参阅 AI 应用 Python API 参考文档。

如需向 AI 应用进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。 

from google.cloud import discoveryengine
from google.longrunning import operations_pb2

# TODO(developer): Uncomment these variables before running the sample.
# Example: `projects/{project}/locations/{location}/collections/{default_collection}/dataStores/{search_engine_id}/branches/{0}/operations/{operation_id}`
# operation_name = "YOUR_OPERATION_NAME"


def cancel_operation_sample(operation_name: str) -> None:
    # Create a client
    client = discoveryengine.DocumentServiceClient()

    # Make CancelOperation request
    request = operations_pb2.CancelOperationRequest(name=operation_name)
    client.cancel_operation(request=request)

    return