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

Knowledge Catalog (舊稱 Dataplex Universal Catalog) 可管理整個機構的資料資產中繼資料。這項中繼資料提供服務專員所需的背景資訊，方便他們探索、瞭解及查詢資料，進而回答使用者問題。

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

  • 內部文件和 Wiki
  • 程式碼存放區
  • 通訊管道，例如 Google Chat 和 Slack

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

  • 擷取脈絡：從知識庫、文件、程式碼或即時通訊中擷取業務脈絡，充實技術中繼資料。
  • 產生文件：根據擷取的內容和其他資訊來源，產生 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
  • Knowledge Catalog API
  • Vertex AI API
  • Service Usage API

啟用 API 時所需的角色

如要啟用 API，您需要服務使用情形管理員 IAM 角色 (roles/serviceusage.serviceUsageAdmin)，其中包含 serviceusage.services.enable 權限。瞭解如何授予角色。

啟用 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. 複製 dataplex-labs 存放區，然後前往範例來源目錄：

    git clone https://github.com/GoogleCloudPlatform/dataplex-labs.git
cd dataplex-labs/knowledge_catalog_enrichment_agent/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) 中的檔案。如要在環境中套用這項工作流程，您可以自訂這些設定檔，將代理程式連結至內部 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],
    )
)

後續步驟