Agent Platform SDK for Python 中的 AdkApp 類別，即可建立會傳回匯率並管理有狀態互動的代理。
本文說明如何開發 ADK 代理，包括定義模型、新增工具，以及管理工作階段和記憶。
如要進一步瞭解如何管理已部署的代理程式，請參閱「管理已部署的代理程式」。
事前準備
請按照「設定環境」中的步驟操作，確保環境已設定完成。
定義及設定模型
指定要使用的模型：
model = "gemini-2.0-flash"
選用：設定模型的安全設定。如要進一步瞭解 Gemini 的安全設定選項，請參閱「設定安全屬性」。以下是安全性設定的範例：
from google.genai import types
safety_settings = [
types.SafetySetting(
category=types.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
threshold=types.HarmBlockThreshold.OFF,
),
]
選用：指定內容生成參數：
from google.genai import types
generate_content_config = types.GenerateContentConfig(
safety_settings=safety_settings,
temperature=0.28,
max_output_tokens=1000,
top_p=0.95,
)
使用模型設定建立 AdkApp：
from google.adk.agents import Agent
from vertexai.agent_engines import AdkApp
agent = Agent(
model=model, # Required.
name='currency_exchange_agent', # Required.
generate_content_config=generate_content_config, # Optional.
)
app = AdkApp(agent=agent)
如果您是在互動式環境 (例如終端機或 Colab 筆記本) 中執行，可以使用 AdkApp.async_stream_query 方法執行查詢，做為中繼測試步驟：
async for event in app.async_stream_query(
user_id="USER_ID", # Required
message="What is the exchange rate from US dollars to Swedish currency?",
):
print(event)
- USER_ID：選擇自己的使用者 ID，字元限制為 128 個。
例如：
user-123。
回應是 Python 字典，類似下列範例：
{'actions': {'artifact_delta': {},
'requested_auth_configs': {},
'state_delta': {}},
'author': 'currency_exchange_agent',
'content': {'parts': [{'text': 'To provide you with the most accurate '
'exchange rate, I need to know the specific '
'currencies you\'re asking about. "Swedish '
'currency" could refer to:\n'
'\n'
'* **Swedish Krona (SEK):** This is the '
'official currency of Sweden.\n'
'\n'
"Please confirm if you're interested in the "
'exchange rate between USD and SEK. Once you '
'confirm, I can fetch the latest exchange rate '
'for you.\n'}],
'role': 'model'},
'id': 'LYg7wg8G',
'invocation_id': 'e-113ca547-0f19-4d50-9dde-f76cbc001dce',
'timestamp': 1744166956.925927}
選用：定義及使用工具
定義模型後，請定義模型用於推論的工具。
定義函式時，請務必加入註解，完整清楚地說明函式的參數、函式的作用，以及函式傳回的內容。模型會根據這項資訊決定要使用的函式。您也必須在本機測試函式，確認函式能正常運作。
使用下列程式碼定義會傳回匯率的函式：
def get_exchange_rate(
currency_from: str = "USD",
currency_to: str = "EUR",
currency_date: str = "latest",
):
"""Retrieves the exchange rate between two currencies on a specified date.
Uses the Frankfurter API (https://api.frankfurter.app/) to obtain
exchange rate data.
Args:
currency_from: The base currency (3-letter currency code).
Defaults to "USD" (US Dollar).
currency_to: The target currency (3-letter currency code).
Defaults to "EUR" (Euro).
currency_date: The date for which to retrieve the exchange rate.
Defaults to "latest" for the most recent exchange rate data.
Can be specified in YYYY-MM-DD format for historical rates.
Returns:
dict: A dictionary containing the exchange rate information.
Example: {"amount": 1.0, "base": "USD", "date": "2023-11-24",
"rates": {"EUR": 0.95534}}
"""
import requests
response = requests.get(
f"https://api.frankfurter.app/{currency_date}",
params={"from": currency_from, "to": currency_to},
)
return response.json()
如要在代理中使用函式前先測試，請執行下列指令：
get_exchange_rate(currency_from="USD", currency_to="SEK")
回應應類似下方內容：
{'amount': 1.0, 'base': 'USD', 'date': '2025-04-03', 'rates': {'SEK': 9.6607}}
如要在 AdkApp 中使用工具，請將工具新增至 tools= 引數下的工具清單：
from google.adk.agents import Agent
agent = Agent(
model=model, # Required.
name='currency_exchange_agent', # Required.
tools=[get_exchange_rate], # Optional.
)
您可以使用 AdkApp.async_stream_query 方法對代理執行測試查詢，在本機測試代理。執行下列指令，在本機使用美元和瑞典克朗測試代理：
from vertexai.agent_engines import AdkApp
app = AdkApp(agent=agent)
async for event in app.async_stream_query(
user_id="USER_ID",
message="What is the exchange rate from US dollars to SEK on 2025-04-03?",
):
print(event)
其中 USER_ID 是您定義的使用者 ID。例如：user-123。
回應是一連串字典，類似如下：
{'author': 'currency_exchange_agent',
'content': {'parts': [{'function_call': {'args': {'currency_date': '2025-04-03',
'currency_from': 'USD',
'currency_to': 'SEK'},
'id': 'adk-e39f3ba2-fa8c-4169-a63a-8e4c62b89818',
'name': 'get_exchange_rate'}}],
'role': 'model'},
'id': 'zFyIaaif',
# ...
}
{'author': 'currency_exchange_agent',
'content': {'parts': [{'function_response': {'id': 'adk-e39f3ba2-fa8c-4169-a63a-8e4c62b89818',
'name': 'get_exchange_rate',
'response': {'amount': 1.0,
'base': 'USD',
'date': '2025-04-03',
'rates': {'SEK': 9.6607}}}}],
'role': 'user'},
'id': 'u2YR4Uom',
# ...
}
{'author': 'currency_exchange_agent',
'content': {'parts': [{'text': 'The exchange rate from USD to SEK on '
'2025-04-03 is 9.6607.'}],
'role': 'model'},
'id': 'q3jWA3wl',
# ...
}
選用：管理工作階段
AdkApp 在本機執行時，會使用記憶體內工作階段，將代理部署至 Agent Runtime 後，則會使用雲端式代管工作階段。本節說明如何設定 ADK 代理程式，以便與受管理工作階段搭配使用。
選用：自訂工作階段資料庫
如要使用自己的資料庫覆寫預設的受管理工作階段服務，可以按照下列方式定義 session_service_builder 函式：
def session_service_builder():
from google.adk.sessions import InMemorySessionService
return InMemorySessionService()
將資料庫以 session_service_builder= 的形式傳遞至 AdkApp：
from vertexai.agent_engines import AdkApp
app = AdkApp(
agent=agent, # Required.
session_service_builder=session_service_builder, # Optional.
)
搭配工作階段使用代理
在本機執行 AdkApp 時，下列操作說明會使用記憶體內工作階段。
如要為代理建立工作階段，請使用 AdkApp.async_create_session 方法：
session = await app.async_create_session(user_id="USER_ID")
print(session)
工作階段會建立為 ADK 工作階段物件的字典表示法。
如要列出與代理相關聯的工作階段，請使用 AdkApp.async_list_sessions 方法：
await app.async_list_sessions(user_id="USER_ID")
如要取得特定工作階段，請使用 AdkApp.async_get_session 方法：
session = await app.async_get_session(user_id="USER_ID", session_id="SESSION_ID")
其中
USER_ID 是您定義的使用者 ID。例如：
user-123。SESSION_ID 是要擷取的特定工作階段 ID。
如要以非同步方式查詢代理，請使用 AdkApp.async_stream_query 方法：
async for event in app.async_stream_query(
user_id="USER_ID",
session_id=SESSION_ID, # Optional. you can pass in the session_id when querying the agent
message="What is the exchange rate from US dollars to Swedish currency on 2025-04-03?",
):
print(event)
服務專員可能會要求您提供下列資訊：
{'author': 'currency_exchange_agent',
'content': {'parts': [{'text': 'I need to know the Swedish currency code to '
'provide you with the exchange rate.'}],
'role': 'model'},
'id': 'wIgZAtQ4',
#...
}
您可以代表 USER_ID 在 session 對應的工作階段中傳送回應 (例如 "SEK")，方法是指定：
async for event in app.async_stream_query(
user_id="USER_ID",
session_id=session.id, # Optional. you can pass in the session_id when querying the agent
message="SEK",
):
print(event)
您應該會收到對話的延續內容，例如下列字典序列：
{'author': 'currency_exchange_agent',
'content': {'parts': [{'function_call': {'args': {'currency_date': '2025-04-03',
'currency_from': 'USD',
'currency_to': 'SEK'},
'id': 'adk-2b9230a6-4b92-4a1b-9a65-b708ff6c68b6',
'name': 'get_exchange_rate'}}],
'role': 'model'},
'id': 'bOPHtzji',
# ...
}
{'author': 'currency_exchange_agent',
'content': {'parts': [{'function_response': {'id': 'adk-2b9230a6-4b92-4a1b-9a65-b708ff6c68b6',
'name': 'get_exchange_rate',
'response': {'amount': 1.0,
'base': 'USD',
'date': '2025-04-03',
'rates': {'SEK': 9.6607}}}}],
'role': 'user'},
'id': '9AoDFmiL',
# ...
}
{'author': 'currency_exchange_agent',
'content': {'parts': [{'text': 'The exchange rate from USD to SEK on '
'2025-04-03 is 1 USD to 9.6607 SEK.'}],
'role': 'model'},
'id': 'hmle7trT',
# ...
}
選用：管理回憶集錦
根據預設，AdkApp 在本機執行時，會使用記憶體內實作的代理記憶體，並在代理部署至 Agent Runtime 後，使用 Agent Platform Memory Bank。
開發 ADK 代理時，您可以加入 PreloadMemoryTool，控管代理擷取記憶的時間，以及記憶如何納入提示。下列範例代理一律會在每個回合開始時擷取記憶，並將記憶納入系統指令：
from google.adk.agents import Agent
from google.adk.tools.preload_memory_tool import PreloadMemoryTool
from vertexai.agent_engines import AdkApp
agent = Agent(
model="gemini-2.0-flash",
name='stateful_agent',
instruction="""You are a Vehicle Voice Agent, designed to assist users with information and in-vehicle actions.
1. **Direct Action:** If a user requests a specific vehicle function (e.g., "turn on the AC"), execute it immediately using the corresponding tool. You don't have the outcome of the actual tool execution, so provide a hypothetical tool execution outcome.
2. **Information Retrieval:** Respond concisely to general information requests with your own knowledge (e.g., restaurant recommendation).
3. **Clarity:** When necessary, try to seek clarification to better understand the user's needs and preference before taking an action.
4. **Brevity:** Limit responses to under 30 words.
""",
tools=[PreloadMemoryTool()],
)
app = AdkApp(agent=agent)
選用：自訂記憶服務
如要覆寫預設記憶體服務，可以定義 memory_service_builder 函式，傳回 BaseMemoryService，如下所示：
def memory_service_builder():
from google.adk.memory import InMemoryMemoryService
return InMemoryMemoryService()
將資料庫以 memory_service_builder= 的形式傳遞至 AdkApp：
from vertexai.agent_engines import AdkApp
app = AdkApp(
agent=agent, # Required.
memory_service_builder=memory_service_builder, # Optional.
)
使用具備記憶功能的代理
測試具備記憶功能的 ADK 代理：
建立工作階段並與代理互動：
initial_session = await app.async_create_session(user_id="USER_ID") async for event in app.async_stream_query( user_id="USER_ID", session_id=initial_session.id, message="Can you update the temperature to my preferred temperature?", ): print(event)由於第一個工作階段沒有可用的記憶，且代理不知道任何使用者偏好設定，因此代理可能會回覆「您偏好的溫度是多少？」您可以使用下列指令回覆：
async for event in app.async_stream_query( user_id="USER_ID", session_id=initial_session.id, message="I like it at 71 degrees", ): print(event)代理可能會傳回「將溫度設為華氏 71 度。溫度已成功變更。」代理的回覆可能會因使用的模型而異。
根據工作階段生成記憶。如要儲存工作階段中的資訊，以便在日後的工作階段中使用，請使用
async_add_session_to_memory方法：await app.async_add_session_to_memory(session=initial_session)建立新的工作階段並提示代理，測試代理是否保留工作階段的記憶 (使用
PreloadMemoryTool)：new_session = await app.async_create_session(user_id="USER_ID") async for event in app.async_stream_query( user_id="USER_ID", session_id=initial_session.id, message="Fix the temperature!", ): print(event)代理可能會傳回「將溫度設為 71 度。對嗎？」代理的回覆可能會因使用的模型和記憶體服務供應商而異。
使用
async_search_memory方法顯示代理的記憶：response = await app.async_search_memory( user_id="USER_ID", query="Fix the temperature!", ) print(response)