使用 Agent Development Kit 代理

事前準備

本教學課程假設您已詳閱並按照下列文章中的操作說明進行：

取得代理程式的執行個體

如要查詢 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)

其中

Python requests 程式庫

請執行下列程式碼：

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 的 AgentEngine 執行個體。

支援的作業

AdkApp 支援下列作業：

如要列出所有支援的作業，請執行下列指令：

Agent Platform SDK

請執行下列程式碼：

adk_app.operation_schemas()

Python requests 程式庫

請執行下列程式碼：

import json

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

REST API

以 spec.class_methods 表示，來自 curl 要求的相關回應。

管理工作階段

AdkApp 會在您將代理部署至 Agent Platform 後，使用雲端代管工作階段。本節說明如何使用受管理的工作階段。

建立課程

如要為使用者建立工作階段，請使用 AdkApp.async_create_session 方法：

Agent Platform SDK 

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

print(session)

Python requests 程式庫

請執行下列程式碼：

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：選擇自己的使用者 ID，最多 128 個字元。 例如：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 requests 程式庫

請執行下列程式碼：

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 requests 程式庫

請執行下列程式碼：

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 requests 程式庫

請執行下列程式碼：

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 requests 程式庫

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

長時間執行的查詢工作

如果查詢需要較長時間才能完成 (最多七天)，您可以將查詢做為長期執行的工作執行。這些工作會以非同步方式執行。稍後可以查看工作狀態並擷取結果。

部署代理以進行非同步查詢

如要部署代理程式，請按照「部署代理程式」中的一般操作說明進行。 如果是以來源為基礎的部署作業，請將 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 欄位必須指向含有查詢內容的檔案。檔案內容必須是 JSON 物件，且具有與 QueryReasoningEngineRequest 的 input 欄位相符的 input 欄位 (例如 { "input": { "user_id": "hello", "message":"$QUERY"} })。如果這個輸入檔案與輸出位置位於不同的值區，您也必須將 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 就會使用 Memory Bank。本節說明如何透過 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 是要執行相似度搜尋的查詢。

後續步驟