Agent Development Kit エージェントを使用する

始める前に

このチュートリアルは、次の手順を読んで理解していることを前提としています。

• Agent Development Kit エージェントを開発する: AdkApp のインスタンスとして agent を開発します。
• ユーザー認証: エージェントにクエリを実行するユーザーとして認証します。
• SDK をインポートして初期化する: デプロイされたインスタンスを取得するためにクライアントを初期化します（必要な場合）。

エージェントのインスタンスを取得する

AdkApp にクエリを実行するには、まず新しいインスタンスを作成するか、既存のインスタンスを取得する必要があります。

特定のリソース ID に対応する AdkApp を取得するには:

Vertex AI SDK for Python

次のコードを実行します。

import vertexai

client = vertexai.Client(  # For service interactions via client.agent_engines
    project="PROJECT_ID",
    location="LOCATION",
)

adk_app = client.agent_engines.get(name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID")

print(adk_app)

ここで

• PROJECT_ID は、エージェントを開発してデプロイする Google Cloud プロジェクト ID です。
• LOCATION は、サポートされているリージョンの一つです。
• RESOURCE_ID は、reasoningEngine リソースとしてのデプロイ済みエージェントの ID です。

Python リクエスト ライブラリ

次のコードを実行します。

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests

def get_identity_token():
    credentials, _ = google_auth.default()
    auth_request = google_requests.Request()
    credentials.refresh(auth_request)
    return credentials.token

response = requests.get(
f"https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    headers={
        "Content-Type": "application/json; charset=utf-8",
        "Authorization": f"Bearer {get_identity_token()}",
    },
)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID

Vertex AI エクスプレス モード

Vertex AI SDK for Python は、エクスプレス モードの API キーで使用できます。

import vertexai

client = vertexai.Client(  # For service interactions via client.agent_engines
    api_key="API_KEY",
)

adk_app = client.agent_engines.get(name="AGENT_ENGINE_NAME")

print(adk_app)

ここで

• AGENT_ENGINE_NAME は、Agent Engine リソースの完全修飾名です。

注: 次のサポートされているオペレーションはすべて、Vertex AI Express Mode ユーザー向けに Vertex AI SDK for Python

でサポートされています。

Vertex AI SDK for Python を使用する場合、adk_app オブジェクトは、次のものを含む AgentEngine クラスに対応します。

• デプロイされたエージェントに関する情報を含む adk_app.api_resource。adk_app.operation_schemas() を呼び出して、adk_app がサポートするオペレーションのリストを返すこともできます。詳しくは、サポートされているオペレーションをご覧ください。
• 同期サービス インタラクションを可能にする adk_app.api_client
• 非同期サービス インタラクションを可能にする adk_app.async_api_client

このセクションの残りの部分では、adk_app という名前の AgentEngine インスタンスがあることを前提としています。

サポートされているオペレーション

AdkApp でサポートされているオペレーションは次のとおりです。

• async_stream_query: クエリへのレスポンスをストリーミングします。

• async_create_session: 新しいセッションを作成します。

• async_list_sessions: 利用可能なセッションを一覧表示します。

• async_get_session: 特定のセッションを取得します。

• async_delete_session: 特定のセッションを削除します。

• async_add_session_to_memory: セッションのメモリを生成します。

• async_search_memory: メモリを取得します。

サポートされているすべてのオペレーションを一覧表示するには:

Vertex AI SDK for Python

次のコードを実行します。

adk_app.operation_schemas()

Python リクエスト ライブラリ

次のコードを実行します。

import json

json.loads(response.content).get("spec").get("classMethods")

REST API

curl リクエストに対するレスポンスから spec.class_methods で表されます。

セッションを管理する

AdkApp は、エージェントを Vertex AI Agent Engine にデプロイした後はクラウドベースのマネージド セッションを使用します。このセクションでは、マネージド セッションの使用方法について説明します。

注: この手順は、エージェントの開発時に session_service_builder を使用してデータベースをカスタマイズしていないことを前提としています。

セッションを作成する

ユーザーのセッションを作成するには:

Vertex AI SDK for Python

session = await adk_app.async_create_session(user_id="USER_ID")

print(session)

Python リクエスト ライブラリ

次のコードを実行します。

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_create_session",
    "input": {"user_id": "USER_ID"},
  }),
)
print(response.content)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_create_session", "input": {"user_id": "USER_ID"},}'

• USER_ID: 128 文字以内でユーザー ID を選択します。例: user-123

セッションは、ADK セッション オブジェクトの辞書表現として作成されます。

セッションを一覧表示する

ユーザーのセッションを一覧表示するには:

Vertex AI SDK for Python

response = await adk_app.async_list_sessions(user_id="USER_ID"):
for session in response.sessions:
    print(session)

Python リクエスト ライブラリ

次のコードを実行します。

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_list_sessions",
    "input": {"user_id": "USER_ID"},
  }),
)
print(response.content)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_list_sessions", "input": {"user_id": "USER_ID"},}'

ここで、USER_ID は定義したユーザー ID です。例: user-123

セッションが返された場合、ADK セッション オブジェクトの辞書形式が使用されます。

セッションを取得する

特定のセッションを取得するには、ユーザー ID とセッション ID の両方が必要です。

Vertex AI SDK for Python

session = await adk_app.async_get_session(user_id="USER_ID", session_id="SESSION_ID")

print(session)

Python リクエスト ライブラリ

次のコードを実行します。

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_get_session",
    "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},
  }),
)
print(response.content)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_get_session", "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},}'

session は、ADK セッション オブジェクトの辞書表現です。

セッションを削除する

セッションを削除するには、ユーザー ID とセッション ID の両方が必要です。

Vertex AI SDK for Python

await adk_app.async_delete_session(user_id="USER_ID", session_id="SESSION_ID")

Python リクエスト ライブラリ

次のコードを実行します。

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests
import json

def get_identity_token():
  credentials, _ = google_auth.default()
  auth_request = google_requests.Request()
  credentials.refresh(auth_request)
  return credentials.token

response = requests.post(
  f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:query",
  headers={
    "Content-Type": "application/json; charset=utf-8",
    "Authorization": f"Bearer {get_identity_token()}",
  },
  data=json.dumps({
    "class_method": "async_delete_session",
    "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},
  }),
)
print(response.content)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:query -d '{"class_method": "async_delete_session", "input": {"user_id": "USER_ID", "session_id": "SESSION_ID"},}'

クエリへのレスポンスをストリーミングする

セッションでエージェントからのレスポンスをストリーミングするには:

Vertex AI SDK for Python

async for event in adk_app.async_stream_query(
    user_id="USER_ID",
    session_id="SESSION_ID",  # Optional
    message="What is the exchange rate from US dollars to SEK today?",
):
  print(event)

Python リクエスト ライブラリ

from google import auth as google_auth
from google.auth.transport import requests as google_requests
import requests

def get_identity_token():
    credentials, _ = google_auth.default()
    auth_request = google_requests.Request()
    credentials.refresh(auth_request)
    return credentials.token

requests.post(
    f"https://{adk_app.api_client.api_endpoint}/v1/{adk_app.resource_name}:streamQuery",
    headers={
        "Content-Type": "application/json",
        "Authorization": f"Bearer {get_identity_token()}",
    },
    data=json.dumps({
        "class_method": "async_stream_query",
        "input": {
            "user_id": "USER_ID",
            "session_id": "SESSION_ID",
            "message": "What is the exchange rate from US dollars to SEK today?",
        },
    }),
    stream=True,
)

REST API

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:streamQuery?alt=sse -d '{
  "class_method": "async_stream_query",
  "input": {
    "user_id": "USER_ID",
    "session_id": "SESSION_ID",
    "message": "What is the exchange rate from US dollars to SEK today?",
  }
}'

注: session_id= 引数は省略可能です。指定しない場合は、新しいセッションが自動的に作成され、そのクエリの実行に使用されます。

Vertex AI SDK for Python を使用している場合は、次のような辞書のシーケンスで会話の続きが返されます。

{'author': 'currency_exchange_agent',
 'content': {'parts': [{'function_call': {'args': {'currency_date': '2025-04-03',
                                                   'currency_from': 'USD',
                                                   'currency_to': 'SEK'},
                                          'id': 'adk-2b9230a6-4b92-4a1b-9a65-b708ff6c68b6',
                                          'name': 'get_exchange_rate'}}],
             'role': 'model'},
 'id': 'bOPHtzji',
 # ...
}
{'author': 'currency_exchange_agent',
 'content': {'parts': [{'function_response': {'id': 'adk-2b9230a6-4b92-4a1b-9a65-b708ff6c68b6',
                                              'name': 'get_exchange_rate',
                                              'response': {'amount': 1.0,
                                                           'base': 'USD',
                                                           'date': '2025-04-03',
                                                           'rates': {'SEK': 9.6607}}}}],
             'role': 'user'},
 'id': '9AoDFmiL',
 # ...
}
{'author': 'currency_exchange_agent',
 'content': {'parts': [{'text': 'The exchange rate from USD to SEK on '
                                '2025-04-03 is 1 USD to 9.6607 SEK.'}],
             'role': 'model'},
 'id': 'hmle7trT',
 # ...
}

思い出を管理する

AdkApp は、エージェント定義に PreloadMemoryTool を含めて、エージェントを Vertex AI Agent Engine にデプロイすると、Vertex AI Agent Engine Memory Bank を使用します。このセクションでは、ADK メモリサービスのデフォルト実装を使用して、エージェントからメモリを生成して取得する方法について説明します。

セッションをメモリーに追加する

セッション内の意味のある情報（将来のセッションで使用できる）のメモリを保持するには、async_add_session_to_memory メソッドを使用します。

Vertex AI SDK for Python

await adk_app.async_add_session_to_memory(session="SESSION_DICT")

ここで、SESSION_DICT は ADK セッション オブジェクトの辞書形式です。

思い出を検索する

エージェントのメモリを検索するには、async_search_memory メソッドを使用します。

Vertex AI SDK for Python

response = await adk_app.async_search_memory(
    user_id="USER_ID",
    query="QUERY",
)
print(response)

ここで

• USER_ID は、関連するメモリのスコープです。
• QUERY は、類似検索を実行するクエリです。

次のステップ

• エージェントを使用する。
• エージェントを評価する。
• デプロイされたエージェントを管理する。
• サポートを利用する。