医療データを更新する

Vertex AI Search のヘルスケア データストアにデータを最初にインポートした後、ソース FHIR ストアで次のいずれかの更新を行った可能性があります。

  • 新しい FHIR リソースを追加しました
  • 既存の FHIR リソースを更新しました
  • FHIR リソースを削除しました

このような場合は、ソース FHIR ストアから Vertex AI Search ヘルスケア データストアへの変更を調整できます。

調整の概要

変更の調整は、増分または完全な方法で行うことができます。次の表に、2 つのモードの比較を示します。

ソース FHIR ストアの変更 増分モード 　　　 フルモード
新しい FHIR リソース Vertex AI Search データストアに新しいドキュメントを追加します 　　　 Vertex AI Search データストアに新しいドキュメントを追加します 　　　
FHIR リソースを更新しました ドキュメント ID を保持しながら、Vertex AI Search データストア内の既存のドキュメントを置き換えます 　　　 ドキュメント ID を保持しながら、Vertex AI Search データストア内の既存のドキュメントを置き換えます 　　　
FHIR リソースを削除しました 調整されません 　　　 Vertex AI Search データストアから対応するドキュメントを削除します 　　　

始める前に

プロジェクトの割り当てと上限を確認します。 Google Cloud Vertex AI Search ヘルスケア データストアには、プロジェクトごとに最大 100 万個のドキュメントを含めることができます。インポート中にこの割り当てに達すると、インポート プロセスは停止します。

増分インポートを実行する

次のサンプルは、documents.import メソッドを使用して Cloud Healthcare API FHIR ストアから増分変更をインポートする方法を示しています。

service-PROJECT_NUMBER@gcp-sa-discoveryengine.iam.gserviceaccount.com サービス アカウントに次の権限、または、次の Identity and Access Management（IAM）ロールを付与します。
RESTPython

  1. 増分インポートを実行します。

    curl -X POST \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json; charset=utf-8" \
-H "X-Goog-User-Project: PROJECT_ID" \
"https://us-discoveryengine.googleapis.com/v1alpha/projects/PROJECT_ID/locations/us/dataStores/DATA_STORE_ID/branches/0/documents:import" \
-d '{
   "reconciliation_mode": "INCREMENTAL",
   "fhir_store_source": {"fhir_store": "projects/PROJECT_ID/locations/CLOUD_HEALTHCARE_DATASET_LOCATION/datasets/CLOUD_HEALTHCARE_DATASET_ID/fhirStores/FHIR_STORE_ID"}
}'

    次のように置き換えます。

    • PROJECT_ID: プロジェクトの ID。 Google Cloud
    • DATA_STORE_ID: Vertex AI Search データストアの ID。
    • CLOUD_HEALTHCARE_DATASET_ID: ソース FHIR ストアを含む Cloud Healthcare API データセットの ID。
    • CLOUD_HEALTHCARE_DATASET_LOCATION: ソース FHIR ストアを含む Cloud Healthcare API データセットのロケーション。
    • FHIR_STORE_ID: Cloud Healthcare API FHIR R4 ストアの ID。

    次のような JSON レスポンスが返されます。レスポンスには、長時間実行オペレーションの ID が含まれます。長時間実行オペレーションは、メソッドの呼び出しが完了するまでにかなりの時間がかかる場合に返されます。IMPORT_OPERATION_ID の値をメモします。この値は、インポートのステータスを確認するために必要です。 

      {
    "name": "projects/PROJECT_ID/locations/us/collections/default_collection/dataStores/DATA_STORE_ID/branches/0/operations/IMPORT_OPERATION_ID",
    "metadata": {
      "@type": "type.googleapis.com/google.cloud.discoveryengine.v1alpha.ImportDocumentsMetadata"
    }
  }

  2. FHIR データのインポート オペレーションが完了しているかどうかを確認します。

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

    次のように置き換えます。

    • PROJECT_ID: プロジェクトの ID。 Google Cloud
    • DATA_STORE_ID: Vertex AI Search データストアの ID。
    • IMPORT_OPERATION_ID: import メソッドの呼び出し時に返される長時間実行オペレーションのオペレーション ID。

    次のような JSON レスポンスが返されます。インポート オペレーションは長時間実行オペレーションです。オペレーションの実行中、レスポンスには次のフィールドが含まれます。

    • successCount: これまでに正常にインポートされた FHIR リソースの数を示します。
    • failureCount: これまでにインポートに失敗した FHIR リソースの数を示します。このフィールドは、インポートに失敗した FHIR リソースがある場合にのみ表示されます。

    オペレーションが完了すると、レスポンスに次のフィールドが含まれます。

    • successCount: 正常にインポートされた FHIR リソースの数を示します。
    • failureCount: インポートに失敗した FHIR リソースの数を示します。このフィールドは、インポートに失敗した FHIR リソースがある場合にのみ表示されます。
    • totalCount: ソース FHIR ストアに存在する FHIR リソースの数を示します。このフィールドは、インポートに失敗した FHIR リソースがある場合にのみ表示されます。
    • done: インポート オペレーションが完了したことを示す値 true があります
    • errorSamples: インポートに失敗したリソースに関する情報を提供します。このフィールドは、インポートに失敗した FHIR リソースがある場合にのみ表示されます。
    • errorConfig: エラーの概要ログファイルを含む Cloud Storage ロケーションのパスを提供します。
    {
 "name": "projects/PROJECT_ID/locations/us/collections/default_collection/dataStores/DATA_STORE_ID/branches/0/operations/IMPORT_OPERATION_ID",
 "metadata": {
   "@type": "type.googleapis.com/google.cloud.discoveryengine.v1alpha.ImportDocumentsMetadata",
   "createTime": "START_TIMESTAMP",
   "updateTime": "END_TIMESTAMP",
   "successCount": "SUCCESS_COUNT",
   "failureCount": "FAILURE_COUNT",
   "totalCount": "TOTAL_COUNT",
 },
 "done": true,
 "response": {
   "@type": "type.googleapis.com/google.cloud.discoveryengine.v1alpha.ImportDocumentsResponse",
  "errorSamples": [ERROR_SAMPLE],
  "errorConfig": {
     "gcsPrefix": "LOG_FILE_LOCATION"
   }
 }
}

詳細については、Vertex AI Agent Builder Python API のリファレンス ドキュメントをご覧ください。

Vertex AI Agent Builder に対する認証を行うには、アプリケーションのデフォルト認証情報を設定します。 詳細については、ローカル開発環境の認証の設定をご覧ください。

from google.api_core.client_options import ClientOptions
from google.cloud import discoveryengine

# TODO(developer): Uncomment these variables before running the sample.
# project_id = "YOUR_PROJECT_ID"
# location = "YOUR_LOCATION" # Values: "us"
# data_store_id = "YOUR_DATA_STORE_ID"
# healthcare_project_id = "YOUR_HEALTHCARE_PROJECT_ID"
# healthcare_location = "YOUR_HEALTHCARE_LOCATION"
# healthcare_dataset_id = "YOUR_HEALTHCARE_DATASET_ID"
# healthcare_fihr_store_id = "YOUR_HEALTHCARE_FHIR_STORE_ID"

#  For more information, refer to:
# https://cloud.google.com/generative-ai-app-builder/docs/locations#specify_a_multi-region_for_your_data_store
client_options = (
    ClientOptions(api_endpoint=f"{location}-discoveryengine.googleapis.com")
    if location != "global"
    else None
)

# Create a client
client = discoveryengine.DocumentServiceClient(client_options=client_options)

# The full resource name of the search engine branch.
# e.g. projects/{project}/locations/{location}/dataStores/{data_store_id}/branches/{branch}
parent = client.branch_path(
    project=project_id,
    location=location,
    data_store=data_store_id,
    branch="default_branch",
)

request = discoveryengine.ImportDocumentsRequest(
    parent=parent,
    fhir_store_source=discoveryengine.FhirStoreSource(
        fhir_store=client.fhir_store_path(
            healthcare_project_id,
            healthcare_location,
            healthcare_dataset_id,
            healthcare_fihr_store_id,
        ),
    ),
    # Options: `FULL`, `INCREMENTAL`
    reconciliation_mode=discoveryengine.ImportDocumentsRequest.ReconciliationMode.INCREMENTAL,
)

# Make the request
operation = client.import_documents(request=request)

print(f"Waiting for operation to complete: {operation.operation.name}")
response = operation.result()

# After the operation is complete,
# get information from operation metadata
metadata = discoveryengine.ImportDocumentsMetadata(operation.metadata)

# Handle the response
print(response)
print(metadata)