Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

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

准备工作

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

创建智能体开发套件智能体：将 agent 作为 AdkApp 的实例进行创建。
用户身份验证，以用户身份进行身份验证，以便查询智能体。
导入并初始化 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 是所部署智能体的 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

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

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

管理会话

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

创建会话

如需为用户创建会话，请使用 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：自行选择用户 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 请求库

运行以下代码：

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. 授予输出文件存储桶的 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"} }）。如果此输入文件与输出位置位于不同的存储桶中，您还必须向服务代理 service-PROJECT_NUMBER@gcp-sa-aiplatform-re. 授予对输入文件所在存储桶的 roles/storage.objectReader 角色。

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",
    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"
}'

管理回忆

如果您在智能体定义中添加了 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 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

后续步骤

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