本頁說明如何將 Agent Development Kit (ADK) 代理程式連結至 Vertex AI Agent Engine 工作階段，以及如何在本地和正式環境中使用受管理的工作階段。
事前準備
請按照「設定環境」一文中的「取得必要角色」和「驗證」步驟，確認環境已設定完成。
建立 Vertex AI Agent Engine 執行個體
如要存取 Vertex AI Agent Engine Sessions，請先建立 Vertex AI Agent Engine 執行個體。您不需要部署任何程式碼，即可開始使用 Sessions。無須部署程式碼，建立 Vertex AI Agent Engine 執行個體只需幾秒鐘。
import vertexai
from vertexai import agent_engines
# Create an agent engine instance
agent_engine = agent_engines.create()
開發 ADK 代理
如要建立 ADK 代理程式，請按照「Agent Development Kit」中的操作說明進行，或使用下列程式碼建立代理程式，以固定問候語向使用者打招呼：
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，與 Vertex AI Agent Engine Sessions 建立連線。
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：使用者的專屬 ID，不得為空值，長度上限為 128 個字元。
與虛擬服務專員互動
定義代理程式並設定 Vertex AI Agent Engine 工作階段後，您可以與代理程式互動，確認工作階段記錄和狀態是否會持續存在。
ADK UI
使用 ADK 使用者介面測試代理程式，並使用 session_db_url 指令列選項連線至 Vertex AI Agent Engine 工作階段：
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"
建立工作階段並傳遞至執行器後，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
如要在執行器外部更新工作階段狀態，請使用 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)