監控長時間執行的作業

本頁說明如何管理 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

在試用這個範例之前，請先按照Python使用用戶端程式庫的 Gemini Enterprise 快速入門導覽課程中的操作說明進行設定。詳情請參閱 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

在試用這個範例之前，請先按照Python使用用戶端程式庫的 Gemini Enterprise 快速入門導覽課程中的操作說明進行設定。詳情請參閱 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

在試用這個範例之前，請先按照Python使用用戶端程式庫的 Gemini Enterprise 快速入門導覽課程中的操作說明進行設定。詳情請參閱 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

在試用這個範例之前，請先按照Python使用用戶端程式庫的 Gemini Enterprise 快速入門導覽課程中的操作說明進行設定。詳情請參閱 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