Google は AI 技術を使用して、コンテンツをご希望の言語に翻訳しています。AI 翻訳には誤りが含まれる場合があります。

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

始める前に

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

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

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

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

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

Agent Platform SDK

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

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

Agent Platform SDK を使用する場合、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: メモリを取得します。

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

Agent Platform SDK

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

adk_app.operation_schemas()

Python リクエストライブラリ

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

import json

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

REST API

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

セッションを管理する

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

セッションを作成する

ユーザーのセッションを作成するには、AdkApp.async_create_session メソッドを使用します。

Agent Platform SDK

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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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 セッションオブジェクトの辞書表現として作成されます。

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

ユーザーのセッションを一覧表示するには、AdkApp.async_list_sessions メソッドを使用します。

Agent Platform SDK

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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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 セッションオブジェクトの辞書形式が使用されます。

セッションを取得する

特定のセッションを取得するには、AdkApp.async_get_session メソッドを使用します。

Agent Platform SDK

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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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 セッションオブジェクトの辞書表現です。

セッションを削除する

セッションを削除するには、AdkApp.async_delete_session メソッドを使用します。

Agent Platform SDK

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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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"},}'

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

セッションでエージェントからのレスポンスをストリーミングするには、AdkApp.async_stream_query メソッドを使用します。

Agent Platform SDK

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://LOCATION-aiplatform.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID: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?",
  }
}'

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

{'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',
 # ...
}

長時間実行クエリジョブ

完了に時間がかかる（最大 7 日間）クエリは、長時間実行ジョブとして実行できます。これらのジョブは非同期で実行されます。ジョブのステータスを確認し、後で結果を取得できます。

非同期クエリのエージェントをデプロイする

エージェントをデプロイするには、エージェントをデプロイするの一般的な手順に沿って操作します。ソースベースのデプロイの場合は、deploymentSpec.agentFramework フィールドを google-adk に設定します。

独自のコンテナイメージをビルドしてカスタム API エンドポイントを使用する場合は、SDK を使用してエージェントを作成するときに、次の環境変数を追加する必要があります。

"env_vars" = {
    "API_ENDPOINT_PREFIX": "/api/myendpoint"
}

長時間実行クエリジョブを開始する

前提条件として、サービスエージェント service-PROJECT_NUMBER@gcp-sa-aiplatform-re.iam.gserviceaccount.com に出力ファイルのストレージバケットに対する roles/storage.objectCreator ロールを付与する必要があります。

長時間実行クエリジョブを開始するには:

Agent Platform SDK

import vertexai

client = vertexai.Client(
    project="PROJECT_ID",
    location="LOCATION",
)

response = client.agent_engines.run_query_job(
    name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    config={
        "query": '{"input":{"user_id":"USER_ID", "message":"What is the exchange rate from US dollars to SEK today?"}}',
        "output_gcs_uri": "gs://GCS_BUCKET_NAME/OUTPUT_FILE",
    },
)
print(response)

SDK では、output_gcs_uri はディレクトリまたはファイル名にできます。ファイル名の場合、システムはこのファイルを使用してレスポンスを保存します。ディレクトリの場合、システムはレスポンス用のファイルを自動的に生成します。どちらの場合も、入力クエリは出力ファイルと同じファイル名接頭辞で同じディレクトリに保存されます。

REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1beta1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:asyncQuery -d \
'{
  "input_gcs_uri": "gs://GCS_BUCKET_NAME/INPUT_FILE",
  "output_gcs_uri": "gs://GCS_BUCKET_NAME/OUTPUT_FILE"
}'

REST API 呼び出しの場合、input_gcs_uri フィールドはクエリを含むファイルを指している必要があります。ファイルの内容は、QueryReasoningEngineRequest の input フィールド（{ "input": { "user_id": "hello", "message":"$QUERY"} } など）と一致する input フィールドを含む JSON オブジェクトである必要があります。この入力ファイルが出力場所とは異なるバケットにある場合は、入力ファイルが配置されているストレージバケットに対する roles/storage.objectReader ロールをサービスエージェント service-PROJECT_NUMBER@gcp-sa-aiplatform-re.iam.gserviceaccount.com に付与する必要があります。

output_gcs_uri はファイル名である必要があります。

長時間実行クエリジョブのステータスを確認する

長時間実行クエリジョブのステータスを確認して結果を取得するには:

Agent Platform SDK

response = client.agent_engines.check_query_job(
    name="JOB_NAME",
    config={
        "retrieve_result": True,
    },
)
print(response)

長時間実行クエリジョブをキャンセルする

長時間実行クエリジョブをキャンセルするには、長時間実行クエリジョブから返される LRO リソース名が必要です。

Agent Platform SDK

response = client.agent_engines.cancel_query_job(
    name="projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
    config={
        "operation_name": "projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID",
    },
)

REST

curl \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
https://LOCATION-aiplatform.googleapis.com/v1beta1/projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID:cancelAsyncQuery -d \
'{
  "name": "projects/PROJECT_ID/locations/LOCATION/reasoningEngines/RESOURCE_ID",
  "operation_name": "projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID"
}'

キャンセルは非同期で行われます。キャンセルリクエストは受け入れられるとすぐに返されますが、ブロッキングツール呼び出しなどの進行中の作業が完了するまで、ジョブは check_query_job から RUNNING のステータスを報告し続けることがあります。

ジョブがキャンセルされると、オペレーションはエラーコード 1（CANCELLED）とメッセージ Cancelled by user. で完了します。check_query_job は、エラーで完了したすべてのオペレーションをステータス FAILED として報告します。そのため、キャンセルされたジョブは、個別のキャンセルステータスではなく FAILED として報告されます。エラーコードを調べて、キャンセルと実際の障害を区別します。

思い出を管理する

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

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

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

Agent Platform SDK

await adk_app.async_add_session_to_memory(session="SESSION_DICT")

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

思い出を検索する

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

Agent Platform SDK

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

ここで

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

Agent Development Kit エージェントを使用する コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。

始める前に

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

Agent Platform SDK

Python リクエスト ライブラリ

REST API

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

Agent Platform SDK

Python リクエスト ライブラリ

REST API

セッションを管理する

セッションを作成する

Agent Platform SDK

Python リクエスト ライブラリ

REST API

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

Agent Platform SDK

Python リクエスト ライブラリ

REST API

セッションを取得する

Agent Platform SDK

Python リクエスト ライブラリ

REST API

セッションを削除する

Agent Platform SDK

Python リクエスト ライブラリ

REST API

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

Agent Platform SDK

Python リクエスト ライブラリ

REST API

長時間実行クエリジョブ

非同期クエリのエージェントをデプロイする

長時間実行クエリジョブを開始する

Agent Platform SDK

REST

長時間実行クエリジョブのステータスを確認する

Agent Platform SDK

長時間実行クエリジョブをキャンセルする

Agent Platform SDK

REST

思い出を管理する

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

Agent Platform SDK

思い出を検索する

Agent Platform SDK

次のステップ

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

Python リクエストライブラリ

Python リクエストライブラリ

Python リクエストライブラリ

Python リクエストライブラリ

Python リクエストライブラリ

Python リクエストライブラリ

Python リクエストライブラリ