使用 Agent Development Kit 的快速入門

將 Agent Development Kit (ADK) 代理程式設定為使用記憶體庫後,代理程式就會協調對記憶體庫的呼叫,為您管理長期記憶

本教學課程說明如何使用 ADK 的 Memory Bank 管理長期記憶:

  1. 建立本機 ADK 代理和執行器。ADK 執行器會將代理程式連結至提供工作階段和記憶體管理功能的服務。

  2. 與代理互動,動態生成可在不同工作階段存取的長期記憶。

  3. 清除所用資源

如要直接呼叫 Memory Bank,而不使用 ADK 編排,請參閱「Agent Engine SDK 快速入門」。使用 Agent Engine SDK 有助於瞭解記憶體庫如何生成記憶,或檢查記憶體庫的內容。

事前準備

如要完成本教學課程中示範的步驟,請先按照「設定 Memory Bank」中的步驟操作。

設定環境變數

如要使用 ADK,請設定環境變數:

Google Cloud 專案

import os

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

更改下列內容:

Vertex AI Express 模式

import os

os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "TRUE"
os.environ["GOOGLE_API_KEY"] = "API_KEY"

更改下列內容:

  • API_KEY:您的 Vertex AI 快捷模式 API 金鑰。如果您是 Google Cloud的新使用者,可以透過 快速模式註冊,使用 Gmail 帳戶註冊並取得 API 金鑰。

建立 ADK 代理程式

  1. 開發 ADK 代理程式時,請加入 Memory 工具,控管代理程式擷取記憶內容的時間,以及記憶內容在提示中的納入方式。範例代理程式使用 PreloadMemoryTool,一律會在每個回合開始時擷取記憶內容,並將記憶內容納入系統指令:

    from google import adk
    
    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()]
    )
    
  2. 建立 VertexAiMemoryBankService 記憶體服務,ADK 執行器會使用這項服務擷取記憶體。如果您使用 Agent Engine ADK 範本,而非定義自己的 ADK 執行階段,則不必執行這項操作。如要設定 Memory Bank 執行個體的行為,請參閱設定指南

    Google Cloud 專案

    from google.adk.memory import VertexAiMemoryBankService
    import vertexai
    
    client = vertexai.Client(
      project="PROJECT_ID",
      location="LOCATION"
    )
    # If you don't have an Agent Engine instance already, create an Agent Engine
    # Memory Bank instance using the default configuration.
    agent_engine = client.agent_engines.create()
    
    # Optionally, print out the Agent Engine resource name. You will need the
    # resource name to interact with your Agent Engine instance later on.
    print(agent_engine.api_resource.name)
    
    agent_engine_id = agent_engine.api_resource.name.split("/")[-1]
    
    memory_service = VertexAiMemoryBankService(
        project="PROJECT_ID",
        location="LOCATION",
        agent_engine_id=agent_engine_id
    )
    

    更改下列內容:

    Vertex AI Express 模式

    from google.adk.memory import VertexAiMemoryBankService
    import vertexai
    
    client = vertexai.Client(
      api_key="API_KEY",
    )
    # If you don't have an Agent Engine instance already, create an Agent Engine
    # Memory Bank instance using the default configuration.
    agent_engine = client.agent_engines.create()
    
    # Optionally, print out the Agent Engine resource name. You will need the
    # resource name to interact with your Agent Engine instance later on.
    print(agent_engine.api_resource.name)
    
    agent_engine_id = agent_engine.api_resource.name.split("/")[-1]
    
    # Make sure your environment variables are set. VertexAiMemoryBankService
    # will read your api key from the GOOGLE_API_KEY environment variable.
    memory_service = VertexAiMemoryBankService(
        agent_engine_id=agent_engine_id
    )
    

    更改下列內容:

    • API_KEY:您的 Vertex AI 快捷模式 API 金鑰。如果您是 Google Cloud的新使用者,可以透過 Express Mode Registration 註冊並取得 API 金鑰。

    VertexAiMemoryBankService 是 ADK 針對 Memory Bank 定義的 ADK 包裝函式,使用與 Agent Engine SDK 不同的介面。BaseMemoryService您可以使用 Agent Engine SDK 直接呼叫 Memory Bank APIVertexAiMemoryBankService 介面包含:

    • memory_service.add_session_to_memory,這會使用提供的 adk.Session 做為來源內容,觸發對 Memory Bank 的 GenerateMemories 要求。ADK 執行器不會協調呼叫這個方法。如要使用 ADK 自動產生記憶體,您必須定義自己的回呼函式。

    • memory_service.search_memory,這會觸發對 Memory Bank 的 RetrieveMemories 要求,以擷取目前 user_idapp_name 的相關記憶。當您為代理提供記憶體工具時,ADK 執行器會自動調度管理對這個方法的呼叫。

  3. 建立 ADK 執行階段,負責協調代理程式、工具和回呼的執行作業。ADK Runner 設定取決於您使用的部署環境:

adk.Runner

adk.Runner 通常用於本機環境,例如 Colab。大多數部署選項 (例如 Agent Engine 執行階段) 都會提供 ADK 專用的執行階段。

import asyncio

from google.adk.sessions import VertexAiSessionService
from google.genai import types

# You can use any ADK session service. This example uses Agent Engine Sessions.
session_service = VertexAiSessionService(
    project="PROJECT_ID",
    location="LOCATION",
    agent_engine_id=agent_engine_id
)

# For Express Mode users, ensure your environment variables are set and
# initialize VertexAiSessionService as follows:
#
# session_service = VertexAiSessionService(
#    agent_engine_id=agent_engine_id
# )

app_name="APP_NAME"

runner = adk.Runner(
    agent=agent,
    app_name=app_name,
    session_service=session_service,
    memory_service=memory_service
)

async def call_agent(query, session, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  events = runner.run_async(
    user_id=user_id, session_id=session, new_message=content)

  async for event in events:
      if event.is_final_response():
          final_response = event.content.parts[0].text
          print("Agent Response: ", final_response)

更改下列內容:

  • APP_NAME:ADK 應用程式的名稱。

Agent Engine ADK 範本

Agent Engine ADK 範本 (AdkApp) 可在本機使用,也能將 ADK 代理部署至 Agent Engine 執行階段。部署至 Agent Engine 執行階段時,Agent Engine ADK 範本會使用 VertexAiMemoryBankService 做為預設記憶體服務,並將同一個 Agent Engine 執行個體用於記憶體庫和 Agent Engine 執行階段。在這種情況下,您不需要明確提供記憶體服務。

如要進一步瞭解如何設定 Agent Engine 執行階段,包括如何自訂記憶體庫的行為,請參閱「設定 Agent Engine」。

使用下列程式碼將 ADK 代理部署至 Agent Engine Runtime:

import asyncio

import vertexai
from vertexai.agent_engines import AdkApp

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

adk_app = AdkApp(agent=agent)

# Create a new Agent Engine with your agent deployed to Agent Engine Runtime.
# The Agent Engine instance will also include an empty Memory Bank.
agent_engine = client.agent_engines.create(
      agent_engine=adk_app,
      config={
            "staging_bucket": "STAGING_BUCKET",
            "requirements": ["google-cloud-aiplatform[agent_engines,adk]"]
      }
)

# Alternatively, update an existing Agent Engine to deploy your agent to Agent Engine Runtime.
# Your agent will have access to the Agent Engine instance's existing memories.
agent_engine = client.agent_engines.update(
      name=agent_engine.api_resource.name,
      agent_engine=adk_app,
      config={
            "staging_bucket": "STAGING_BUCKET",
            "requirements": ["google-cloud-aiplatform[agent_engines,adk]"]
      }
)

async def call_agent(query, session_id, user_id):
    async for event in agent_engine.async_stream_query(
        user_id=user_id,
        session_id=session_id,
        message=query,
    ):
        print(event)

更改下列內容:

  • PROJECT_ID:您的專案 ID。
  • LOCATION:您的區域。請參閱「支援記憶體空間的區域」。
  • AGENT_ENGINE_ID:用於記憶體庫的 Agent Engine ID。例如,456 位於 projects/my-project/locations/us-central1/reasoningEngines/456 中。
  • STAGING_BUCKET:用於暫存 Agent Engine Runtime 的 Cloud Storage 值區。

在本機執行時,ADK 範本會使用 InMemoryMemoryService 做為預設記憶體服務。不過,您可以覆寫預設記憶體服務,改用 VertexAiMemoryBankService

# For Google Cloud Project users:
def memory_bank_service_builder():
    return VertexAiMemoryBankService(
        project="PROJECT_ID",
        location="LOCATION",
        agent_engine_id="AGENT_ENGINE_ID"
    )
# For Express Mode users, ensure your environment variables are set.
#
# def memory_bank_service_builder():
#    return VertexAiMemoryBankService(
#        agent_engine_id="AGENT_ENGINE_ID"
#    )

adk_app = AdkApp(
      agent=adk_agent,
      # Override the default memory service.
      memory_service_builder=memory_bank_service_builder
)

async def call_agent(query, session_id, user_id):
  # adk_app is a local agent. If you want to deploy it to Agent Engine Runtime,
  # use `client.agent_engines.create(...)` or `client.agent_engines.update(...)`
  # and call the returned Agent Engine instance instead.
  async for event in adk_app.async_stream_query(
      user_id=user_id,
      session_id=session_id,
      message=query,
  ):
      print(event)

更改下列內容:

  • PROJECT_ID:您的專案 ID。
  • LOCATION:您的區域。請參閱「支援記憶體空間的區域」。
  • AGENT_ENGINE_ID:用於記憶體庫的 Agent Engine ID。例如,456 位於 projects/my-project/locations/us-central1/reasoningEngines/456 中。

與虛擬服務專員互動

定義代理並設定記憶庫後,即可與代理互動。

  1. 建立第一個講座。由於在與使用者進行第一次對話時,沒有可用的記憶內容,因此服務專員不知道任何使用者偏好設定,例如偏好的溫度:

    adk.Runner

    使用 adk.Runner 時,您可以直接呼叫 ADK 記憶體和工作階段服務。

    # Use `asyncio.run(session_service.create(...))` if you're running this
    # code as a standard Python script.
    session = await session_service.create_session(
        app_name="APP_NAME",
        user_id="USER_ID"
    )
    
    # Use `asyncio.run(call_agent(...))` if you're running this code as a
    # standard Python script.
    await call_agent(
        "Can you update the temperature to my preferred temperature?",
        session.id,
        "USER_ID"
    )
    # Agent response: "What is your preferred temperature?"
    
    # Use `asyncio.run(call_agent(...))` if you're running this code as a
    # standard Python script.
    await call_agent(
      "I like it at 71 degrees", session.id, "USER_ID")
    # Agent Response:  Setting the temperature to 71 degrees Fahrenheit.
    # Temperature successfully changed.
    

    更改下列內容:

    • APP_NAME:執行器的應用程式名稱。
    • USER_ID:使用者的 ID。系統會使用這個不透明的 ID,為這個工作階段產生的回憶內容建立索引。產生的記憶集錦範圍會儲存為 {"user_id": "USER_ID"}

    Agent Engine ADK 範本

    使用 Agent Engine ADK 範本時,您可以呼叫 Agent Engine Runtime,與記憶體和工作階段互動。

    # Use `asyncio.run(agent_engine.async_create_session(...))` if you're
    # running this code as a standard Python script.
    session = await agent_engine.async_create_session(user_id="USER_ID")
    
    # Use `asyncio.run(call_agent(...))` if you're running this code as a
    # standard Python script.
    await call_agent(
        "Can you update the temperature to my preferred temperature?",
        session.get("id"),
        "USER_ID"
    )
    # Agent response: "What is your preferred temperature?"
    
    # Use `asyncio.run(call_agent(...))` if you're running this code as a
    # standard Python script.
    await call_agent("I like it at 71 degrees", session.get("id"), "USER_ID")
    # Agent Response:  Setting the temperature to 71 degrees Fahrenheit.
    # Temperature successfully changed.
    

    更改下列內容:

    • USER_ID:使用者的 ID。系統會使用這個不透明的 ID,為這個工作階段產生的回憶內容加上索引鍵。產生的記憶集錦範圍會儲存為 {"user_id": "USER_ID"}
  2. 為目前的工作階段生成回憶集錦。如果記憶庫從對話中擷取回憶,這些回憶會儲存在 {"user_id": USER_ID, "app_name": APP_NAME} 範圍內。

    adk.Runner

    async def trigger_memory_generation(app_name, session):
        # Refresh your `session` object so that all of the session events locally available.
        session = await session_service.get_session(
            app_name=app_name,
            user_id="USER_ID",
            session_id=session.id
        )
        await memory_service.add_session_to_memory(session)
    
    # Use `asyncio.run(trigger_memory_generation(...))` if you're running this
    # code as a standard Python script.
    await trigger_memory_generation(app_name, session)
    

    Agent Engine ADK 範本

    # Use `asyncio.run(agent_engine.async_add_session_to_memory(...))` if
    # you're running this code as a standard Python script.
    await agent_engine.async_add_session_to_memory(session=session)
    
  3. 建立第二個講座。如果您使用 PreloadMemoryTool,代理程式會在每一回合開始時擷取記憶內容,存取使用者先前向代理程式傳達的偏好設定。

    adk.Runner

    # Use `asyncio.run(session_service.create_session(...))` if you're running
    # this code as a standard Python script.
    session = await session_service.create_session(
        app_name=app_name,
        user_id="USER_ID"
    )
    
    # Use `asyncio.run(call_agent(...))` if you're running this code as a
    # standard Python script.
    await call_agent("Fix the temperature!", session.id, "USER_ID")
    # Agent Response:  Setting temperature to 71 degrees.  Is that correct?
    

    您也可以使用 memory_service.search_memory 直接擷取回憶集錦:

    # Use `asyncio.run(memory_service.search_memory(...))` if you're running
    # this code as a standard Python script.
    await memory_service.search_memory(
        app_name="APP_NAME",
        user_id="USER_ID",
        query="Fix the temperature!",
    )
    

    Agent Engine ADK 範本

    # Use `asyncio.run(agent_engine.async_add_session_to_memory(...))` if
    # you're running this code as a standard Python script.
    session = await agent_engine.async_create_session(user_id="USER_ID")
    
    # Use `asyncio.run(call_agent(...))` if you're running this code as a
    # standard Python script.
    await call_agent("Fix the temperature!", session.get("id"), "USER_ID")
    # Agent Response:  Setting temperature to 71 degrees.  Is that correct?
    

    您也可以使用 agent_engine.async_search_memory 直接擷取回憶集錦: 注意:如要使用 async_search_memory,您的 AdkApp 必須以 google-cloud-aiplatform 1.110.0 以上版本建立。否則,您可以直接呼叫 Memory Bank 來擷取記憶體。

    # Use `asyncio.run(agent_engine.async_search_memory(...))` if you're
    # running this code as a standard Python script.
    await agent_engine.async_search_memory(
        user_id="USER_ID",
        query="Fix the temperature!",
    )
    

清除所用資源

如要清除此專案中使用的所有資源,您可以刪除用於本快速入門導覽課程的 Google Cloud 專案

或者,您也可以按照下列步驟,刪除在本教學課程中建立的個別資源:

  1. 使用下列程式碼範例刪除 Vertex AI Agent Engine 執行個體,這也會刪除屬於該 Vertex AI Agent Engine 的任何工作階段或記憶體。

    agent_engine.delete(force=True)
    
  2. 刪除所有在本機建立的檔案。

後續步驟