はじめに
こんにちは!Ryuです。
AIエージェントを本番環境で運用したいと思ったとき、インフラ管理やスケーラビリティ、監視など多くの課題が出てきますよね。Google Cloud の Agent Engine は、そんな課題を一気に解決してくれる、AIエージェントのためのフルマネージドなデプロイメント環境です。
本記事では、Google Agent Development Kit(ADK)で作成したシンプルなAIエージェントを、Agent Engine にデプロイする方法を分かりやすくステップバイステップでご紹介していきます!
Agent Engineとは
Agent EngineはGoogle Cloud Vertex AI の一部として提供される、AIエージェントのためのフルマネージドなデプロイメントサービスです。
一言で魅力を説明すると、「エンジニアがAIエージェント開発のみに集中できるように、インフラやDBなどを全てフルマネージド二提供してくれるサービス」です。
前提条件
1. 開発環境
- Python 3.9以上 の実行環境
- Google Cloud プロジェクト の作成済み
2. Google Cloud 設定
- Vertex AI API が有効化済み
- 適切な権限(Vertex AI User, Editor等)が付与されたサービスアカウント
- Cloud Storage バケット(デプロイ用のステージング領域)
3. 必要なライブラリ
以下のライブラリをインストールしてください:
# Google ADK と関連ライブラリ
pip install google-adk==1.4.1
pip install google-genai==1.20.0
pip install google-cloud-aiplatform==1.97.0
4. 環境変数の設定
export GOOGLE_CLOUD_PROJECT="your-project-id"
export GOOGLE_CLOUD_LOCATION="us-central1"
export GOOGLE_APPLICATION_CREDENTIALS="path/to/service-account.json"
ステップ1: シンプルなエージェントの作成
それでは、Agent Engineにデプロイするシンプルなエージェントを作ってみましょう!
エージェントコードの作成
# agent.py
from google.adk.agents import Agent
# シンプルなカスタマーサポートエージェント
root_agent = Agent(
name="customer_support_agent",
model="gemini-2.0-flash-exp",
description="カスタマーサポート用AIアシスタント",
instruction="""
あなたはカスタマーサポート用のAIアシスタントです。
以下の点を意識して応答してください:
1. 丁寧で親しみやすい口調を保つ
2. 正確な情報を提供する
3. 不明な点は適切に案内する
4. 日本語で応答する
""",
tools=[], # 必要に応じてツールを追加
)
プロジェクト構成の整理
今回は以下のような構成です。
project/
├── agent.py # エージェント定義(上記コード)
├── deploy.py # デプロイスクリプト(設定も含む)
├── requirements.txt # 依存関係
└── pyproject.toml # プロジェクト設定
設定ファイルの作成
# config.py
import os
class Config:
# Google Cloud 設定
PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "your-project-id")
LOCATION = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1")
STAGING_BUCKET = f"gs://{PROJECT_ID}-agent-staging" # 自動生成
# Agent Engine 設定
AGENT_DISPLAY_NAME = "Production Customer Support Agent"
AGENT_DESCRIPTION = "Customer support agent powered by Google ADK"
# バージョン管理
VERSION = "1.0.0"
ステップ2: Agent Engineへのデプロイ
いよいよ作成したエージェントをAgent Engineにデプロイしてみます。
# deploy.py
from vertexai import agent_engines
from vertexai.preview import reasoning_engines
from agent import root_agent
import vertexai
import os
def main():
"""Agent Engineデプロイメインプロセス"""
print("=== Agent Engineデプロイ開始 ===")
# Google Cloud 設定
PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "your-project-id")
LOCATION = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1")
STAGING_BUCKET = f"gs://{PROJECT_ID}-agent-staging"
# Agent Engine 設定
AGENT_DISPLAY_NAME = "Production Customer Support Agent"
AGENT_DESCRIPTION = "Customer support agent powered by Google ADK"
# 重要: Vertex AI用の環境変数設定
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"
os.environ["GOOGLE_CLOUD_LOCATION"] = LOCATION
os.environ["GOOGLE_CLOUD_PROJECT"] = PROJECT_ID
# Vertex AI SDKの初期化
vertexai.init(
project=PROJECT_ID,
location=LOCATION,
staging_bucket=STAGING_BUCKET,
)
print(f"✅ Vertex AI初期化完了 - プロジェクト: {PROJECT_ID}")
# ADKエージェントをVertex AI形式に変換
app = reasoning_engines.AdkApp(
agent=root_agent,
enable_tracing=True, # 本番運用では必須
)
print(f"✅ エージェント変換完了 - {root_agent.name}")
# Agent Engineへのデプロイ実行
print("🚀 Agent Engineへのデプロイを開始...")
try:
remote_app = agent_engines.create(
agent_engine=app,
requirements=[
'google-adk==1.4.1',
'google-genai==1.20.0',
'google-cloud-aiplatform==1.97.0',
],
extra_packages=["."], # 現在のプロジェクトを含める
display_name=AGENT_DISPLAY_NAME,
description=AGENT_DESCRIPTION,
)
print(f"✅ デプロイ成功! リソース名: {remote_app.resource_name}")
return remote_app
except Exception as e:
print(f"❌ デプロイ失敗: {e}")
raise
print("=== デプロイ完了 ===")
if __name__ == "__main__":
remote_app = main()
デプロイ時のポイント
依存関係の管理
- バージョン固定: 本番環境では依存関係のバージョンを固定
- 最小限の依存: 必要最小限のパッケージのみを指定
トレーシングの有効化
enable_tracing=True # 本番運用では必須
これにより、Cloud Traceでエージェントの実行状況を詳細に監視できます。
リソース命名
display_name=config.AGENT_DISPLAY_NAME, # "Production Customer Support Agent"
description=config.AGENT_DESCRIPTION,
ステップ3: デプロイしたエージェントの動作確認
デプロイが完了したら、さっそく本番環境でエージェントをテストしてみましょう。
3-1. エージェントへの接続
# test_agent.py
from vertexai import agent_engines
from config import Config
def test_deployed_agent():
"""デプロイされたエージェントをテスト"""
config = Config()
# デプロイしたエージェントの取得
# 注意: resource_nameは前のステップのデプロイ結果から取得
resource_name = "projects/{project_id}/locations/{location}/reasoningEngines/{agent_id}"
try:
adk_app = agent_engines.get(resource_name)
print(f"✅ エージェント接続成功: {resource_name}")
return adk_app
except Exception as e:
print(f"❌ エージェント接続失敗: {e}")
raise
def create_user_session(adk_app, user_id="test_user_001"):
"""ユーザーセッションを作成"""
session = adk_app.create_session(user_id=user_id)
print(f"✅ セッション作成完了 (ユーザー: {user_id})")
return session
3-2. Pythonファイルでの対話テスト
デプロイしたエージェントの動作確認を、Pythonコードで実際にテストしてみましょう!
# test_chat.py
from vertexai import agent_engines
import os
def test_agent_chat():
"""デプロイしたエージェントと実際に対話テスト"""
# 環境変数設定
PROJECT_ID = os.environ.get("GOOGLE_CLOUD_PROJECT", "your-project-id")
# デプロイしたエージェントのリソース名
resource_name = "projects/{project_id}/locations/{location}/reasoningEngines/{agent_id}"
try:
# エージェントの取得
adk_app = agent_engines.get(resource_name)
print(f"✅ エージェント接続成功")
# セッションの作成
session = adk_app.create_session(user_id="test_user_001")
print(f"✅ セッション作成完了")
# 実際に対話テスト
test_messages = [
"こんにちは!初めて利用させていただきます。",
"サービスの使い方を教えてください。",
"ありがとうございました!"
]
print("✨ 対話テスト開始 ✨")
print("-" * 50)
for i, message in enumerate(test_messages, 1):
print(f"👤 ユーザー: {message}")
# エージェントにメッセージ送信
response = session.send(message)
print(f"🤖 エージェント: {response}")
print("-" * 50)
print("✅ 全てのテストが成功しました!")
except Exception as e:
print(f"❌ テスト失敗: {e}")
raise
if __name__ == "__main__":
test_agent_chat()
まとめ
いかがでしたでしょうか?本記事では、Google ADKで作成したシンプルなAIエージェントを、Agent Engineを使って本番環境にデプロイする手順をご紹介させていただきました。
いいねやストックがモチベーションに繋がるので、ぜひお願いします!