构建代理以丰富元数据

借助 Knowledge Catalog（以前称为 Dataplex Universal Catalog），您可以自动为整个组织中的数据资产大规模丰富元数据。添加来自外部来源的业务上下文有助于用户和下游 AI 智能体发现、理解和查询回答问题所需的数据。在本教程中，您将使用 knowledge-catalog 代码库中的示例代码构建一个代理，用于生成 BigQuery 表文档并将其发布到 Knowledge Catalog。

元数据扩充的运作方式

虽然 Knowledge Catalog 会自动管理资源、跟踪技术架构并生成说明和数据分析，但有价值的业务上下文通常位于其他位置，例如：

  • 内部文档和 Wiki
  • 代码库
  • 沟通渠道，例如 Google Chat 和 Slack

您可以构建 AI 智能体，从这些来源中提取上下文，将其与技术元数据相结合，并持续丰富 Knowledge Catalog 中的目录条目。

目标

在本教程中，您将构建一个执行以下任务的智能体：

  • 从知识库、文档、代码或聊天中提取业务背景信息，以丰富技术元数据。
  • 根据提取的上下文和其他信息源为 BigQuery 表生成文档。
  • 将生成的文档发布到 Knowledge Catalog，以改进搜索和发现功能，使条目更易于查找和理解。

准备工作

如需运行 Knowledge Catalog 丰富代理，您必须满足以下要求：

所需的角色

如需获得使用丰富代理所需的权限，请让您的管理员为您授予 Google Cloud 项目 iam.gserviceaccount.com的以下 IAM 角色：

如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

这些预定义角色包含使用富化代理所需的权限。如需查看所需的确切权限，请展开所需权限部分：

所需权限

如需使用丰富代理，您需要具有以下权限：

  • bigquery.projects.get/createDatasets
  • dataplex.projects.search
  • dataplex.entryGroups.get/updateEntries
  • aiplatform.endpoints.predict
  • serviceusage.services.use

您也可以使用自定义角色或其他预定义角色来获取这些权限。

启用 API

如需使用 Knowledge Catalog 扩充智能体，请在项目中启用以下 API：

  • BigQuery API
  • Dataplex API
  • Vertex AI API
  • Service Usage API

启用 API 所需的角色

如需启用 API，您需要拥有 serviceusage.services.enable 权限。如果您创建了项目，则可能已经通过 Owner 角色 (roles/owner) 拥有此权限。否则，您可以通过 Service Usage Admin 角色 (roles/serviceusage.serviceUsageAdmin) 获得此权限。了解如何授予角色。

启用 API

安装依赖项

您需要以下 Python 软件包和工具才能运行示例：

  • google-adk（智能体开发套件 [ADK]）
  • google-cloud-dataplex Knowledge Catalog Python 客户端
  • google-auth 管理应用默认凭证
  • mcp[cli]（用于构建示例 MCP 服务器）
  • gcloud，用于身份验证和配置。如需安装 Google Cloud CLI，请参阅 Google Cloud SDK 文档。

设置环境

  1. 配置 gcloud 并登录：

    gcloud auth application-default login
gcloud config set core/project PROJECT_ID

    替换以下内容：

    • 将 PROJECT_ID 替换为项目的 ID

  2. 克隆 knowledge-catalog 代码库，然后前往示例源目录：

    git clone https://github.com/GoogleCloudPlatform/knowledge-catalog.git
cd knowledge-catalog/samples/enrichment/src

  3. 如需安装依赖项，请使用提供的脚本，该脚本会设置 Python 虚拟环境和必要的环境变量：

    source env.sh --install

  4. 如需在 Cloud 项目的 us 区域中创建名为 kc_sample_analytics 的示例 BigQuery 数据集，请运行 create_data.py 脚本：

    python3 ../sample/data/create_data.py

    该示例还在 sample/docs 目录中包含许多文档。这些文档构成了一个本地知识库。扩充智能体使用此知识库来提取信息并生成文档。

下载元数据

首先，运行下载工具，从 Knowledge Catalog 中提取 BigQuery 数据集及其表的元数据快照。这会创建本地元数据制品。

--dir 参数指定元数据文件的写入目录。

python3 -m enrichment.download \
  --dir ../sample/metadata.initial \
  --dataset ${KC_ENRICH_SAMPLE_PROJECT}.kc_sample_analytics

该脚本会在 sample/metadata 目录中为每个表创建一个 Markdown 文件，并使用以下命名惯例：<project_id>.<dataset_id>.<table_id>.md。

丰富元数据

创建本地 Markdown 文件后，运行丰富代理。智能体会遍历每个文件，查找与表格相关的信息，并汇总发现结果以及引用内容，以生成更新后的 Markdown 文件。

  • --dir：指定包含本地元数据文件的目录。
  • --output-dir：指定更新后的元数据文件的目标目录。
  • --config-dir：指定包含智能体指令、MCP 工具和技能的目录。
python3 -m enrichment.enrich \
  --dir ../sample/metadata.initial \
  --output-dir ../sample/metadata.new \
  --config-dir ../sample/config

查看元数据

经过丰富处理的元数据文件包含智能体生成的文档。在将更改发布到 Knowledge Catalog 之前，请根据需要查看和修改文件。

git diff --no-index ../sample/metadata.initial ../sample/metadata.new

发布元数据

运行发布工具，将丰富化元数据部署到 Knowledge Catalog。

python3 -m src.enrichment.publish --dir ../sample/metadata.new

根据您的数据进行自定义

在上一步中，您使用了 --config-dir 参数，让代理指向 ../sample/config 目录以获取其配置。这样，智能体就知道在哪里可以找到信息，以及如何与不同的来源进行交互。

该示例附带默认配置，可指示智能体使用本地 MCP 服务器访问本地知识库 (sample/docs) 中的文件。如需在您的环境中应用此工作流，您可以自定义这些配置文件，将智能体连接到您的内部 Wiki、代码库、Google 云端硬盘或其他系统。

sample/config/ 目录包含以下文件：

sample/config/
├─ instructions.md
├─ mcp.json
└─ skills/
    └─ kb-search/
        └─ SKILL.md
  • instructions.md：使用与贵组织相关的详细信息来扩充智能体的基准指令，例如指示其搜索特定知识库。
  • mcp.json：配置 MCP 服务器，智能体可以使用这些服务器访问信息源的工具，例如从本地目录读取文件的工具。
  • SKILL.md：描述智能体应如何使用特定工具与信息源互动，例如使用 list_contents、read_file 和 search_content 在本地文档中查找信息。

探索 Knowledge Catalog 示例代码

丰富流程部分中的 download 和 publish 工具使用 Knowledge Catalog API 读取和写入元数据。

本部分介绍了这些 API 的工作原理，以便您根据自己的集成情况调整示例。

搜索和检索元数据

该示例使用以下 API 来搜索和检索元数据：

  • SearchEntries，用于检索数据集的条目和位置元数据。
  • ListEntries，用于枚举 Catalog EntryGroup 中的 BigQuery 表。
  • GetEntry，以提取每个 BigQuery 表的特定元数据。

以下代码展示了如何搜索数据集以找到其条目组、列出其中包含的所有表，以及检索其特定元数据：

import google.cloud.dataplex_v1 as dataplex

BIGQUERY_TABLE_TYPE = "projects/dataplex-types/locations/global/entryTypes/bigquery-table"
OVERVIEW_ASPECT_TYPE = "projects/dataplex-types/locations/global/aspectTypes/overview"

catalog = dataplex.CatalogServiceClient()

dataset_reference = '...'   # project_id.dataset_id
project_id, dataset_id = dataset_reference.split('.')

# 1. Search for dataset to determine its location
search_response = catalog.search_entries(
    request=dataplex.SearchEntriesRequest(
        name=f"projects/{project_id}/locations/global",
        query=f"type=dataset name={dataset_id}",
        page_size=1
    )
)
dataset_entry = search_response.results[0].dataplex_entry
location_id = dataset_entry.entry_source.location

# 2. List resources in the underlying group
entry_group_name = f"projects/{project_id}/locations/{location_id}/entryGroups/@bigquery"
entry_filter = f'parent_entry="{dataset_entry.name}"'
list_response = catalog.list_entries(
    request=dataplex.ListEntriesRequest(
        parent=entry_group_name,
        entry_filter=entry_filter,
    )
)

# 3. Retrieve metadata for each table in the list
for table_entry in list_response.entries:
    entry = catalog.get_entry(
        request=dataplex.GetEntryRequest(
            name=table_entry.name,
            view="CUSTOM",
            aspect_types=[OVERVIEW_ASPECT_TYPE]
        )
    )

更新表元数据

以下代码展示了如何将生成的文档发布到表的“概览”切面并更新其元数据：

import google.cloud.dataplex_v1 as dataplex
import google.protobuf.field_mask_pb2 as field_mask_pb2
import google.protobuf.json_format as jsonpb

OVERVIEW_ASPECT_TYPE = "projects/dataplex-types/locations/global/aspectTypes/overview"
OVERVIEW_ASPECT_KEY = "dataplex-types.global.overview"

catalog = dataplex.CatalogServiceClient()

table_reference = "..."    # project_id.dataset_id.table_id
project_id, dataset_id, table_id = table_reference.split('.')

entry_data = {
    "name": f"bigquery.googleapis.com/projects/{project_id}/datasets/{dataset_id}/tables/{table_id}",
    "aspects": {
        OVERVIEW_ASPECT_KEY: {
            "aspectType": OVERVIEW_ASPECT_TYPE,
            "data": {
                "content": "...", # content parsed from local markdown file
                "contentType": "MARKDOWN"
            }
        }
    }
}

entry = dataplex.Entry()
jsonpb.ParseDict(entry_data, entry._pb)

catalog.update_entry(
    request=dataplex.UpdateEntryRequest(
        entry=entry,
        update_mask=field_mask_pb2.FieldMask(paths=["aspects"]),
        aspect_keys=[OVERVIEW_ASPECT_KEY],
    )
)

后续步骤