長時間実行オペレーションをモニタリングする

このページでは、Vertex AI Agent Builder で長時間実行オペレーション（LRO）のライフサイクルを管理する方法について説明します。

メソッドの呼び出しが完了するまでに時間がかかる可能性がある場合は、長時間実行オペレーション オブジェクトが返されます。たとえば、Vertex AI Agent Builder API は、API またはクライアント ライブラリを介して documents.import を呼び出すと、長時間実行オペレーションを作成します。このオペレーションは、処理ジョブのステータスを追跡します。

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 コンソールの URL では、データストア 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. 次のコマンドを実行します。このコマンドは、各リクエストの間に 10 秒のバックオフを使用して、operations.get メソッドを繰り返し呼び出します。

    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