部署至 Vertex AI Agent Engine
Agent Engine 是 Google Cloud 提供的全托管服务，可帮助开发者在生产环境中部署、管理和扩展 AI 代理。该服务负责处理生产环境的基础设施扩展，让开发者能专注于构建智能高效的应用系统。
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 Python SDK 的组成部分。更多信息请参阅 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() 封装代理以实现可部署性：
from vertexai.preview import reasoning_engines
app = reasoning_engines.AdkApp(
agent=root_agent,
enable_tracing=True,
)
本地测试代理
部署前可先在本地验证功能。
创建会话（本地）
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，远程使用时则存储在 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 实例可避免产生意外费用。
force=True 将同时删除该代理衍生的所有子资源（如会话记录）。