メタデータを拡充するエージェントを構築する

Knowledge Catalog（旧称: Dataplex Universal Catalog）は、組織全体のデータアセットのメタデータを管理します。このメタデータは、エージェントがユーザーの質問に回答するために必要なデータを検出、理解、クエリするために使用するコンテキストを提供します。

Knowledge Catalog はリソースの管理、技術スキーマの追跡、説明とデータプロファイルの生成を自動的に行いますが、価値のあるビジネス コンテキストは、次のような他の場所に存在することがよくあります。

  • 内部ドキュメントと Wiki
  • コード リポジトリ
  • Google Chat や Slack などのコミュニケーション チャネル

AI エージェントを構築して、これらのソースからコンテキストを抽出し、メタデータを大規模に継続的に拡充できます。このチュートリアルでは、knowledge-catalog リポジトリのサンプルコードを使用して、次の処理を行うエージェントを構築する方法について説明します。

  • コンテキストの抽出: ナレッジベース、ドキュメント、コード、チャットからビジネス コンテキストを抽出し、技術メタデータを拡充します。
  • ドキュメントの生成: 抽出されたコンテキストとその他の情報源に基づいて、BigQuery テーブルのドキュメントを生成します。
  • 検索と見つけやすさの改善: 生成されたドキュメントを Knowledge Catalog に公開し、検索を通じてエントリを見つけやすくします。

始める前に

Knowledge Catalog 拡充エージェントを実行するには、次の要件を満たしている必要があります。

必要なロール

拡充エージェントを使用するために必要な権限を取得するには、プロジェクトに対する次の IAM ロールの付与を管理者に依頼してください。 Google Cloud iam.gserviceaccount.com

ロールの付与については、プロジェクト、フォルダ、組織へのアクセス権の管理をご覧ください。

これらの事前定義ロールには 拡充エージェントの使用に必要な権限が含まれています。必要とされる正確な権限については、「必要な権限」セクションを開いてご確認ください。

必要な権限

拡充エージェントを使用するには、次の権限が必要です。

  • 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 を有効にするには、serviceusage.services.enable 権限が必要です。プロジェクトを作成した場合は、オーナーロール（roles/owner）を通じてこの権限が付与されている可能性があります。それ以外の場合は、Service Usage 管理者ロール（roles/serviceusage.serviceUsageAdmin）を通じてこの権限を取得できます。ロールを付与する方法をご確認ください。

API を有効にする

依存関係のインストール

サンプルを実行するには、次の Python パッケージとツールが必要です。

  • google-adk（Agent Development Kit（ADK））
  • google-cloud-dataplex Knowledge Catalog Python クライアント
  • google-auth はアプリケーションのデフォルト認証情報を管理します
  • サンプル MCP サーバーを構築するための mcp[cli]
  • 認証と構成のための 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 ディレクトリに複数のドキュメントも含まれています。これらのドキュメントはローカル ナレッジベースを形成します。拡充エージェントは、このナレッジベースを使用して情報を抽出し、ドキュメントを作成します。

メタデータをダウンロードする

まず、ダウンロード ツールを実行して、BigQuery データセットとそのテーブルのメタデータ スナップショットを Knowledge Catalog から抽出します。これにより、ローカル メタデータ アーティファクトが作成されます。

--dir 引数で、メタデータ ファイルが書き込まれるディレクトリを指定します。

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

このスクリプトは、sample/metadata ディレクトリに、次の命名規則を使用してテーブルごとに 1 つの 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],
    )
)

次のステップ