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

如需在 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 仅在 美国多区域 (us)。因此，您的医疗保健搜索应用和数据存储区 必须位于 us 多区域中。

创建数据存储区

您可以在 Google Cloud 控制台中或使用 API 创建数据存储区。通过 两种方法在以下方面有所不同：

  • 在 Google Cloud 控制台中：在 Cloud Healthcare API 中选择来源 FHIR 存储区，并在创建医疗保健搜索数据存储区过程中导入 FHIR 数据。如需流式传输 FHIR 数据（预览版），请使用您的来源 FHIR 存储区 必须与目标位于同一 Google Cloud 项目中 Vertex AI Search 数据存储区。
  • 通过 REST API：您可以从 Cloud Healthcare API FHIR 存储区导入 FHIR 数据 属于同一 Google Cloud 项目或其他 Google Cloud 项目。
    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 store 字段。
      2. 在此列表中，选择位于允许的位置中的数据集，然后选择使用 FHIR 版本 R4 的 FHIR 存储区。

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

    • 手动进入 FHIR 存储区：
      1. 展开 FHIR 存储区字段。
      2. 点击手动输入 FHIR 存储区。
      3. 在 FHIR store name 对话框中，按以下格式输入 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 的显示名 搜索要创建的数据存储区。

  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

后续步骤