監控長時間執行的作業

本頁面說明如何在 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