部署到 Vertex AI Agent Engine¶
Agent Engine 是一项完全托管的 Google Cloud 服务，使开发者能够在生产环境中部署、管理 和扩展 AI 智能体。Agent Engine 处理基础设施以便在生产环境中扩展智能体， 让你可以专注于创建智能且有影响力的应用程序。
from vertexai import agent_engines
remote_app = agent_engines.create(
agent_engine=root_agent,
requirements=[
"google-cloud-aiplatform[adk,agent_engines]",
]
)
安装 Vertex AI SDK¶
Agent Engine 是 Vertex AI SDK for Python 的一部分。有关更多信息，你可以查看 Agent Engine 快速入门文档。
安装 Vertex AI SDK¶
Info
Agent Engine 仅支持 Python 版本 >=3.9 且 <=3.12。
初始化¶
import vertexai
PROJECT_ID = "your-project-id"
LOCATION = "us-central1"
STAGING_BUCKET = "gs://your-google-cloud-storage-bucket"
vertexai.init(
project=PROJECT_ID,
location=LOCATION,
staging_bucket=STAGING_BUCKET,
)
对于 LOCATION，你可以查看 Agent Engine 支持的区域列表。
创建你的智能体¶
你可以使用下面的示例智能体，它有两个工具（获取天气或查询指定城市的时间）：
import datetime
from zoneinfo import ZoneInfo
from google.adk.agents import Agent
def get_weather(city: str) -> dict:
"""Retrieves the current weather report for a specified city.
Args:
city (str): The name of the city for which to retrieve the weather report.
Returns:
dict: status and result or error msg.
"""
if city.lower() == "new york":
return {
"status": "success",
"report": (
"The weather in New York is sunny with a temperature of 25 degrees"
" Celsius (41 degrees Fahrenheit)."
),
}
else:
return {
"status": "error",
"error_message": f"Weather information for '{city}' is not available.",
}
def get_current_time(city: str) -> dict:
"""Returns the current time in a specified city.
Args:
city (str): The name of the city for which to retrieve the current time.
Returns:
dict: status and result or error msg.
"""
if city.lower() == "new york":
tz_identifier = "America/New_York"
else:
return {
"status": "error",
"error_message": (
f"Sorry, I don't have timezone information for {city}."
),
}
tz = ZoneInfo(tz_identifier)
now = datetime.datetime.now(tz)
report = (
f'The current time in {city} is {now.strftime("%Y-%m-%d %H:%M:%S %Z%z")}'
)
return {"status": "success", "report": report}
root_agent = Agent(
name="weather_time_agent",
model="gemini-2.0-flash",
description=(
"Agent to answer questions about the time and weather in a city."
),
instruction=(
"You are a helpful agent who can answer user questions about the time and weather in a city."
),
tools=[get_weather, get_current_time],
)
为 Agent Engine 准备你的智能体¶
使用 reasoning_engines.AdkApp() 包装你的智能体，使其可以部署到 Agent Engine
from vertexai.preview import reasoning_engines
app = reasoning_engines.AdkApp(
agent=root_agent,
enable_tracing=True,
)
在本地尝试你的智能体¶
你可以在部署到 Agent Engine 之前在本地尝试。
创建会话（本地）¶
create_session（本地）的预期输出：
Session(id='c6a33dae-26ef-410c-9135-b434a528291f', app_name='default-app-name', user_id='u_123', state={}, events=[], last_update_time=1743440392.8689594)
列出会话（本地）¶
list_sessions（本地）的预期输出：
获取特定会话（本地）¶
get_session（本地）的预期输出：
Session(id='c6a33dae-26ef-410c-9135-b434a528291f', app_name='default-app-name', user_id='u_123', state={}, events=[], last_update_time=1743681991.95696)
向你的智能体发送查询（本地）¶
for event in app.stream_query(
user_id="u_123",
session_id=session.id,
message="whats the weather in new york",
):
print(event)
stream_query（本地）的预期输出：
{'parts': [{'function_call': {'id': 'af-a33fedb0-29e6-4d0c-9eb3-00c402969395', 'args': {'city': 'new york'}, 'name': 'get_weather'}}], 'role': 'model'}
{'parts': [{'function_response': {'id': 'af-a33fedb0-29e6-4d0c-9eb3-00c402969395', 'name': 'get_weather', 'response': {'status': 'success', 'report': 'The weather in New York is sunny with a temperature of 25 degrees Celsius (41 degrees Fahrenheit).'}}}], 'role': 'user'}
{'parts': [{'text': 'The weather in New York is sunny with a temperature of 25 degrees Celsius (41 degrees Fahrenheit).'}], 'role': 'model'}
将你的智能体部署到 Agent Engine¶
from vertexai import agent_engines
remote_app = agent_engines.create(
agent_engine=root_agent,
requirements=[
"google-cloud-aiplatform[adk,agent_engines]"
]
)
这一步可能需要几分钟才能完成。
授予已部署智能体权限¶
在继续查询 Agent Engine 上的智能体之前，你的已部署智能体必须首先被授予额外的权限，然后才能使用托管会话。托管会话是 Agent Engine 的内置组件，使智能体能够跟踪对话状态。如果不授予部署的智能体以下权限，在查询已部署的智能体时可能会看到错误。
你可以按照 设置服务智能体权限 中的说明，通过 IAM 管理页面 授予以下权限：
- Vertex AI 用户（
roles/aiplatform.user）权限给你的service-PROJECT_NUMBER@gcp-sa-aiplatform-re.iam.gserviceaccount.com服务账号
在 Agent Engine 上尝试你的智能体¶
创建会话（远程）¶
create_session（远程）的预期输出：
{'events': [],
'user_id': 'u_456',
'state': {},
'id': '7543472750996750336',
'app_name': '7917477678498709504',
'last_update_time': 1743683353.030133}
id 是会话 ID，而 app_name 是部署在 Agent Engine 上的智能体的资源 ID。
列出会话（远程）¶
获取特定会话（远程）¶
Note
在本地使用智能体时，会话 ID 存储在 session.id 中，而在 Agent Engine 上远程使用智能体时，会话 ID 存储在 remote_session["id"] 中。
向你的智能体发送查询（远程）¶
for event in remote_app.stream_query(
user_id="u_456",
session_id=remote_session["id"],
message="whats the weather in new york",
):
print(event)
stream_query（远程）的预期输出：
{'parts': [{'function_call': {'id': 'af-f1906423-a531-4ecf-a1ef-723b05e85321', 'args': {'city': 'new york'}, 'name': 'get_weather'}}], 'role': 'model'}
{'parts': [{'function_response': {'id': 'af-f1906423-a531-4ecf-a1ef-723b05e85321', 'name': 'get_weather', 'response': {'status': 'success', 'report': 'The weather in New York is sunny with a temperature of 25 degrees Celsius (41 degrees Fahrenheit).'}}}], 'role': 'user'}
{'parts': [{'text': 'The weather in New York is sunny with a temperature of 25 degrees Celsius (41 degrees Fahrenheit).'}], 'role': 'model'}
清理¶
完成后，清理云资源是一个良好的实践。 你可以删除已部署的 Agent Engine 实例，以避免在 Google Cloud 账户上产生任何意外 费用。
force=True 还将删除从已部署智能体生成的任何子资源，例如会话。