Google 會運用 AI 技術將內容翻譯成你偏好的語言，但可能會出錯。

使用 ADK 代理

事前準備

本教學課程假設您已閱讀並按照下列操作說明進行：

建立 Agent Development Kit (ADK) 代理：建立 agent 做為 AdkApp 的執行個體。
使用者驗證：以使用者身分驗證，以便查詢代理。
匯入並初始化 SDK，初始化用戶端，以便取得已部署的執行個體 (如有需要)。

取得代理執行個體

如要查詢 AdkApp，您必須先建立新執行個體或取得現有執行個體。

如要取得與特定資源 ID 對應的 AdkApp，請按照下列步驟操作：

Agent Platform SDK

請執行下列程式碼：

import agentplatform

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

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

print(adk_app)

其中

PROJECT_ID 是您建立和部署代理的 Google Cloud 專案 ID，
LOCATION 是支援的地區之一，且
RESOURCE_ID 是已部署代理程式的 ID，為 reasoningEngine 資源。

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 物件會對應至 AgentRuntime 類別，其中包含下列內容：

adk_app.api_resource，其中包含已部署代理程式的相關資訊。您也可以呼叫 adk_app.operation_schemas()，傳回 adk_app 支援的作業清單。詳情請參閱「支援的操作」。
adk_app.api_client，可進行同步服務互動
adk_app.async_api_client：可進行非同步服務互動

本節其餘部分假設您有一個名為 adk_app 的 AgentRuntime 執行個體。

支援的操作

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

請執行下列程式碼：

import json

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

REST API

在 curl 要求的回應中，以 spec.class_methods 表示。

管理工作階段

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

建立工作階段

如要為使用者建立工作階段，請使用 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"
}

啟動長時間執行的查詢工作

您必須先將輸出檔案儲存空間 bucket 的 roles/storage.objectCreator 角色授予服務代理 service-PROJECT_NUMBER@gcp-sa-aiplatform-re.iam.gserviceaccount.com，才能執行這項操作。

如要啟動長時間執行的查詢工作，請按照下列步驟操作：

Agent Platform SDK

import agentplatform

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

response = client.runtimes.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 物件，且 input 欄位與 QueryReasoningEngineRequest 的 input 欄位相符 (例如 { "input": { "user_id": "hello", "message":"$QUERY"} })。如果這個輸入檔案與輸出位置位於不同的 bucket，您也必須將輸入檔案所在儲存空間 bucket 的 roles/storage.objectReader 角色授予服務代理 service-PROJECT_NUMBER@gcp-sa-aiplatform-re.iam.gserviceaccount.com。

output_gcs_uri 必須是檔案名稱。

查看長時間執行的查詢工作狀態

如要查看長時間執行的查詢工作狀態並擷取結果，請按照下列步驟操作：

Agent Platform SDK

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

取消長時間執行的查詢工作

如要取消長時間執行的查詢工作，您必須取得長時間執行查詢工作傳回的 LRO 資源名稱。

Agent Platform SDK

response = client.runtimes.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 是要執行相似度搜尋的查詢。

使用 ADK 代理 透過集合功能整理內容 你可以依據偏好儲存及分類內容。

事前準備

取得代理執行個體

Agent Platform SDK

Python requests 程式庫

REST API

支援的操作

Agent Platform SDK

Python requests 程式庫

REST API

管理工作階段

建立工作階段

Agent Platform SDK

Python requests 程式庫

REST API

列出工作階段

Agent Platform SDK

Python requests 程式庫

REST API

取得工作階段

Agent Platform SDK

Python requests 程式庫

REST API

刪除工作階段

Agent Platform SDK

Python requests 程式庫

REST API

串流查詢回覆

Agent Platform SDK

Python requests 程式庫

REST API

長時間執行的查詢工作

部署代理來執行非同步查詢

啟動長時間執行的查詢工作

Agent Platform SDK

REST

查看長時間執行的查詢工作狀態

Agent Platform SDK

取消長時間執行的查詢工作

Agent Platform SDK

REST

管理回憶集錦

將工作階段新增至記憶庫

Agent Platform SDK

搜尋回憶集錦

Agent Platform SDK

後續步驟

使用 ADK 代理