本頁面由 Cloud Translation API 翻譯而成。

使用 Agent Development Kit 代理程式

事前準備

本教學課程假設您已詳閱並按照下列教學課程的指示操作：

開發 Agent Development Kit 代理程式：開發 agent 做為 AdkApp 的執行個體。
使用者驗證：以使用者身分進行驗證，以便查詢代理程式。
匯入並初始化 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 是已部署代理程式的 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

使用 Python 適用的 Vertex AI 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：用於擷取回憶集錦。

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

Vertex AI SDK for Python

請執行下列程式碼：

adk_app.operation_schemas()

Python requests 程式庫

請執行下列程式碼：

import json

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

REST API

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

管理工作階段

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

建立工作階段

如要為使用者建立工作階段，請按照下列步驟操作：

Vertex AI SDK for Python

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://{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：選擇自己的使用者 ID，最多 128 個字元。例如：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 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://{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 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://{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 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://{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 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://{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?",
  }
}'

如果您使用 Python 適用的 Vertex AI 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',
 # ...
}

管理回憶集錦

AdkApp 會使用 Vertex AI Agent Engine Memory Bank 如果您在代理程式定義中加入 PreloadMemoryTool 並將代理程式部署至 Vertex AI Agent Engine，本節說明如何透過 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 是要執行相似度搜尋的查詢。

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

事前準備

取得代理程式的執行個體

Vertex AI SDK for Python

Python requests 程式庫

REST API

支援的作業

Vertex AI SDK for Python

Python requests 程式庫

REST API

管理工作階段

建立工作階段

Vertex AI SDK for Python

Python requests 程式庫

REST API

列出工作階段

Vertex AI SDK for Python

Python requests 程式庫

REST API

取得工作階段

Vertex AI SDK for Python

Python requests 程式庫

REST API

刪除工作階段

Vertex AI SDK for Python

Python requests 程式庫

REST API

串流查詢的回覆

Vertex AI SDK for Python

Python requests 程式庫

REST API

管理回憶集錦

將對話新增至個人化記憶

Vertex AI SDK for Python

搜尋回憶集錦

Vertex AI SDK for Python

後續步驟

使用 Agent Development Kit 代理程式