監控長時間執行的作業

本頁說明如何在 AI 應用程式中管理長時間執行作業 (LRO) 的生命週期。

如果對方法的呼叫可能需要很長的時間才能完成，系統會傳回 長時間執行的作業物件。舉例來說，當您透過 API 或用戶端程式庫呼叫 documents.import 時，AI Applications API 會建立長時間執行的作業。作業會追蹤處理工作的狀態。

您可以使用 AI Applications 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：使用引擎建立的 AI 應用程式資料儲存庫 ID。在 Google Cloud 控制台網址中，資料儲存庫 ID 會出現在 engines/ 之後、/data 之前。

Python

詳情請參閱 AI Applications Python API 參考說明文件。

如要向 AI Applications 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定驗證機制」。

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

詳情請參閱 AI Applications Python API 參考說明文件。

如要向 AI Applications 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定驗證機制」。

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

詳情請參閱 AI Applications Python API 參考說明文件。

如要向 AI Applications 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定驗證機制」。

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

詳情請參閱 AI Applications Python API 參考說明文件。

如要向 AI Applications 進行驗證，請設定應用程式預設憑證。詳情請參閱「為本機開發環境設定驗證機制」。

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