监控长时间运行的操作

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

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

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

Python

如需了解详情，请参阅 Agent Search Python API 参考文档。

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

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

如需了解详情，请参阅 Agent Search Python API 参考文档。

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

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

如需了解详情，请参阅 Agent Search Python API 参考文档。

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

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

如需了解详情，请参阅 Agent Search Python API 参考文档。

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

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