使用智能体开发套件智能体

准备工作

本教程假定您已阅读并遵循以下说明：

• 开发智能体开发套件智能体：将 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 请求库

运行以下代码：

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 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

以 spec.class_methods 形式表示，来自对 curl 请求的响应。

管理会话

在您将智能体部署到 Vertex AI Agent Engine 后，AdkApp 使用基于云的托管式会话。本部分介绍了如何使用托管式会话。

注意：这些说明假定您在开发智能体时未使用 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：自行选择用户 ID，字符数限制为 128。 例如 user-123。

会话以 ADK 会话对象的字典表示形式创建。

列出会话

如需列出用户的会话，请执行以下操作：

Python 版 Vertex AI 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://{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：

Python 版 Vertex AI 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://{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：

Python 版 Vertex AI 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://{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"},}'

对查询的响应进行流式传输

如需在会话中从代理流式传输响应，请执行以下操作：

Python 版 Vertex AI 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://{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',
 # ...
}

管理回忆

如果您在智能体定义中添加了 PreloadMemoryTool 并将智能体部署到 Vertex AI Agent Engine，AdkApp 将使用 Vertex AI Agent Engine 记忆库。本部分介绍如何通过 ADK 记忆服务的默认实现，从智能体生成和检索记忆。

将会议添加到记忆

如需保留会话中有意义的信息（可在未来的会话中使用），请使用 async_add_session_to_memory 方法：

Python 版 Vertex AI SDK

await adk_app.async_add_session_to_memory(session="SESSION_DICT")

其中，SESSION_DICT 是 ADK 会话对象的字典形式。

搜索回忆

如需搜索代理的记忆，您可以使用 async_search_memory 方法：

Python 版 Vertex AI SDK

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

其中

• USER_ID 是相关记忆的范围。
• QUERY 是要执行相似度搜索的查询。

后续步骤

• 使用代理。
• 评估智能体。
• 管理已部署的智能体。
• 获取支持。