本页面介绍了如何将 Agent Development Kit (ADK) 代理与 Vertex AI Agent Engine 会话相关联，以及如何在本地环境和生产环境中使用受管理的会话。
准备工作
请确保按照设置环境中的获取所需角色和身份验证步骤设置您的环境。
创建 Vertex AI Agent Engine 实例
如需访问 Vertex AI Agent Engine 会话，您首先需要创建 Vertex AI Agent Engine 实例。您无需部署任何代码即可开始使用会话。无需部署代码，创建 Vertex AI Agent Engine 实例只需几秒钟。
import vertexai
from vertexai import agent_engines
# Create an agent engine instance
agent_engine = agent_engines.create()
开发 ADK 智能体
如需创建 ADK 代理，请按照代理开发套件中的说明操作，或使用以下代码创建可使用固定问候语向用户问好的代理：
from google import adk
def greetings(query: str):
"""Tool to greet user."""
if 'hello' in query.lower():
return {"greeting": "Hello, world"}
else:
return {"greeting": "Goodbye, world"}
# Define an ADK agent
root_agent = adk.Agent(
model="gemini-2.0-flash",
name='my_agent',
instruction="You are an Agent that greet users, always use greetings tool to respond.",
tools=[greetings]
)
设置 ADK 运行器
ADK 运行时负责对智能体、工具及回调的执行进行编排，并编排对会话的读取和写入调用。使用 VertexAiSessionService 初始化 Runner，该 Runner 会与 Vertex AI Agent Engine 会话建立连接。
from google.adk.sessions import VertexAiSessionService
app_name="AGENT_ENGINE_ID"
user_id="USER_ID"
# Create the ADK runner with VertexAiSessionService
session_service = VertexAiSessionService(
"PROJECT_ID", "LOCATION")
runner = adk.Runner(
agent=root_agent,
app_name=app_name,
session_service=session_service)
# Helper method to send query to the runner
def call_agent(query, session_id, user_id):
content = types.Content(role='user', parts=[types.Part(text=query)])
events = runner.run(
user_id=user_id, session_id=session_id, new_message=content)
for event in events:
if event.is_final_response():
final_response = event.content.parts[0].text
print("Agent Response: ", final_response)
替换以下内容：
PROJECT_ID：您的项目 ID。
LOCATION：您的区域。
AGENT_ENGINE_ID：Vertex AI Agent Engine 实例的资源 ID。
对于已部署的代理，资源 ID 会列为
GOOGLE_CLOUD_AGENT_ENGINE_ID环境变量对于本地代理，您可以使用
agent_engine.name.split("/")[-1]检索资源 ID。
USER_ID：用户的非空唯一标识符，长度上限为 128 个字符。
与您的代理互动
定义代理并设置 Vertex AI Agent Engine 会话后，您可以与代理互动，以检查会话历史记录和状态是否会保留。
ADK 界面
使用 ADK 界面测试智能体，并通过 session_db_url 命令行选项连接到 Vertex AI 智能体引擎会话：
agent_engine_id="AGENT_ENGINE_ID"
adk web --session_db_url=agentengine://${agent_engine_id}
# Sample output
+-----------------------------------------------------------------------------+
| ADK Web Server started |
| |
| For local testing, access at http://localhost:8000. |
+-----------------------------------------------------------------------------+
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
Python
使用 ADK Python 代码管理会话和状态。
创建会话并查询智能体
使用以下代码创建会话并向代理发送查询：
# Create a session
session = await session_service.create_session(
app_name=app_name,
user_id=user_id)
call_agent("Hello!", session.id, user_id)
# Agent response: "Hello, world"
call_agent("Thanks!", session.id, user_id)
# Agent response: "Goodbye, world"
在创建会话并将其传递给 runner 后，ADK 会使用该会话来存储当前互动中的事件。您还可以通过提供相应会话的 ID 来恢复之前的会话。
列出现有会话
列出与指定用户 ID 相关联的所有现有会话。
# List sessions
await session_service.list_sessions(app_name=app_name,user_id=user_id)
# ListSessionsResponse(session_ids=['1122334455', '9988776655'])
管理会话状态
状态包含代理在对话中需要的信息。您可以在创建会话时以字典形式提供初始状态：
# Create a session with state
session = await session_service.create_session(
app_name=app_name,
user_id=user_id,
state={'key': 'value'})
print(session.state['key'])
# value
如需在 runner 之外更新会话状态，请使用 state_delta 将新事件附加到会话：
from google.adk.events import Event, EventActions
import time
# Define state changes
state_changes = {'key': 'new_value'}
# Create event with actions
actions_with_update = EventActions(state_delta=state_changes)
system_event = Event(
invocation_id="invocation_id",
author="system", # Or 'agent', 'tool' etc.
actions=actions_with_update,
timestamp=time.time()
)
# Append the event
await session_service.append_event(session, system_event)
# Check updated state
updated_session = await session_service.get_session(
app_name=app_name,
user_id=user_id,
session_id=session.id)
# State is updated to new value
print(updated_session.state['key'])
# new_value
删除会话
删除与用户 ID 关联的特定会话：
await session_service.delete_session(app_name=app_name, user_id=user_id, session_id=session.id)
将代理部署到 Vertex AI Agent Engine
在本地测试代理后，您可以通过使用以下参数更新 Vertex AI Agent Engine 实例，将代理部署到生产环境：
agent_engines.update(resource_name=agent_engine.name, agent_engine=AGENT, requirements=REQUIREMENTS)
替换以下内容：
- AGENT：实现
query / stream_query方法的应用（例如，对于 ADK 代理，为AdkApp）。如需了解详情，请参阅部署注意事项。
清理
如需清理此项目中使用的所有资源，您可以删除 Vertex AI Agent Engine 实例及其子资源：
agent_engine.delete(force=True)