開發 Agent Development Kit 代理程式

本頁說明如何使用 Agent Development Kit 範本 (Vertex AI SDK for Python 中的 AdkApp 類別) 開發代理程式。代理程式會傳回指定日期的兩種貨幣匯率。

操作步驟如下：

  1. 定義及設定模型
  2. (選用) 定義及使用工具
  3. (選用) 管理工作階段
  4. (選用) 管理回憶集錦

事前準備

請按照「設定環境」一文中的步驟，確認環境已設定完成。

定義及設定模型

定義模型版本：

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 筆記本等互動式環境中執行作業，可以執行查詢做為中繼測試步驟：

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.
)

您可以對代理執行測試查詢，在本機測試代理。執行下列指令，使用美元和瑞典克朗在本機測試代理程式：

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 在本機執行時會使用記憶體內工作階段，將代理程式部署至 Vertex AI Agent Engine 後，則會使用雲端式管理工作階段。本節說明如何設定 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 時，下列指令會使用記憶體內的工作階段：

為代理程式建立工作階段：

session = await app.async_create_session(user_id="USER_ID")
print(session)

工作階段會以 ADK 工作階段物件的字典表示法建立。

列出與代理程式相關聯的工作階段：

await app.async_list_sessions(user_id="USER_ID")

取得特定工作階段：

session = await app.async_get_session(user_id="USER_ID", session_id="SESSION_ID")

其中

  • USER_ID 是您定義的使用者 ID。例如：user-123。

  • SESSION_ID 是要擷取的特定工作階段 ID。

使用工作階段查詢 AdkApp：

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',
 #...
}

您可以在對應於 session 的工作階段中，代表 USER_ID 傳送回覆 (例如 "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 會在本地執行時使用代理記憶體的記憶體內實作，並在您將代理部署至 Vertex AI Agent Engine 後，使用 Vertex AI Agent Engine Memory Bank 。

開發 ADK 代理程式時，您可以加入 PreloadMemoryTool，控制代理程式擷取記憶的時間，以及記憶在提示中的納入方式。在下列範例中，代理程式一律會在每個回合開始時擷取記憶內容，並將記憶內容納入系統指令：

from google import adk
from vertexai.agent_engines import AdkApp

agent = adk.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=[adk.tools.preload_memory_tool.PreloadMemoryTool()],
)
app = AdkApp(agent=agent)

(選用) 自訂記憶體服務

如要覆寫預設記憶體服務，可以定義傳回 BaseMemoryService 的 memory_service_builder 函式，如下所示：

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 代理程式：

  1. 建立工作階段並與代理程式互動：

    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)

    代理程式可能會傳回「將溫度設為 22 度。溫度已成功變更。代理程式的回覆內容可能會因使用的模型而異。

  2. 從工作階段生成回憶集錦。如要儲存工作階段的資訊，以供日後使用，請使用 async_add_session_to_memory 方法：

    await app.async_add_session_to_memory(session=initial_session)

  3. 建立新工作階段並提示代理程式，測試代理程式是否保留工作階段的記憶體 (使用 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)

    代理程式可能會傳回「將溫度設為 22 度。Is that correct?" 代理程式的回覆內容可能因使用的模型和記憶體服務供應商而異。

  4. 使用 async_search_memory 方法顯示代理程式的記憶內容：

    response = await app.async_search_memory(
    user_id="USER_ID",
    query="Fix the temperature!",
)
print(response)

後續步驟