ADK マルチエージェントを BigQuery と連携させる

はじめに

この記事は Google ADK によるマルチエージェント社内ITヘルプデスク構築 をテーマにした全4回シリーズの第3回です。

第2回までで作ったエージェントは、FAQ データを Python のリストとして直接コードに書いていました。
デモには十分ですが、実運用を考えるといくつか困るところが出てきます。

• FAQ を 1 件追加するたびにコードを書き換えてデプロイし直す必要がある
• どんな問い合わせがどう処理されたかの履歴がどこにも残らない
• FAQ が増えてきたときにカテゴリ別の傾向を分析できない

本記事ではこれらの課題を解決するため、データソースをBigQueryに移します。

何を作ったか

全体構成

第2回までと同じ 3 つのエージェント構成は維持しつつ、データの置き場所を、Python のリストから BigQuery へと切り替えます。

ユーザー問い合わせ
    ↓
ルーターエージェント
    ↓
FAQ 検索エージェント ──[SELECT]──> faq_master テーブル
    │
    ├─ FAQ ヒット → 自分が回答 ──[INSERT resolved=true]──> inquiry_log テーブル
    │
    └─ FAQ なし
        ↓
        エスカレーションエージェント
        ↓ チケット情報を返す
        ──[INSERT resolved=false]──> inquiry_log テーブル

用意した 2 つのテーブル

inquiry_log にはsatisfaction_score（満足度スコア）カラムも事前に用意してあります。
本記事では NULL のままですが、第4回（評価）で活用します。

第2回からの変更点

設計のポイント

① キーワードは「固有な複合語」にする

faq_master の keywords カラムには、各 FAQ にひもづく検索ワードをカンマ区切りで入れます。
最初は素直に「アカウント」「ログイン」のような単語を入れていたのですが、これだと無関係な問い合わせまでヒットしてしまう問題が発生しました。

ユーザーの質問: 「自分のアカウントで身に覚えのないログイン履歴があります」
→ 本来はセキュリティインシデントとしてエスカレーションすべき
→ ところが「アカウント」「ログイン」というキーワードに引っかかり、
   パスワードリセット FAQ がヒットしてしまう

そこで、キーワードを「アカウントロック」「VPN接続」「パスワードを忘れ」のような固有な複合語に書き換えました。これだけで誤マッチが減少します。

単語そのものより「その FAQ でしか出てこない言い回し」を選ぶと精度が上がります。

② FAQ 検索は「ユーザー入力にキーワードが含まれるか」で判定

ユーザーが「VPN に接続しようとすると認証エラーが…」と長文で質問してきたとき、faq_master のどの行を返すべきか。
本記事では「keywords の各単語がユーザー入力に含まれていれば、その FAQ を返す」というシンプルな仕様にしました。SQL で書くと、keywords をカンマで分割し、各キーワードがユーザー入力に部分一致するかを STRPOS で判定する形になります。

SELECT faq_id, category, question, answer
FROM faq_master, UNNEST(SPLIT(keywords, ',')) AS kw
WHERE STRPOS(LOWER(@q), LOWER(TRIM(kw))) > 0
ORDER BY LENGTH(kw) DESC
LIMIT 1

ORDER BY LENGTH(kw) DESC で「より長い（具体的な）キーワード」に一致した FAQ を優先しています。
これで「アカウントロック」が「ロック」よりも先に評価されます。

BigQuery の全文検索 SEARCH 関数も検討しましたが、デフォルトのアナライザは日本語の形態素解析に対応しておらず、自然文クエリで期待どおりにヒットしませんでした。今回は仕様がシンプルな UNNEST + STRPOS 方式を採用しました。

③ ログ重複を防ぐ「責務分離」

inquiry_log への書き込みは、両方のエージェントが何の制限もなく行うと1 つの問い合わせで 2 行のログが残ってしまいます（FAQ 検索エージェントが書いた後、エスカレーションエージェントも書く）。
そこで、エージェントの instruction（指示書）で書き分けました。

• FAQ 検索エージェント: FAQ がヒットして自分が回答した場合 のみ ログを書く（resolved=true）
• エスカレーションエージェント: 必ずログを書く（resolved=false）

「最終応答を返したエージェントが 1 回だけログを書く」というルールにすることで、1 問い合わせ＝1 レコード を担保しています。

実装のキモ

本記事では設計判断が現れるポイントを抜粋します（エラーハンドリングや細かい整形処理は省略）。

FAQ 検索ツール（agents/faq_searcher.py）

search_faq() の中身は、上で説明した SQL を BigQuery に投げてヒットした 1 行を整形して返すだけです。

_BQ_CLIENT = bigquery.Client()  # モジュールトップで初期化（再利用される）

def search_faq(query: str) -> str:
    sql = """
    SELECT faq_id, category, question, answer
    FROM `helpdesk_demo.faq_master`,
         UNNEST(SPLIT(keywords, ',')) AS kw
    WHERE STRPOS(LOWER(@q), LOWER(TRIM(kw))) > 0
    ORDER BY LENGTH(kw) DESC
    LIMIT 1
    """
    # ... クエリを実行して結果を整形して返す

第2回では同じ関数の中で for faq in MOCK_FAQ_DATA: ... と Python ループしていた部分が、BigQuery への SELECT クエリに置き換わっただけです。
エージェントの定義側（instruction、tools の渡し方など）はほとんど変更していません。

ログ書き込みツール（agents/inquiry_logger.py）

新規ファイルです。BigQuery の inquiry_log テーブルに 1 行 INSERT するだけのシンプルなツール。

def log_inquiry(user_question, category, routed_to, agent_response, resolved):
    row = {
        "inquiry_id": str(uuid.uuid4()),
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "user_question": user_question,
        "category": category,
        "routed_to": routed_to,
        "agent_response": agent_response,
        "resolved": resolved,
        "satisfaction_score": None,
    }
    _BQ_CLIENT.insert_rows_json(_TABLE_ID, [row])

satisfaction_score は本記事では None のまま入れていますが、第4回（評価）で値を更新して使います。
このツールを faq_searcher_agent と escalation_agent の両方の tools に登録し、それぞれの instruction で「いつ呼ぶか」を書き分けることで、LLM が応答を返したあとに、文脈から判断してログを残す仕組みになります。

動作確認

ローカル実行

demo.py を実行すると、4 ケースの問い合わせがそれぞれ正しく振り分けられて応答が返ります（出力は冗長なので一部省略）。

[問い合わせ 1] VPN に接続しようとすると認証エラーが出て接続できません
→ FAQ 検索エージェントが NW-001 を返す（カテゴリ：ネットワーク系）

[問い合わせ 2] パスワードを忘れてしまいました
→ FAQ 検索エージェントが AC-001 を返す（カテゴリ：アカウント系）

[問い合わせ 3] 自分のアカウントで身に覚えのないログイン履歴があります
→ FAQ ヒットなし → エスカレーション（カテゴリ：セキュリティ、緊急度：高）

[問い合わせ 4] 社内の業務システムが全員使えなくなっています
→ FAQ ヒットなし → エスカレーション（カテゴリ：システム障害、緊急度：高）

inquiry_log に何が記録されたか

実行後に inquiry_log を SELECT すると、4 件分のレコードが残っていました。

各問い合わせが 1 行ずつ・最終応答したエージェントの記録として残っています。
ここで重要なのは、FAQ ヒットケースは resolved=true、エスカレーションケースは resolved=false で記録されている点です。
これによって、運用が始まったあとに「どれくらいの問い合わせを FAQ で自動解決できているか」を SQL ひとつで集計できるようになりました。

クラウド（Vertex AI Agent Engine）への再デプロイ

第2回でデプロイしたエージェントを、BigQuery 連携版に差し替えます。コード上の変更は requirements に google-cloud-bigquery を追加するだけ です。
ただし、デプロイ前にひとつだけ準備が要ります。

Reasoning Engine の SA に BigQuery 権限を付与

Vertex AI には、リソース管理用の汎用サービスアカウント（gcp-sa-aiplatform）と、Reasoning Engine 上でユーザーコードを実行する専用サービスアカウント（gcp-sa-aiplatform-re）があります。BigQuery にアクセスするのは後者なので、権限はこちらに付与する必要があります。
具体的には service-<番号>@gcp-sa-aiplatform-re.iam.gserviceaccount.com（-re サフィックス）です。この SA に BigQuery へのアクセス権限を 2 つ付与しないと、エージェントから faq_master を読んだり inquiry_log に書き込んだりができません。

権限さえ付与できれば、python deploy.py で約 5 分ほどで再デプロイが完了します。

クラウド側でも同じ動作を確認

クラウドにデプロイしたエージェントに同じ 4 問い合わせを送ったところ、ローカルと同じ応答が返り、inquiry_log にも同じスキーマで 4 行のレコードが追加されていました。

ローカルとクラウドで同じデータが同じスキーマで貯まっていく状態が完成しました。

ハマりどころ

まとめと次回について

第3回で達成したのは、見た目の派手さは少ないものの運用に近づくための地味で重要な変更です。

• FAQ がコードから切り離され、テーブルの行を追加するだけで増やせる
• どんな問い合わせがどう処理されたかが、すべて SQL で集計できる形で残る
• ローカルでもクラウドでも、同じスキーマでデータが貯まる
  これで 第4回（Vertex AI Evaluation でマルチエージェントの応答品質を測る） に進む土台が整いました。次回は inquiry_log に貯まったデータを評価のサンプルとして使い、ルーティング・回答・フォールバックの3軸でエージェントの応答品質を定量的に測ります。

参考リンク

• Google ADK ドキュメント
• Vertex AI Agent Engine ドキュメント
• BigQuery ドキュメント