监控长时间运行的操作

本页面介绍了如何管理长时间运行的操作的生命周期 Vertex AI Agent Builder。

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

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

操作记录会在操作后保留大约 30 天 完成后，您将无法查看或列出操作。

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

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

REST

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

  1. 通过以下两种方式之一查找操作的名称：

    • 在调用返回长时间运行的消息的方法之后 请查看响应。

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

      {
  "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.v1beta.ImportDocumentsMetadata"
  }
}

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

    • 通过列出长时间运行的命令 操作。

  2. 调用 operations.get 方法：

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

    OPERATION_NAME：上一步中的操作名称。

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

    {
  "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.v1alpha.ImportDocumentsMetadata",
      }
    }
  ]
}

Python

有关详情，请参阅 Vertex AI Agent Builder Python API 参考文档。

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

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

列出长时间运行的操作

下面演示了如何列出 Google Cloud 的 资源。

REST

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

  • 调用 operations.list 方法

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

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

Python

有关详情，请参阅 Vertex AI Agent Builder Python API 参考文档。

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

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. 运行以下命令，该命令会重复调用 operations.get 方法，并在每次请求之间使用 10 秒的退避时间：

    while true; \
    do curl -X GET \
    -H "Authorization: Bearer $(gcloud auth print-access-token)" \
    "https://discoveryengine.googleapis.com/v1beta/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

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

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

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