Agent Development Kit のクイックスタート

Memory Bank を使用するように Agent Development Kit���ADK）エージェントを構成すると、エージェントは Memory Bank への呼び出しをオーケストレートして、長期記憶を管理します。

このチュートリアルでは、ADK で Memory Bank を使用して長期記憶を管理する方法について説明します。

1. ローカル ADK エージェントとランナーを作成します。

2. エージェントとやり取りして、セッション間でアクセス可能な長期記憶を動的に生成します。

3. クリーンアップします。

ADK オーケストレーションなしで Memory Bank への呼び出しを直接行うには、Agent Engine SDK のクイックスタートをご覧ください。Agent Engine SDK を使用すると、Memory Bank がメモリを生成する方法を理解することや、Memory Bank の内容を検査することが可能になります。

始める前に

このチュートリアルで説明する手順を完了するには、まず Memory Bank の設定の手順に沿って操作する必要があります。

環境変数を設定する

ADK を使用するには、環境変数を設定します。

import os

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

次のように置き換えます。

• PROJECT_ID: プロジェクト ID。
• LOCATION: リージョン。Memory Bank のサポートされているリージョンをご覧ください。

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. ADK ランナーがメモリの取得に使用する VertexAiMemoryBankService メモリサービスを作成します。独自の ADK ランタイムを定義する代わりに Agent Engine ADK テンプレートを使用する場合は、これは省略可能です。

   from google.adk.memory import VertexAiMemoryBankService
   
   agent_engine_id = agent_engine.api_resource.name.split("/")[-1]
   
   memory_service = VertexAiMemoryBankService(
       project="PROJECT_ID",
       location="LOCATION",
       agent_engine_id=agent_engine_id
   )
   

   VertexAiMemoryBankService は、ADK の BaseMemoryService で定義され、Agent Engine SDK とは異なるインターフェースを使用する Memory Bank の ADK ラッパーです。Agent Engine SDK を使用して、Memory Bank に API を直接呼び出すことが可能です。VertexAiMemoryBankService インターフェースには次のものが含まれます。

   • memory_service.add_session_to_memory。指定された adk.Session をソース コンテンツとして使用して、Memory Bank への GenerateMemories リクエストをトリガーします。このメソッドの呼び出しは ADK ランナーによってオーケストレーションされません。ADK でメモリ生成を自動化する場合は、独自のコールバック関数を定義する必要があります。

   • memory_service.search_memory。これにより、Memory Bank への RetrieveMemories リクエストがトリガーされ、現在の user_id と app_name に関連するメモリが取得されます。このメソッドの呼び出しは、エージェントにメモリツールを提供したときに、ADK ランナーによってオーケストレーションされます。

3. エージェント、ツール、コールバックの実行をオーケストレートする ADK ランタイムを作成します。ADK ランナーの設定は、使用しているデプロイ環境によって異なります。

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

# You can use any ADK session service.
session_service = VertexAiSessionService(
    project="PROJECT_ID",
    location="LOCATION",
    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
)

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

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

import vertexai
from vertexai.preview.reasoning_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)

def memory_bank_service_builder():
    return VertexAiMemoryBankService(
        project="PROJECT_ID",
        location="LOCATION",
        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)

エージェントを操作する

エージェントを定義して Memory Bank を設定したら、エージェントとやり取りできます。

1. 最初のセッションを作成します。ユーザーとの最初のセッションでは利用可能なメモリがないため、エージェントはユーザーの好み（ユーザーが好む温度など）を認識していません。

   adk.Runner

   adk.Runner を使用すると、ADK メモリサービスとセッション サービスを直接呼び出すことが可能です。

   session = await session_service.create_session(
       app_name="APP_NAME",
       user_id="USER_ID"
   )
   
   call_agent(
       "Can you update the temperature to my preferred temperature?",
       session.id,
       "USER_ID"
   )
   
   # Agent response: "What is your preferred temperature?"
   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 をキーとしています。生成されたメモリのスコープは {"user_id": "USER_ID"} として保存されます。

   Agent Engine ADK テンプレート

   Agent Engine ADK テンプレートを使用すると、Agent Engine Runtime を呼び出して、メモリとセッションを操作できます。

   session = await agent_engine.async_create_session(user_id="USER_ID")
   
   await call_agent(
       "Can you update the temperature to my preferred temperature?",
       session.get("id"),
       "USER_ID"
   )
   
   # Agent response: "What is your preferred temperature?"
   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 をキーとしています。生成されたメモリのスコープは {"user_id": "USER_ID"} として保存されます。

2. 現在のセッションのメモリを生成します。Memory Bank が会話からメモリを抽出すると、それらはスコープ {"user_id": USER_ID, "app_name": APP_NAME} に保存されます。

   adk.Runner

   session = await session_service.get_session(
       app_name=app_name,
       user_id="USER_ID",
       session_id=session.id
   )
   memory_service.add_session_to_memory(session)
   

   Agent Engine ADK テンプレート

   await agent_engine.async_add_session_to_memory(session=session)
   

3. 2 番目のセッションを作成します。PreloadMemoryTool ���使用��た場合、エージェントは各ターンの開始時にメモリを取得し、ユーザーが以前にエージェントに伝えた設定にアクセスします。

   adk.Runner

   session = await session_service.create_session(
       app_name=app_name,
       user_id="USER_ID"
   )
   
   call_agent("Fix the temperature!", session.id, "USER_ID")
   # Agent Response:  Setting temperature to 71 degrees.  Is that correct?
   

   memory_service.search_memory を使用して、メモリを直接取得することもできます。

   await memory_service.search_memory(
       app_name="APP_NAME",
       user_id="USER_ID",
       query="Fix the temperature!",
   )
   

   Agent Engine ADK テンプレート

   session = await agent_engine.async_create_session(user_id="USER_ID")
   
   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 への呼び出しを直接行うことでメモリを取得できます。

   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. ローカルで作成したファイルを削除します。

次のステップ

• Agent Engine SDK によるクイックスタート。