监控长时间运行的操作

本页面介绍了如何管理 Gemini Enterprise 中长时间运行的操作 (LRO) 的生命周期。

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

您可以使用 Gemini Enterprise 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：使用您的引擎创建的 Gemini Enterprise 数据存储区的 ID。在 Google Cloud 控制台网址中，数据存储区 ID 显示在 engines/ 之后和 /data 之前。

Python

试用此示例之前，请按照 Gemini Enterprise 快速入门：使用客户端库中的 Python 设置说明进行操作。 如需了解详情，请参阅 Gemini Enterprise Python API 参考文档。

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

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

试用此示例之前，请按照 Gemini Enterprise 快速入门：使用客户端库中的 Python 设置说明进行操作。 如需了解详情，请参阅 Gemini Enterprise Python API 参考文档。

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

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

试用此示例之前，请按照 Gemini Enterprise 快速入门：使用客户端库中的 Python 设置说明进行操作。 如需了解详情，请参阅 Gemini Enterprise Python API 参考文档。

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

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

试用此示例之前，请按照 Gemini Enterprise 快速入门：使用客户端库中的 Python 设置说明进行操作。 如需了解详情，请参阅 Gemini Enterprise Python API 参考文档。

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

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