ADK でマルチエージェントを構築し Google Cloud にデプロイする

はじめに

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

前回では、Gemini CLI を使って GEMINI.md 1つで ITヘルプデスクエージェントを動かし、エージェントの基本的な動きを体感しました。
同時に、シングルエージェントには物足りなさを感じる場面も出てきました（FAQ データを参照できない・会話の文脈が続かない・複合問題を単一エージェントが全部引き受けてしまう）。

本記事ではその物足りなさを解消していきます。Google ADK（Agent Development Kit）を使い、「ルーター・FAQ検索・エスカレーション」の3つのエージェントが連携するマルチエージェントシステムを構築し、Google Cloud にデプロイします。

Google ADK とは

Google ADK（Agent Development Kit） は、Google が提供するエージェント開発用の Python フレームワークです。

第1回はファイル1枚でエージェントを動かしましたが、ADK ではコードを書く分だけ自由度が上がります。

構築するシステム

ユーザー問い合わせ
    ↓
① ルーターエージェント（受付・分類）
    └─ ② FAQ検索エージェント（BigQuery照会 → 回答生成）
           └─ ③ エスカレーションエージェント（チケット生成）

ADK のエージェントは親子構造で繋がります。親エージェントがユーザーの入力を受け取り、sub_agents として登録された子エージェントに処理を任せる、という流れです。
子エージェントはさらに孫エージェントを持てるので、3段階以上のチェーンも作れます。
*本記事ではモックデータを使って構造だけを作り、動作を確認します。

article2-adk-multiagent/
├── agents/
│   ├── __init__.py
│   ├── escalation.py   ← ③ エスカレーションエージェント
│   ├── faq_searcher.py ← ② FAQ検索エージェント（モックデータ）
│   └── router.py       ← ① ルーターエージェント（エントリーポイント）
├── .env                ← 環境変数（Google Cloud プロジェクト設定）
├── demo.py             ← ローカル動作確認スクリプト
├── deploy.py           ← Vertex AI Agent Engine デプロイスクリプト
└── requirements.txt

実装

③ エスカレーションエージェント（escalation.py）

他のエージェントを呼び出さない一番シンプルな構造なので、ここから作ります。

from google.adk.agents import Agent

escalation_agent = Agent(
    name="escalation_agent",
    model="gemini-2.5-flash",
    description=(
        "FAQ で解決できなかった問い合わせや、管理者権限・物理障害・"
        "セキュリティインシデントが必要な複雑なケースを担当する。"
        "有人対応チケットを生成して引き継ぎメッセージを返す。"
    ),
    instruction="""
あなたは社内 IT ヘルプデスクのエスカレーション担当エージェントです。
FAQ 検索エージェントが解決できなかった問い合わせを受け取り、有人対応チケットを作成します。

【有人対応チケット】
カテゴリ：（問い合わせのカテゴリ）
要約：（問い合わせ内容を 1〜2 文で要約）
緊急度：高 / 中 / 低
    """,
)

description は、親エージェントが「どのケースでこの子エージェントを呼ぶか」を判断するための説明文です。LLM がこのテキストを読んで「このエージェントに任せるべきか」を考えます。

コードは抜粋です。実装では instruction に出力フォーマットと、緊急度（高 / 中 / 低）の判定基準（例：高＝セキュリティインシデント疑い・業務停止）を続けて書いています。

② FAQ検索エージェント（faq_searcher.py）

from google.adk.agents import Agent
from agents.escalation import escalation_agent

# モック FAQ データ（記事3で BigQuery に差し替える）
# 実際にはネットワーク・アカウント・ハードウェア・ソフトウェアの4カテゴリ分を用意
MOCK_FAQ_DATA = [
    {
        "faq_id": "NW-001",
        "category": "ネットワーク系",
        "question": "VPN に接続できない / 認証エラーが出る",
        "answer": "...",
        "keywords": "VPN,接続できない,認証エラー,ログインできない",
    },
    # ... 他3カテゴリ（アカウント・ハードウェア・ソフトウェア）も同じ形式で用意
]

def search_faq(query: str) -> str:
    """FAQ データをキーワード検索する"""
    query_lower = query.lower()
    for faq in MOCK_FAQ_DATA:
        keywords = faq["keywords"].split(",")
        if any(kw.strip() in query_lower for kw in keywords):
            return f"カテゴリ：{faq['category']}\n\n【FAQ ID: {faq['faq_id']}】\n..."
    return "該当する FAQ が見つかりませんでした。"

faq_searcher_agent = Agent(
    name="faq_searcher_agent",
    model="gemini-2.5-flash",
    description="ネットワーク・アカウント・ハードウェア・ソフトウェアに関する IT 問い合わせを FAQ から検索して回答する。",
    instruction="search_faq ツールで検索し、FAQ が見つかった場合は回答する。見つからない場合は escalation_agent に引き継ぐ。",
    tools=[search_faq],           # Python 関数をツールとして渡す
    sub_agents=[escalation_agent], # 解決できない場合はエスカレーションへ
)

tools=[search_faq] と書くだけで、search_faq 関数がエージェントから「呼び出せるツール」として扱われます。
LLM が会話の流れから「いまツールを呼ぶべきか」を判断して、必要なら関数を実行してくれます。

コードは抜粋です。実装では instruction に役割・回答形式・注意事項（FAQ にない情報を勝手に補わない、見つからない場合は必ず escalation_agent に引き継ぐ等）を細かく書いています。

① ルーターエージェント（router.py）

from google.adk.agents import Agent
from agents.faq_searcher import faq_searcher_agent

router_agent = Agent(
    name="router_agent",
    model="gemini-2.5-flash",
    description="社内 IT ヘルプデスクのルーターエージェント。",
    instruction="""
すべての IT 問い合わせを受け取り、faq_searcher_agent に振り分けます。
faq_searcher_agent が FAQ 照会・回答生成・エスカレーション判定を担当します。
自分で回答を生成せず、必ず faq_searcher_agent に振り分けること。
    """,
    sub_agents=[faq_searcher_agent],
)

# エントリーポイント（ADK が認識するルートエージェント）
root_agent = router_agent

sub_agents に登録された子エージェントは、親エージェントから「呼び出せる相手」として扱われます。
ルーターが「どの子エージェントに任せるか」を、ユーザーの問い合わせ内容と各エージェントの description を読んで判断します。

コードは抜粋です。実装では instruction に対応カテゴリ（ネットワーク・アカウント・ハードウェア・ソフトウェア・セキュリティ）の一覧と、ルーター自身は回答を生成しないルールを明記しています。

ローカルで動かす

ADK では Runner にルートエージェントを渡し、セッションを作って問い合わせを流す構成になります。

from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from agents.router import root_agent

runner = Runner(agent=root_agent, app_name="helpdesk_demo",
                session_service=InMemorySessionService())

# session 作成 → runner.run_async() に問い合わせを渡し、
# event.is_final_response() で最終回答を拾う

.env ファイルで Vertex AI への接続先を設定します。

GOOGLE_GENAI_USE_VERTEXAI=1
GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=us-central1

実行結果

python demo.py で、用意した4ケースを順に投げて挙動を確認しました。代表として「FAQ で解決できるケース」と「エスカレーションが必要なケース」を1件ずつ載せます。
FAQ で解決できるケース（VPN 認証エラー）

ユーザー: VPN に接続しようとすると認証エラーが出て接続できません

エージェント:
カテゴリ：ネットワーク系

以下の手順をお試しください：
1. Caps Lock がオフになっているか確認する
2. パスワードを再入力する
3. MFA アプリの認証コードが最新か確認する
4. 別のネットワーク（テザリング等）に切り替えて試す
上記で解決しない場合は IT 担当者へお問い合わせください。

エスカレーションが必要なケース（セキュリティインシデント）

ユーザー: 自分のアカウントで身に覚えのないログイン履歴があります

エージェント:
申し訳ございません。お問い合わせの内容について、自動対応での解決が難しい状況です。
IT 担当者へ引き継ぎますので、以下のチケット情報をご確認ください。

【有人対応チケット】
カテゴリ：セキュリティ
要約：アカウントに身に覚えのないログイン履歴がある
緊急度：高

他2ケース（パスワードリセット・全社障害）も期待どおり、それぞれ FAQ 回答とエスカレーションチケットを返しました。ルーター → FAQ検索 → エスカレーションという3段階の引き渡しが、if 文によるルールを1行も書かずに LLM の判断だけで動いています。

Google Cloud へのデプロイ（Vertex AI Agent Engine）

ローカルで動作確認できたエージェントを、deploy.py で Google Cloud にデプロイします。

事前準備：ステージング用 GCS バケットの作成

Vertex AI Agent Engine はデプロイ時にコードや依存関係を Cloud Storage 経由で受け渡すため、ステージング用バケットが必須です。バケット名は一意なので、helpdesk-agent-staging-<任意の文字列> のように自分の環境で重複しない名前に置き換えてください。

gcloud storage buckets create gs://helpdesk-agent-staging-<任意の固有文字列> \
    --project=your-project-id \
    --location=us-central1 \
    --uniform-bucket-level-access

.env の追記

ローカル実行時の .env に、ステージングバケットを1行追記します。

GOOGLE_CLOUD_STAGING_BUCKET=gs://helpdesk-agent-staging-<任意の固有文字列>

deploy.py

import os

import vertexai
from dotenv import load_dotenv
from vertexai import agent_engines

from agents.router import root_agent

# .env の値をシェル環境変数より優先（シェルに古い GOOGLE_CLOUD_LOCATION
# が残っていて意図しないリージョンに飛ぶ事故を防ぐ）
load_dotenv(override=True)

vertexai.init(
    project=os.environ["GOOGLE_CLOUD_PROJECT"],
    location=os.environ["GOOGLE_CLOUD_LOCATION"],
    staging_bucket=os.environ["GOOGLE_CLOUD_STAGING_BUCKET"],
)

# エージェントを AdkApp でラップ（Agent Engine が認識できる形式に変換）
adk_app = agent_engines.AdkApp(agent=root_agent, enable_tracing=True)

# extra_packages: agents/ パッケージをコンテナに同梱
# （これがないと Reasoning Engine 環境で from agents.router import ... に失敗する）
remote_agent = agent_engines.create(
    adk_app,
    requirements=[
        "google-adk>=1.0.0",
        "google-cloud-aiplatform[agent_engines,adk]>=1.93.0",
    ],
    extra_packages=["agents"],
    display_name="IT ヘルプデスク マルチエージェント",
    description="ルーター・FAQ検索・エスカレーションの3エージェント構成",
)

print(f"デプロイ完了: {remote_agent.resource_name}")

ハマりどころ

• extra_packages=["agents"] を忘れると、コンテナ起動時に ModuleNotFoundError: No module named 'agents' で失敗します。AdkApp は root_agent を pickle 化して送りますが、agents/ ディレクトリ自体は別途同梱する必要があります。
• staging_bucket を省略すると ValueError: Please provide a staging_bucket で停止します（最近の SDK で必須化されました）。
• シェル環境変数 GOOGLE_CLOUD_LOCATION などが既に設定されていると .env の値より優先されるため、load_dotenv(override=True) で .env を適用するのが安全です。

実行結果

$ python deploy.py
（agent_engine.pkl・requirements.txt・dependencies.tar.gz をステージングへアップロード）
Creating AgentEngine
Create AgentEngine backing LRO: projects/<YOUR_PROJECT_NUMBER>/locations/us-central1/reasoningEngines/<YOUR_REASONING_ENGINE_ID>/operations/...
デプロイ完了: projects/<YOUR_PROJECT_NUMBER>/locations/us-central1/reasoningEngines/<YOUR_REASONING_ENGINE_ID>

ステージングへのアップロードからコンテナビルド、Reasoning Engine の起動まで、デプロイには5分ほどかかりました。

デプロイしたエージェントを呼び出す

デプロイしたエージェントは、リソース名から agent_engines.get() で取得して呼び出します。セッションを作ってから stream_query で問い合わせを送る流れです（ADK エージェントは streaming で結果を受け取る形のみ用意されています）。

from vertexai import agent_engines

# `deploy.py` の実行結果に出力された自分のリソース名に置き換える
remote_agent = agent_engines.get(
    "projects/<YOUR_PROJECT_NUMBER>/locations/us-central1/reasoningEngines/<YOUR_REASONING_ENGINE_ID>"
)

session = remote_agent.create_session(user_id="blog-demo-user")
for event in remote_agent.stream_query(
    user_id="blog-demo-user",
    session_id=session["id"],
    message="VPN に接続できません",
):
    for part in (event.get("content") or {}).get("parts", []):
        if text := part.get("text"):
            print(text, end="")

※冒頭に load_dotenv(override=True) と vertexai.init(project=..., location=...) が必要です。
クラウド側での実行結果（VPN 認証エラーのケースを抜粋。他3ケースもローカルと同じ結果になりました）

ユーザー: VPN に接続しようとすると認証エラーが出て接続できません

エージェント:
カテゴリ：ネットワーク系
以下の手順をお試しください：
1. Caps Lock がオフになっているか確認する
2. パスワードを再入力する
3. MFA アプリの認証コードが最新か確認する
4. 別のネットワーク（テザリング等）に切り替えて試す

上記で解決しない場合は IT 担当者へお問い合わせください。

コスト管理

Reasoning Engine はデプロイした瞬間から課金が始まります。動作確認だけなら、終わったらすぐ削除しておくと安心です。

gcloud ai reasoning-engines delete <RESOURCE_ID> \
    --region=us-central1 \
    --project=your-project-id

まとめと次回について

ADK を使って3つのエージェントが連携するマルチエージェントシステムを構築し、Google Cloud にデプロイできました。

現時点では FAQ データはモックです。
次回は BigQuery に実際の FAQ データを格納し、FAQ検索エージェントが BigQuery にクエリを実行する構成に差し替えます。

参考リンク

• Google ADK ドキュメント
• Vertex AI Agent Engine ドキュメント
• google-adk PyPI