Google 會運用 AI 技術將內容翻譯成你偏好的語言，但可能會出錯。

使用 ADK 管理工作階段

本頁面說明如何將 Agent Development Kit (ADK) 代理連結至 Agent Platform 工作階段，並在本機和正式環境中使用代管工作階段。

事前準備

這些操作說明使用下列基本專案檔案結構，定義 ADK 代理及其支援的執行工具和部署程式碼：

my_agent/
    agent.py      # main agent code
    runner.py     # code for interacting with the agent
    deploy.py     # code for deploying the agent to Google Cloud

請按照「設定環境」中的「取得必要角色」和「驗證」步驟，確認環境已設定完成。

設定環境變數

如要使用 ADK，請設定環境變數：

import os

os.environ["GOOGLE_GENAI_USE_ENTERPRISE"] = "TRUE"
os.environ["GOOGLE_CLOUD_PROJECT"] = "PROJECT_ID"
os.environ["GOOGLE_CLOUD_LOCATION"] = "LOCATION"

更改下列內容：

PROJECT_ID：專案 ID。
LOCATION：所在區域。請參閱支援工作階段的區域。

建立 Agent Runtime 執行個體

如要存取 Agent Platform 工作階段，請先使用 Agent Runtime 執行個體。您不需要部署任何程式碼，就能開始使用工作階段。如果您曾使用過 Agent Runtime，只需幾秒鐘就能建立 Agent Runtime 執行個體，不必部署程式碼。如果您是第一次使用 Agent Runtime，可能需要較長時間。

Google Cloud 專案

import agentplatform

client = agentplatform.Client(
  project="PROJECT_ID",
  location="LOCATION"
)

# If you don't have an Agent Runtime instance already, create an instance.
remote_agent = client.runtimes.create(
    config={
      "display_name": "My Session Store",
    }
)

# Print the runtime ID, you will need it in the later steps to initialize
# the ADK `VertexAiSessionService`.
print(remote_agent.api_resource.name.split("/")[-1])

更改下列內容：

PROJECT_ID：專案 ID。
LOCATION：所在區域。請參閱支援工作階段的區域。

開發 ADK 代理

如要建立 ADK 代理，請按照 ADK 中的操作說明，或使用下列程式碼建立代理，以固定問候語向使用者打招呼。將這段程式碼儲存至名為 agent.py 的檔案。

# file: my_agent/agent.py
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，以便連結至 Agent Platform 工作階段。將這段程式碼儲存至名為 runner.py 的檔案。

Google Cloud 專案

# file: my_agent/runner.py
import agent # Import from your agent.py
from google.adk import Runner
from google.adk.sessions import VertexAiSessionService
from google.genai import types

app_name="APP_NAME"
user_id="USER_ID"

# Create the ADK runner with VertexAiSessionService
session_service = VertexAiSessionService(
      project="PROJECT_ID",
      location="LOCATION",
      agent_engine_id="AGENT_ENGINE_ID"
)
runner = Runner(
    agent=agent.root_agent,
    app_name=app_name,
    session_service=session_service)

# Helper method to send query to the runner
async def call_agent(query, session_id, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  async for event in runner.run_async(
      user_id=user_id, session_id=session_id, new_message=content):
      if event.is_final_response():
          final_response = event.content.parts[0].text
          print("Agent Response: ", final_response)

更改下列內容：

APP_NAME：代理應用程式的名稱。
USER_ID：選擇自己的使用者 ID，字元限制為 128 個。例如：user-123。
AGENT_ENGINE_ID：Agent Runtime 執行個體的資源 ID。
如果是已部署的代理，資源 ID 會列為 GOOGLE_CLOUD_AGENT_ENGINE_ID 環境變數
如果是本機代理，您可以使用 remote_agent.api_resource.name.split("/")[-1] 擷取資源 ID。

與代理互動

定義代理並設定 Agent Platform 工作階段後，即可與代理互動，確認工作階段記錄和狀態是否保留。

ADK UI

如要使用 ADK 使用者介面測試代理，並連線至 Agent Platform 工作階段，請在終端機執行 adk web 指令。

雖然 adk web 通常會從 .env 檔案讀取環境變數，但使用 --session_service_uri 旗標時，系統會忽略這個檔案。您必須先在終端機工作階段中設定 GOOGLE_CLOUD_PROJECT 和 GOOGLE_CLOUD_LOCATION 環境變數，才能執行指令。

export GOOGLE_CLOUD_PROJECT="PROJECT_ID"
export GOOGLE_CLOUD_LOCATION="LOCATION"
adk web --session_service_uri=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)

ADK UI

Python

使用 ADK Python 程式碼管理工作階段和狀態。在 runner.py 檔案結尾加入下列程式碼，與代理互動。

為求簡潔，下列程式碼片段包含頂層 await 呼叫。如要以 Python 指令碼的形式執行這段程式碼，請將程式碼片段放入 async 函式，並使用 asyncio.run() 執行，如以下範例所示：

import asyncio

async def main():
  # Place one or more snippets here.
  # For example:
  session = await session_service.create_session(
         app_name=app_name,
         user_id=user_id)

  await call_agent("Hello!", session.id, user_id)

asyncio.run(main())

建立工作階段並查詢代理

使用下列程式碼建立工作階段，並將查詢傳送至代理。指定自訂工作階段 ID，即可使用自己的命名配置來管理工作階段，或快速從其他服務轉換。如未指定工作階段 ID，Agent Platform 會自動產生一個。

# file: my_agent/runner.py
# Create a session
session = await session_service.create_session(
       app_name=app_name,
       user_id=user_id,
       session_id=session_id)

await call_agent("Hello!", session.id, user_id)
# Agent response: "Hello, world"

await call_agent("Thanks!", session.id, user_id)
# Agent response: "Goodbye, world"

如要使用自訂工作階段 ID，請考量下列限制，以免與系統產生的 ID 衝突：

如果第一個字元是字母，ID 長度最多可達 63 個字元。有效字元包括小寫字母、數字和連字號 ([a-z0-9-])。最後一個字元必須是字母或數字。
如果第一個字元是數字，ID 最多可有 9 個字元。有效字元為數字 ([0-9])，開頭不得為零。

建立工作階段並傳遞給執行工具後，ADK 會使用工作階段儲存目前互動的事件。您也可以提供工作階段 ID，繼續執行先前的工作階段。

設定工作階段存留時間 (TTL)

所有工作階段都必須設定到期時間。您可以在建立或更新工作階段時，定義這個到期時間。工作階段及其子事件會在到期後自動刪除。您可以直接設定到期時間 (expire_time)，也可以設定存留時間 (ttl)，以秒為單位。如未指定，系統會套用預設存留時間 (365 天)。

存留時間

如果您設定存留時間，伺服器會將新建立工作階段的到期時間計算為 create_time + ttl，或將更新工作階段的到期時間計算為 update_time + ttl。

  session = await session_service.create_session(
        app_name=app_name,
        user_id=user_id,
        # Session will be deleted 10 days after creation time.
        ttl=f"{24 * 60 * 60 * 10}s"
  )
  ```

到期時間

  import datetime

  expire_time = datetime.datetime.now(
        tz=datetime.timezone.utc) + datetime.timedelta(seconds=24 * 60 * 60 * 10)

  session = await session_service.create_session(
        app_name=app_name,
        user_id=user_id,
        # Session will be deleted at the provided time (10 days after current time).
        expire_time=expire_time.isoformat()
  )

列出現有工作階段

列出與指定使用者 ID 相關聯的所有現有工作階段。

# List sessions
sessions = await session_service.list_sessions(app_name=app_name,user_id=user_id)
print(sessions)
# 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 將新事件附加至工作階段：

# file: my_agent/runner.py
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

活動的自訂欄位

ADK 支援彈性的事件結構定義，因此您可以在工作階段事件中加入任意資料。這項功能有助於與其他代理架構互通，或儲存自訂事件資料。

當您附加含有自訂欄位的事件時，ADK 會將事件資料序列化至儲存的工作階段事件 raw_event 欄位。

# Create an event with custom fields
custom_event = Event(
  invocation_id="invocation_id",
  author="user",
  timestamp=time.time(),
  custom_field="custom_value",
  another_field={"nested_key": "nested_value"}
)

# Append the event
await session_service.append_event(session, custom_event)

擷取工作階段時，擷取的事件會保留這些自訂欄位。

刪除工作階段

刪除與使用者 ID 相關聯的特定工作階段：

await session_service.delete_session(app_name=app_name, user_id=user_id, session_id=session.id)

將代理部署至 Agent Platform

在本機測試代理後，您可以更新 Agent Platform 執行個體，並加入下列參數，將代理部署至正式環境：

Google Cloud 專案

client.runtimes.update(
    resource_name=remote_agent.api_resource.name,
    agent=AGENT,
    config={
      "display_name": DISPLAY_NAME,      # Optional.
      "requirements": REQUIREMENTS,      # Optional.
      "staging_bucket": STAGING_BUCKET,  # Required.
    },
)

更改下列內容：

AGENT：實作 query / stream_query 方法的應用程式 (例如 ADK 代理的 AdkApp)。
DISPLAY_NAME：容易使用的代理名稱。
REQUIREMENTS：代理所需的 pip 套件清單。例如：["google-cloud-storage", "google-cloud-agentplatform[runtimes,adk]"]。
STAGING_BUCKET：Cloud Storage bucket，前置字串為 gs://。

清除所用資源

如要清除這項專案中使用的所有資源，您可以刪除 Agent Runtime 執行個體及其子資源：

remote_agent.delete(force=True)

後續步驟

使用 API 呼叫管理工作階段。

使用 ADK 管理工作階段 透過集合功能整理內容 你可以依據偏好儲存及分類內容。

事前準備

設定環境變數

建立 Agent Runtime 執行個體

Google Cloud 專案

開發 ADK 代理

設定 ADK 執行工具

Google Cloud 專案

與代理互動

ADK UI

Python

建立工作階段並查詢代理

設定工作階段存留時間 (TTL)

存留時間

到期時間

列出現有工作階段

管理工作階段狀態

活動的自訂欄位

刪除工作階段

將代理部署至 Agent Platform

Google Cloud 專案

清除所用資源

後續步驟

使用 ADK 管理工作階段