🦧

Google ADK × Agent Engineで複数エージェントをデプロイするリポジトリ構成

2025/11/02に公開

Agent Development Kit

tech

Google Cloud の Vertex AI Agent Engine を使用すると、AI エージェントをクラウド上にデプロイして実行できる。

Agent Engine へのデプロイを自動化する構成パターン
動的インポートを使った柔軟なエージェント管理
デプロイ済みエージェントのテスト方法

プロジェクト構成

ai-agent/
├── .env                      # 環境変数
├── pyproject.toml            # 依存関係定義
├── requirements.txt          # 自動生成される依存関係ファイル
├── agents/                   # エージェント定義
│   └── {agent_name}/
│       ├── __init__.py
│       └── agent.py          # root_agent を定義
└── deployment/               # デプロイ・テストスクリプト
    ├── deploy.py             # デプロイスクリプト
    └── test_agent_engine.py  # テストスクリプト

環境変数の設定

.env ファイルに以下の環境変数を設定

GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=us-central1
GOOGLE_CLOUD_LOCATION_AGENT_ENGINE=asia-northeast1
STAGING_BUCKET=gs://your-staging-bucket
GOOGLE_GENAI_USE_VERTEXAI=True

依存関係の管理

pyproject.toml で依存関係を管理

[project]
name = "ai-agent"
version = "0.1.0"
requires-python = ">=3.13"
dependencies = [
    "google-cloud-aiplatform[adk,agent-engines]>=1.124.0",
    "python-dotenv",
]

デプロイスクリプトの実装

処理の流れ

デプロイスクリプト (deploy.py) は以下の流れで処理

実装のポイント

1. コマンドライン引数の処理

argparse を使用してエージェント名を指定できるようにする。

import argparse

parser = argparse.ArgumentParser(description='Deploy agent to Vertex AI')
parser.add_argument('--agent', required=True,
                   help='Agent name (folder name under agents/)')
args = parser.parse_args()

2. 環境変数の読み込み

python-dotenv を使用して .env ファイルから環境変数を読み込む。

from pathlib import Path
from dotenv import load_dotenv
import os

env_path = Path(__file__).parent.parent / '.env'
load_dotenv(dotenv_path=env_path)

GOOGLE_CLOUD_PROJECT = os.getenv('GOOGLE_CLOUD_PROJECT')
GOOGLE_CLOUD_LOCATION = os.getenv('GOOGLE_CLOUD_LOCATION_AGENT_ENGINE')
STAGING_BUCKET = os.getenv('STAGING_BUCKET')

3. requirements.txt の自動生成

デプロイ時に uv pip compile を実行して、最新の依存関係を反映します。

import subprocess
import sys

project_root = Path(__file__).parent.parent
pyproject_path = project_root / 'pyproject.toml'
requirements_path = project_root / 'requirements.txt'

print("Generating requirements.txt from pyproject.toml...")
result = subprocess.run(
    ['uv', 'pip', 'compile', str(pyproject_path), '-o', str(requirements_path)],
    cwd=project_root,
    capture_output=True,
    text=True
)

if result.returncode != 0:
    print(f"Error generating requirements.txt: {result.stderr}")
    sys.exit(1)

4. エージェントの動的インポート

importlib を使用して、指定されたエージェントを動的に読み込む。

import importlib.util

agent_path = Path(__file__).parent.parent / 'agents' / args.agent / 'agent.py'
if not agent_path.exists():
    raise FileNotFoundError(f"Agent file not found: {agent_path}")

spec = importlib.util.spec_from_file_location("agent_module", agent_path)
agent_module = importlib.util.module_from_spec(spec)
sys.modules["agent_module"] = agent_module
spec.loader.exec_module(agent_module)
root_agent = agent_module.root_agent

5. Vertex AI へのデプロイ

Vertex AI SDK を初期化し、エージェントをデプロイします。

import vertexai
from vertexai import agent_engines

# SDKの初期化
vertexai.init(
    project=GOOGLE_CLOUD_PROJECT,
    location=GOOGLE_CLOUD_LOCATION,
    staging_bucket=STAGING_BUCKET,
)

# エージェントをAdkAppでラップ
app = agent_engines.AdkApp(
    agent=root_agent,
    enable_tracing=True,
)

# デプロイ実行
remote_app = agent_engines.create(
    agent_engine=app,
    requirements=str(requirements_path)
)

print(f"Deployment finished!")
print(f"Resource Name: {remote_app.resource_name}")

実行方法

cd ai-agent
uv run deployment/deploy.py --agent my_agent

テストスクリプトの実装

処理の流れ

テストスクリプト (test_agent_engine.py) は以下の流れで処理します。

実装のポイント

1. コマンドライン引数の処理

デプロイ時に出力された Resource Name を指定します。

parser = argparse.ArgumentParser(
    description='Test a deployed agent on Vertex AI Agent Engine'
)
parser.add_argument(
    '--resource-name',
    required=True,
    help='Resource name of the deployed agent'
)
args = parser.parse_args()

2. 非同期処理の実装

Agent Engine API は非同期処理を使用します。

import asyncio

async def test_agent(resource_name: str, user_id: str, test_message: str):
    """Test the deployed agent by creating a session and sending a query."""

    # エージェントに接続
    remote_app = agent_engines.get(resource_name)

    # セッションを作成
    remote_session = await remote_app.async_create_session(user_id=user_id)
    print(f"Session Details: {remote_session}")

    # クエリを送信
    async for event in remote_app.async_stream_query(
        user_id=user_id,
        session_id=remote_session["id"],
        message=test_message,
    ):
        print(event)

3. メイン関数での実行

asyncio.run() を使用して非同期関数を実行します。

def main():
    # 引数パース、環境変数読み込み、SDK初期化
    # ...

    # テストパラメータ
    user_id = "test_user"
    test_message = "Hello, this is a test message. Can you help me?"

    # 非同期テストを実行
    asyncio.run(test_agent(args.resource_name, user_id, test_message))

if __name__ == "__main__":
    main()

実行方法

cd ai-agent
uv run deployment/test_agent_engine.py --resource-name projects/{PROJECT_NUMBER}/locations/{LOCATION}/reasoningEngines/{RESOURCE_ID}

実行結果の例

============================================================
Testing Agent: projects/.../reasoningEngines/...
============================================================

Connecting to deployed agent...
✓ Successfully connected to agent

Creating session for user: test_user
✓ Session created successfully
  Session Details: {'events': [], 'user_id': 'test_user', 'state': {}, 'id': '...'}

Sending test message: 'Hello, this is a test message. Can you help me?'

============================================================
Agent Response (streaming):
============================================================

{'parts': [...], 'role': 'model'}
...

============================================================
✓ Test completed successfully
============================================================

アーキテクチャ解説

全体のフロー

設計の利点

1. 動的インポートによる柔軟性

エージェント名をコマンドライン引数で指定することで、複数のエージェントを同じスクリプトでデプロイできます。

# エージェントAをデプロイ
uv run deployment/deploy.py --agent agent_a

# エージェントBをデプロイ
uv run deployment/deploy.py --agent agent_b

2. 依存関係の自動管理

pyproject.toml を更新するだけで、デプロイ時に自動的に最新の requirements.txt が生成されます。手動での同期作業が不要になります。

3. 環境変数の一元管理

.env ファイルで環境変数を管理することで、プロジェクト固有の設定を簡単に変更できます。

4. テストの自動化

テストスクリプトにより、デプロイ後すぐに動作確認ができます。CI/CD パイプラインにも組み込みやすい構成です。

エージェント定義の例

agents/my_agent/agent.py の実装例です。

from vertexai.preview.agent_engines import Agent

# エージェントの定義
root_agent = Agent(
    name="my_agent",
    model="gemini-2.0-flash-exp",
    instructions="You are a helpful assistant.",
    # その他の設定...
)

まとめ

重要なポイント

動的インポート: importlib を使用して、複数のエージェントを柔軟に管理
依存関係の自動生成: デプロイ時に uv pip compile で最新の依存関係を反映
環境変数の管理: .env ファイルで設定を一元管理
非同期テスト: asyncio を使用してデプロイ済みエージェントを効率的にテスト

実用的な活用シーン

複数環境の管理: 開発・ステージング・本番環境ごとに .env を切り替え
CI/CD パイプライン: GitHub Actions などで自動デプロイ・テストを実行
バージョン管理: エージェントごとにディレクトリを分けて、独立して管理

この実装パターンを応用することで、AI エージェントの開発・デプロイサイクルを大幅に効率化できます。

参考資料

箇所	詳細	URL
デプロイの基本	Vertex AI Agent Engine へのデプロイの基本ドキュメント	https://google.github.io/adk-docs/deploy/agent-engine/#deploy-to-vertex-ai-agent-engine
	デプロイ時の`requirement.txt`出力フローの参考実装	https://github.com/google/adk-samples/blob/main/python/agents/data-science/deployment/deploy.py
	デプロイ後のテストに関する参考実装	https://github.com/google/adk-samples/blob/main/python/agents/data-science/deployment/test_deployment.py
	`deploy.py`の実行方法に関する参考情報	https://qiita.com/nakamasato/items/97dad767e3b3de598b7e#:~:text=stream_query(%20user_id=user_id%2C%20message,existing%20remote%20agent.%22%22%22
リポジトリ構成	サンプルエージェントのリポジトリ構成の参考例	https://github.com/google/adk-samples/tree/main/python/agents/data-science

プロジェクト構成

環境変数の設定

依存関係の管理

デプロイスクリプトの実装

処理の流れ

実装のポイント

1. コマンドライン引数の処理

2. 環境変数の読み込み

3. requirements.txt の自動生成

4. エージェントの動的インポート

5. Vertex AI へのデプロイ

実行方法

テストスクリプトの実装

処理の流れ

実装のポイント

1. コマンドライン引数の処理

2. 非同期処理の実装

3. メイン関数での実行

実行方法

実行結果の例

アーキテクチャ解説

全体のフロー

設計の利点

1. 動的インポートによる柔軟性

2. 依存関係の自動管理

3. 環境変数の一元管理

4. テストの自動化

エージェント定義の例

まとめ

重要なポイント

実用的な活用シーン

参考資料

Discussion