Memory Bank を使用するように Agent Development Kit（ADK）エージェントを構成すると、エージェントは Memory Bank への呼び出しをオーケストレートして、長期記憶を管理します。
このチュートリアルでは、ADK で Memory Bank を使用して長期記憶を管理する方法について説明します。
エージェントとやり取りして、セッション間でアクセス可能な長期記憶を動的に生成します。
クリーンアップします。
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 エージェントを作成する
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()] )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 ランナーによって調整されます。
エージェント、ツール、コールバックの実行をオーケストレートする ADK ランタイムを作成します。ADK Runner の設定は、使用しているデプロイ環境によって異なります。
adk.Runner
adk.Runner は通常、Colab などのローカル環境で使用されます。Agent Engine ランタイムなどのほとんどのデプロイ オプションでは、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)
次のように置き換えます。
- APP_NAME: ADK アプリの名前。
Agent Engine ADK テンプレート
Agent Engine ADK テンプレート（AdkApp）は、ローカルで使用することも、ADK エージェントを Agent Engine ランタイムにデプロイすることもできます。Agent Engine ランタイムにデプロイすると、Agent Engine ADK テンプレートはデフォルトのメモリ サービスとして VertexAiMemoryBankService を使用し、Agent Engine ランタイムと同じ Agent Engine インスタンスを Memory Bank に使用します。この場合、メモリ サービスを明示的に指定する必要はありません。
メモリバンクの動作をカスタマイズする方法など、Agent Engine ランタイムの設定の詳細については、Agent Engine を構成するをご覧ください。
次のコードを使用して、ADK エージェントを Agent Engine ランタイムにデプロイします。
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)
次のように置き換えます。
- PROJECT_ID: プロジェクト ID。
- LOCATION: リージョン。Memory Bank のサポートされているリージョンをご覧ください。
- AGENT_ENGINE_ID: メモリバンクで使用する Agent Engine ID。たとえば、
projects/my-project/locations/us-central1/reasoningEngines/456の456。 - STAGING_BUCKET: Agent Engine Runtime のステージングに使用する Cloud Storage バケット。
ローカルで実行する場合、ADK テンプレートはデフォルトのメモリ サービスとして InMemoryMemoryService を使用します。ただし、デフォルトのメモリ サービスをオーバーライドして VertexAiMemoryBankService を使用することはできます。
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)
次のように置き換えます。
- PROJECT_ID: プロジェクト ID。
- LOCATION: リージョン。Memory Bank のサポートされているリージョンをご覧ください。
- AGENT_ENGINE_ID: メモリバンクで使用する Agent Engine ID。たとえば、
projects/my-project/locations/us-central1/reasoningEngines/456の456。
エージェントを操作する
エージェントを定義してメモリバンクを設定したら、エージェントとやり取りできます。
最初のセッションを作成します。ユーザーとの最初のセッションでは利用可能なメモリがないため、エージェントはユーザーの好みの温度などのユーザー設定を認識していません。
adk.Runneradk.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: ユーザーの識別子。このセッションから生成されたメモリーは、この不透明な識別子でキー設定されます。生成された思い出のスコープは
{"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: ユーザーの識別子。このセッションから生成されたメモリーは、この不透明な識別子でキー設定されます。生成された思い出のスコープは
{"user_id": "USER_ID"}として保存されます。
現在のセッションの思い出を生成します。Memory Bank が会話から思い出を抽出すると、それらはスコープ
{"user_id": USER_ID, "app_name": APP_NAME}に保存されます。adk.Runnersession = 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)2 つ目のセッションを作成します。
PreloadMemoryToolを使用した場合、エージェントは各ターンの開始時にメモリを取得し、ユーザーが以前にエージェントに伝えた設定にアクセスします。adk.Runnersession = 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 プロジェクトを削除します。
それ以外の場合は、このチュートリアルで作成した個々のリソースを次のように削除できます。
次のコードサンプルを使用して Vertex AI Agent Engine インスタンスを削除します。これにより、その Vertex AI Agent Engine に属するセッションまたはメモリも削除されます。
agent_engine.delete(force=True)ローカルで作成したファイルを削除します。