G-gen の堂原です。本記事では Google Cloud の Agent Development Kit を用いて開発したエージェントを、Vertex AI Agent Engine にデプロイし、呼び出す方法について紹介します。

はじめに

本記事の趣旨

本記事では、Agent Development Kit を使用してエージェントを開発し、Vertex AI Agent Engine にデプロイして実際に呼び出すまでの一連の流れを、具体的なコードを交えながら解説します。

Agent Development Kit

Agent Development Kit（以下、ADK）は、Google Cloud が提供する、AI エージェントを構築するための Python ライブラリです。ADK を使用することで、エージェントへの関数処理組み込みや外部ツールの利用、マルチエージェントの実装が行えます。

参考 : Agent Development Kit

Agent Engine

Vertex AI Agent Engine（以下、Agent Engine）は、AI エージェントの実行環境を提供するフルマネージドサービスです。

Agent Engine にはマルチターン実装のためのセッション機能が組み込まれており、通常 Gemini を用いるときのようにこれまでの会話履歴を都度リクエストに含める必要がありません。その他、自動スケーリング機能や VPC Service Controls による保護など、AI エージェントをプロダクトレベルで利用していくための機能が備わっています。

2025年5月現在、GUI は提供されておらず、Python SDK または REST API を用いて各種操作を行う必要があります。

参考 : Vertex AI Agent Engine の概要 | Generative AI on Vertex AI | Google Cloud

ADK を用いた AI エージェント開発

題材として、ユーザが都市の名前を与えると、その都市の現在時刻を回答するエージェントを開発します。以下のページで、ADK を用いて AI エージェントを作成する手順が紹介されているので、それをベースとしています。

参考 : Quickstart - Agent Development Kit

import datetime
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
from google.adk.agents import Agent
 
def get_current_time(city: str) -> dict:
    """Returns the current time in a specified city.
 
    Args:
        city (str): The IANA zone name of the city for which to retrieve the current time.
 
    Returns:
        dict: status and result or error msg.
    """
 
    try:
        tz = ZoneInfo(city)
        now = datetime.datetime.now(tz)
        report = (
            f'The current time in {city} is {now.strftime("%Y-%m-%d %H:%M:%S %Z%z")}'
        )
        return {"status": "success", "report": report}
 
    except ZoneInfoNotFoundError:
        return {
            "status": "error",
            "error_message": (
                f"Sorry, I don't have timezone information for {city}."
            ),
        }
 
 
root_agent = Agent(
    name="city_time_agent",
    model="gemini-2.0-flash",
    description=(
        "(Japanese) Agent to answer questions about the time in a city."
    ),
    instruction=(
        "You are a helpful agent who answers questions written in Japanese, responding in Japanese, about the time in a city."
        "First, obtain the IANA zone name for the city specified by the user."
        "Next, retrieve the current time in that city based on the IANA zone name."
    ),
    tools=[get_current_time],
)

このコードでは、Agent クラスのインスタンスとして root_agent を作成しています。model には gemini-2.0-flash を指定し、tools には get_current_time 関数をリストとして渡しています。instruction では、AI エージェントの振る舞いを指示しています。

ADK で作成した AI エージェントは、関数（ここでいう get_current_time）の名前および docstring を元に呼び出すべき関数を判断しているため、docstring は適切に記載することが推奨されます。

参考 : Function tools - Agent Development Kit

Agent Engine へのデプロイ

開発したエージェントのソースコードを、Agent Engine にデプロイします。デプロイにあたり、以下のソースコードを実行します。

import vertexai
from vertexai import agent_engines
 
PROJECT_ID = "{プロジェクトID}"
LOCATION = "us-central1"
STAGING_BUCKET = "gs://{バケット名}"
 
vertexai.init(
    project=PROJECT_ID,
    location=LOCATION,
    staging_bucket=STAGING_BUCKET,
)
 
remote_app = agent_engines.create(
    display_name="city_time_agent",
    description="(Japanese) Agent to answer questions about the time in a city.",
    agent_engine=root_agent,
    requirements=[
        "google-cloud-aiplatform[adk,agent_engines]"
    ]
)

STAGING_BUCKET には、AI エージェントをデプロイする際の中間ファイルを格納するために使われる Cloud Storage バケットを指定します。

remote_app の display_name と description はメタデータであり、デプロイする AI エージェントの挙動に影響は与えません。agent_engine で先ほどインスタンス化した root_agent を指定し、requirements に AI エージェントが必要とするライブラリを記載します。

上記のコードを実行すると、出力の一部として以下が得られます。

Creating AgentEngine
Create AgentEngine backing LRO: projects/{プロジェクト番号}/locations/us-central1/reasoningEngines/{リソースID}/operations/xxx

この出力に含まれる {リソースID} が、デプロイされた Agent Engine のリソース ID となります。この ID は後ほどエージェントを呼び出す際に使用します。

デプロイした AI エージェントの呼び出し

デプロイした AI エージェントを呼び出すには、まずその AI エージェントのインスタンスを取得します。先ほど取得したリソース ID を使用します。

Agent Engine へのデプロイ後にそのまま処理を続ける場合は本コードは不要で、agent_engines.create() の返り値をそのまま利用できます。

remote_agent = agent_engines.get("{リソースID}")

参考 : エージェントを使用する | Generative AI on Vertex AI | Google Cloud

次に、取得した AI エージェントに対して新しいセッションを作成し、クエリを投げます。以下のコードは、「東京の現在時刻を教えて下さい」というクエリを投げ、その結果をストリーミングで受け取る例です。

user_id = "hoge"
session = remote_agent.create_session(user_id=user_id)
 
for event in remote_agent.stream_query(
    user_id=user_id,
    session_id=session["id"],
    message="東京の現在時刻を教えて下さい",
):
    print(event)

参考 : Agent Development Kit エージェントを使用する | Generative AI on Vertex AI | Google Cloud

上記のコードを実行すると、以下のような出力が得られます。

{'content': {'parts': [{'function_call': {'id': 'xxx', 'args': {'city': 'Asia/Tokyo'}, 'name': 'get_current_time'}}], 'role': 'model'}, 'invocation_id': 'xxx', 'author': 'city_time_agent', 'actions': {'state_delta': {}, 'artifact_delta': {}, 'requested_auth_configs': {}}, 'long_running_tool_ids': [], 'id': xxx', 'timestamp': xxx}
{'content': {'parts': [{'function_response': {'id': 'xxx', 'name': 'get_current_time', 'response': {'status': 'success', 'report': 'The current time in Asia/Tokyo is 2025-04-28 18:35:35 JST+0900'}}}], 'role': 'user'}, 'invocation_id': 'xxx', 'author': 'city_time_agent', 'actions': {'state_delta': {}, 'artifact_delta': {}, 'requested_auth_configs': {}}, 'id': 'xxx', 'timestamp': xxx}
{'content': {'parts': [{'text': '東京の現在の時刻は2025-04-28 18:35:35 JST+0900です。'}], 'role': 'model'}, 'invocation_id': 'xxx', 'author': 'city_time_agent', 'actions': {'state_delta': {}, 'artifact_delta': {}, 'requested_auth_configs': {}}, 'id': 'xxx', 'timestamp': xxx}

3 つ目の出力内に、AI エージェントの生成文が含まれており、東京の現在時刻が回答されていることがわかります。また、AI エージェントが get_current_time 関数を呼び出し、その結果に基づいて応答を生成している様子が確認できます。

同じ user_id と session_id を使用してクエリを投げ続けることで、マルチターンな会話も可能です。

なお、今回はローカルで Python のコードを実行してエージェントを呼び出しましたが、Agent Engine にホストしたエージェントは、別のアプリケーションやチャットツールなどから呼び出すことが可能です。また、今後は Google Agentspace など他の Google サービスとの連携も強化されていくとみられています。

AI エージェントの削除

エージェントを削除する前に、そのエージェントに関連付けられているすべてのセッションを削除しておく必要があります。

以下のコードは、指定した user_id に関連付けられているすべてのセッションを削除し、その後エージェント自体を削除する例です。

user_id = "hoge"
session_list = remote_agent.list_sessions(user_id=user_id)
 
for session in session_list["sessions"]:
    remote_agent.delete_session(user_id=user_id, session_id=session["id"])
 
remote_agent.delete()