建構代理程式來擴充中繼資料

Knowledge Catalog (舊稱 Dataplex Universal Catalog) 可管理整個組織資料資產的中繼資料。代理會使用這項中繼資料提供的背景資訊，找出、理解及查詢回答使用者問題所需的資料。

雖然 Knowledge Catalog 會自動管理資源、追蹤技術結構定義，以及產生說明和資料剖析，但重要的業務背景資訊通常位於其他位置，例如：

  • 內部文件和維基
  • 程式碼存放區
  • Google Chat 和 Slack 等通訊管道

您可以建構 AI 代理，從這些來源擷取背景資訊，並持續大規模充實中繼資料。本教學課程使用 knowledge-catalog 存放區中的程式碼範例，說明如何建構可執行下列操作的代理：

  • 擷取背景資訊：從知識庫、文件、程式碼或對話中擷取業務背景資訊，充實技術中繼資料。
  • 生成說明文件：根據擷取的背景資訊和其他資訊來源，生成 BigQuery 資料表的說明文件。
  • 改善搜尋與發現功能：將生成的說明文件發布至 Knowledge Catalog，讓項目更容易透過搜尋找到及瞭解。

事前準備

如要執行 Knowledge Catalog 擴充代理程式，必須符合下列規定：

必要的角色

如要取得使用擴充代理所需的權限，請要求管理員在 Google Cloud 專案 iam.gserviceaccount.com中授予您下列 IAM 角色：

如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和組織的存取權」。

這些預先定義的角色具備使用擴充代理所需的權限。如要查看確切的必要權限，請展開「Required permissions」(必要權限) 部分：

所需權限

必須具備下列權限，才能使用擴充代理：

  • 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 所需的角色

您必須具備 serviceusage.services.enable 權限，才能啟用 API。如果您建立了專案，可能已透過「擁有者」角色 (roles/owner) 取得這項權限。否則，您可以透過「Service Usage 管理員」角色 (roles/serviceusage.serviceUsageAdmin) 取得這項權限。瞭解如何授予角色。

啟用 API

安裝依附元件

您需要下列 Python 套件和工具，才能執行範例：

  • google-adk (Agent Development Kit (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. 如要在雲端專案的 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) 中的檔案。如要在環境中套用這個工作流程，您可以自訂這些設定檔，將代理連結至內部維基、程式碼存放區、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：列舉目錄 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],
    )
)

後續步驟