刷新健康数据

在将数据初始导入 Vertex AI Search 医疗保健数据存储区后，您可能已在源 FHIR 存储区中执行以下任一更新：

  • 添加了新的 FHIR 资源
  • 更新了现有 FHIR 资源
  • 已删除的 FHIR 资源

在这种情况下，您可以将源 FHIR 存储区中的更改协调到 Vertex AI Search 医疗保健数据存储区。

对账概览

您可以增量或完全对更改进行协调。下表比较了这两种模式。

对源 FHIR 存储区所做的更改 增量模式 完整模式
新的 FHIR 资源 向 Vertex AI Search 数据存储区添加新文档 向 Vertex AI Search 数据存储区添加新文档
更新了 FHIR 资源 替换 Vertex AI Search 数据存储区中的现有文档，同时保留文档 ID 替换 Vertex AI Search 数据存储区中的现有文档，同时保留文档 ID
已删除的 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：您的 Google Cloud 项目的 ID。
    • 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 响应。响应包含长时间运行的操作的标识符。当方法调用可能需要大量时间才能完成时，会返回长时间运行的操作。记下 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：您的 Google Cloud 项目的 ID。
    • 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)