创建医疗保健搜索数据存储区

如需在 Vertex AI Search 中搜索临床数据，您可以按照以下任一工作流程操作：

  • 创建一个医疗保健数据存储区，将 FHIR R4 数据导入该数据存储区，将其连接到医疗保健搜索应用，然后查询临床数据。
  • 在应用创建过程中，创建一个医疗保健搜索应用、创建一个医疗保健数据存储区，并将 FHIR R4 数据导入该数据存储区，然后查询临床数据。如需了解详情，请参阅创建医疗保健搜索应用。

本页介绍了第一种方法。

数据导入频率简介

您可以通过以下方式将 FHIR R4 数据导入数据存储区：

  • 批量导入：一次性导入。数据会导入到数据存储区。如需进一步增量导入，请参阅刷新医疗保健数据。

  • 流式导入（预览版）：近乎实时流式数据导入。源 FHIR 存储区中的任何增量更改都会同步到 Vertex AI Search 数据存储区。流式传输需要使用数据连接器，这是一种包含实体的数据存储区。实体也是数据存储区实例。

    给定 Google Cloud 项目的数据流式传输速率取决于以下配额。如果超出配额，您可能会遇到流式传输延迟。

    • 每分钟待处理的 FHIR 或 BigQuery 流式写入操作的数量。 如需了解详情，请参阅配额和限制。
    • 每个区域每分钟的 FHIR 读取操作数。如需了解详情，请参阅 Cloud Healthcare API FHIR 配额。

您可以在创建数据存储区时选择数据导入频率，之后便无法更改此配置。

流式导入适用于 Vertex AI Search 支持的所有资源。如需了解详情，请参阅 Healthcare FHIR R4 数据架构参考文档。

准备工作

在创建医疗保健数据存储区并向其中导入数据之前，请完成以下要求：

  • 了解医疗保健搜索应用与数据存储区之间的关系。 如需了解详情，请参阅应用和数据存储区简介。

  • 准备 FHIR 数据以进行提取。

  • Vertex AI Search for Healthcare 仅在美国多区域 (us) 提供搜索服务。因此，您的医疗保健搜索应用和数据存储区必须位于 us 多区域。

创建数据存储区

您可以在 Google Cloud 控制台中或使用 API 创建数据存储区。这两种方法的不同之处体现在以下方面：

  • 在 Google Cloud 控制台中：在 Cloud Healthcare API 中选择来源 FHIR 存储区，并在创建医疗保健搜索数据存储区过程中导入 FHIR 数据。如需流式传输 FHIR 数据（预览版），您的源 FHIR 存储区必须与目标 Vertex AI Search 数据存储区位于同一 Google Cloud 项目中。
  • 通过 REST API：您可以从同一 Google Cloud 项目或其他项目中的 Cloud Healthcare API FHIR 存储区导入 FHIR 数据。
    1. 使用 dataStores.create 方法创建医疗保健数据存储区。
    2. 使用 documents.import 方法在 Cloud Healthcare API 中指定 FHIR 存储区并导入 FHIR R4 数据。

如需创建医疗保健数据存储区，请完成以下步骤。

控制台

  1. 在 Google Cloud 控制台中，前往 Agent Builder 页面。

    Agent Builder

  2. 在导航菜单中，点击数据存储区。

  3. 点击 Create data store（创建数据存储区）。

  4. 在选择数据源窗格中，选择 Healthcare API (FHIR) 作为数据源。
  5. 如需从 FHIR 存储区导入数据，请执行以下操作之一：
    • 从可用 FHIR 存储区列表中选择 FHIR 存储区：
      1. 展开 FHIR 存储区字段。
      2. 在此列表中，选择位于允许的位置中的数据集，然后选择使用 FHIR 版本 R4 的 FHIR 存储区。

        如需流式传输 FHIR 数据（预览版），您的源 FHIR 存储区必须与目标 Vertex AI Search 数据存储区位于同一 Google Cloud 项目中。

    • 手动进入 FHIR 存储区：
      1. 展开 FHIR 存储区字段。
      2. 点击手动输入 FHIR 存储区。
      3. 在 FHIR 存储区名称对话框中，按以下格式输入 FHIR 存储区的全名：

        project/PROJECT_ID/locations/LOCATION/datasets/DATASET_ID/fhirStores/FHIR_STORE_ID

      4. 点击保存。
  6. 在同步部分，选择以下选项之一。 数据存储区创建后，此选择将无法更改。
    • 一次性：执行一次性批量数据导入。如需进一步增量导入，请参阅刷新健康数据。
    • 流式传输：用于执行近乎实时的流式数据导入。如需流式传输数据，您必须创建数据连接器，这是一种数据存储区类型。这是预览版功能。如需使用 REST API 设置流式数据存储区，请与您的客户工程师联系。
  7. 在此数据采用什么架构？部分，选择以下选项之一：
    • Google 预定义架构：用于保留 Google 为受支持的 FHIR 资源和元素定义的架构配置，例如可编入索引性、可搜索性和可检索性。选择此选项后，您将无法在创建数据存储区后更新架构。如果您希望能够在创建数据存储区后更改架构，请选择自定义架构（预览版）选项。
      1. 点击继续。
      2. 在您的数据存储区名称字段中，输入数据存储区的名称。
      3. 点击创建。

        4. 您创建的数据存储区会显示在数据存储区页面上。

    • 自定义架构（预览版）：用于为受支持的 FHIR 资源和元素定义您自己的架构配置，例如可编入索引性、可搜索性和可检索性。这是预览版功能。如需设置可配置的架构，请与您的客户工程师联系。
      1. 点击继续。
      2. 查看架构、展开每个字段，然后修改字段设置。
      3. 点击添加新字段，在受支持的 FHIR 资源中添加新字段。您无法移除 Google 定义的架构中提供的字段。
      4. 点击继续。
      5. 在您的数据连接器名称字段中，为数据连接器输入名称。
      6. 点击创建。

        7. 您创建的数据连接器会显示在数据存储区页面上。 源 FHIR 存储区会作为数据连接器中的实体添加。

  8. 点击继续。

REST

  1. 创建数据存储区。

    curl -X POST\
 -H "Authorization: Bearer $(gcloud auth print-access-token)" \
 -H "Content-Type: application/json"\
 -H "X-Goog-User-Project: PROJECT_ID" \
"https://us-discoveryengine.googleapis.com/v1alpha/projects/PROJECT_ID/locations/us/collections/default_collection/dataStores?dataStoreId=DATA_STORE_ID" \
 -d '{
    "displayName": "DATA_STORE_DISPLAY_NAME",
    "industryVertical": "HEALTHCARE_FHIR",
    "solutionTypes": ["SOLUTION_TYPE_SEARCH"],
    "searchTier": "STANDARD",
    "searchAddOns": ["LLM"]
}'

    替换以下内容：

    • PROJECT_ID：您的 Google Cloud 项目的 ID。
    • DATA_STORE_ID：您要创建的 Vertex AI Search 数据存储区的 ID。此 ID 只能包含小写字母、数字、下划线和连字符。
    • DATA_STORE_DISPLAY_NAME：您要创建的 Vertex AI Search 数据存储区的显示名称。

  2. 如果源 FHIR 存储区和目标 Vertex AI Search 数据存储区位于同一 Google Cloud 项目中，请调用以下方法以执行一次性批量导入。如果它们不在同一项目中，请执行下一步。

    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": "FULL",
   "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。

  3. 如果源 FHIR 存储区和目标 Vertex AI Search 数据存储区位于不同的 Google Cloud 项目中，请调用以下方法执行一次性批量导入。如果它们位于同一项目中，请返回上一步。

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

    替换以下内容：

    • TARGET_PROJECT_ID：包含 Vertex AI Search 数据存储区的 Google Cloud 项目的 ID。
    • DATA_STORE_ID：Vertex AI Search 数据存储区的 ID。
    • SOURCE_PROJECT_ID：包含 Cloud Healthcare API 数据集和 FHIR 存储区的 Google Cloud 项目的 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。

  4. 可选：如需使用 REST API 设置流式数据导入（预览版），请与您的客户工程师联系。

Python

如需了解详情，请参阅 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: "global"
# data_store_id = "YOUR_DATA_STORE_ID"


def create_data_store_sample(
    project_id: str,
    location: str,
    data_store_id: str,
) -> str:
    #  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.DataStoreServiceClient(client_options=client_options)

    # The full resource name of the collection
    # e.g. projects/{project}/locations/{location}/collections/default_collection
    parent = client.collection_path(
        project=project_id,
        location=location,
        collection="default_collection",
    )

    data_store = discoveryengine.DataStore(
        display_name="My Data Store",
        # Options: GENERIC, MEDIA, HEALTHCARE_FHIR
        industry_vertical=discoveryengine.IndustryVertical.GENERIC,
        # Options: SOLUTION_TYPE_RECOMMENDATION, SOLUTION_TYPE_SEARCH, SOLUTION_TYPE_CHAT, SOLUTION_TYPE_GENERATIVE_CHAT
        solution_types=[discoveryengine.SolutionType.SOLUTION_TYPE_SEARCH],
        # TODO(developer): Update content_config based on data store type.
        # Options: NO_CONTENT, CONTENT_REQUIRED, PUBLIC_WEBSITE
        content_config=discoveryengine.DataStore.ContentConfig.CONTENT_REQUIRED,
    )

    request = discoveryengine.CreateDataStoreRequest(
        parent=parent,
        data_store_id=data_store_id,
        data_store=data_store,
        # Optional: For Advanced Site Search Only
        # create_advanced_site_search=True,
    )

    # Make the request
    operation = client.create_data_store(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.CreateDataStoreMetadata(operation.metadata)

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

    return operation.operation.name

导入文档

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)

验证数据存储区创建和 FHIR 数据导入

本任务将向您展示如何验证数据存储区是否已成功创建，以及 FHIR 数据是否已成功导入数据存储区。

  • 在 Google Cloud 控制台中：选择数据存储区并验证其详细信息。
  • 通过 REST API：
    1. 使用 dataStores.get 方法获取医疗保健数据存储区详细信息。
    2. 使用 operations.get 方法获取导入操作的详细信息。

如需验证数据存储区创建和数据导入，请完成以下步骤。

控制台

  1. 在 Google Cloud 控制台中，前往 Agent Builder 页面。

    Agent Builder

  2. 在导航菜单中，点击数据存储区。

    数据存储区页面会显示 Google Cloud 项目中的数据存储区及其详细信息的列表。

  3. 验证您创建的数据存储区或数据连接器是否在数据存储区列表中。

  4. 选择数据存储区或数据连接器，然后验证其详细信息。

    • 对于数据存储区：
      • 摘要表会列出以下详细信息：
        • 数据存储区 ID、类型和区域。
        • 文档数量，表示导入的 FHIR 资源数量。
        • 上次导入文档的时间戳。
        • （可选）点击查看详情可查看文档导入详情，例如导入成功、部分成功或失败的详细信息。
      • 文档标签页会在分页表格中列出导入的 FHIR 资源的资源 ID 及其资源类型。您可以过滤此表格，以验证是否导入了特定资源。
      • 活动标签页会列出文档导入详情，例如有关导入成功、部分成功或失败的详细信息。
    • 对于数据连接器：
      • 摘要表会列出以下详细信息：
        • 合集 ID、类型和区域。
        • 关联的应用的名称。
        • 连接器的状态，可以是“有效”或“已暂停”。
      • 实体表会显示数据连接器中的实体。实体的名称是来源 FHIR 存储区的名称。实体的 ID 是数据连接器的 ID 后附源 FHIR 存储区名称。
        • 点击实体名称可查看其详细信息。由于实体是数据连接器中的数据存储区实例，因此实体详情与数据存储区详情相同。

  5. 在架构标签页中，查看受支持的 FHIR 资源和元素的属性。点击修改以配置架构。这是“不公开预览”功能。如需设置可配置的架构，请与您的客户工程师联系。

REST

  1. 验证数据存储区创建情况。

    curl -X GET -H "Authorization: Bearer $(gcloud auth print-access-token)" \
 -H "Content-Type: application/json"\
 -H "X-Goog-User-Project: PROJECT_ID" \
 "https://us-discoveryengine.googleapis.com/v1alpha/projects/PROJECT_ID/locations/us/collections/default_collection/dataStores/DATA_STORE_ID"

    替换以下内容：

    • PROJECT_ID：您的 Google Cloud 项目的 ID。
    • DATA_STORE_ID：Vertex AI Search 数据存储区的 ID。

  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

后续步骤