Agent Development Kit を使用してセッションを管理する

このページでは、Agent Development Kit（ADK）エージェントを Vertex AI Agent Engine セッションに接続し、ローカル環境と本番環境でマネージド セッションを使用する方法について説明します。

始める前に

環境の設定の必要なロールを取得すると認証の手順に沿って環境が設定されていることを確認します。

Vertex AI Agent Engine インスタンスを作成する

Vertex AI Agent Engine セッションにアクセスするには、まず Vertex AI Agent Engine インスタンスを作成する必要があります。セッションの使用を開始するためにコードをデプロイする必要はありません。コードをデプロイしなくても、Vertex AI Agent Engine インスタンスの作成は数秒しかかかりません。

import vertexai
from vertexai import agent_engines

# Create an agent engine instance
agent_engine = agent_engines.create()

ADK エージェントを開発する

ADK エージェントを作成するには、エージェント開発キットの手順に沿って作成するか、次のコードを使用して、固定の挨拶でユーザーに応答するエージェントを作成します。

from google import adk

def greetings(query: str):
  """Tool to greet user."""
  if 'hello' in query.lower():
    return {"greeting": "Hello, world"}
  else:
    return {"greeting": "Goodbye, world"}

# Define an ADK agent
root_agent = adk.Agent(
    model="gemini-2.0-flash",
    name='my_agent',
    instruction="You are an Agent that greet users, always use greetings tool to respond.",
    tools=[greetings]
)

ADK ランナーを設定する

ADK Runtime は、エージェント、ツール、コールバックの実行をオーケストレートし、セッションの読み取りと書き込みの呼び出しをオーケストレートします。Vertex AI Agent Engine セッションに接続する VertexAiSessionService を使用してランナーを初期化します。

from google.adk.sessions import VertexAiSessionService

app_name="AGENT_ENGINE_ID"
user_id="USER_ID"

# Create the ADK runner with VertexAiSessionService
session_service = VertexAiSessionService(
       "PROJECT_ID", "LOCATION")
runner = adk.Runner(
    agent=root_agent,
    app_name=app_name,
    session_service=session_service)

# Helper method to send query to the runner
def call_agent(query, session_id, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  events = runner.run(
      user_id=user_id, session_id=session_id, new_message=content)

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

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

  • PROJECT_ID: プロジェクト ID。

  • LOCATION: リージョン。Vertex AI エージェント エンジン セッションでは us-central1 のみがサポートされています。

  • AGENT_ENGINE_ID: Vertex AI Agent Engine インスタンスのリソース ID。

    • デプロイされたエージェントの場合、リソース ID は GOOGLE_CLOUD_AGENT_ENGINE_ID 環境変数として一覧表示されます。

    • ローカル エージェントの場合は、agent_engine.name.split("/")[-1] を使用してリソース ID を取得できます。

  • USER_ID: ユーザーの一意の識別子（空でないもの）。最大長は 128 文字です。

エージェントを操作する

エージェントを定義して Vertex AI Agent Engine セッションを設定したら、エージェントとやり取りして、セッション履歴と状態が保持されていることを確認できます。

ADK UI

ADK ユーザー インターフェースでエージェントをテストし、session_db_url コマンドライン オプションを使用して Vertex AI Agent Engine セッションに接続します。

agent_engine_id="AGENT_ENGINE_ID"

adk web --session_db_url=agentengine://${agent_engine_id}

# Sample output
+-----------------------------------------------------------------------------+
| ADK Web Server started                                                      |
|                                                                             |
| For local testing, access at http://localhost:8000.                         |
+-----------------------------------------------------------------------------+

INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

ADK UI

Python

ADK Python コードを使用して、セッションと状態を管理します。

セッションを作成し、エージェントにクエリを実行する

次のコードを使用してセッションを作成し、エージェントにクエリを送信します。

# Create a session
session = await session_service.create_session(
       app_name=app_name,
       user_id=user_id)

call_agent("Hello!", session.id, user_id)
# Agent response: "Hello, world"

call_agent("Thanks!", session.id, user_id)
# Agent response: "Goodbye, world"

セッションが作成されてランナーに渡されると、ADK はセッションを使用して現在のインタラクションのイベントを保存します。前のセッションの ID を指定して、そのセッションを再開することもできます。

既存のセッションを一覧表示する

特定のユーザー ID に関連付けられている既存のセッションをすべて一覧表示します。

# List sessions
await session_service.list_sessions(app_name=app_name,user_id=user_id)

# ListSessionsResponse(session_ids=['1122334455', '9988776655'])

セッションの状態を管理する

状態には、エージェントが会話に必要とする情報が保持されます。セッションを作成するときに、初期状態を辞書として指定できます。

# Create a session with state
session = await session_service.create_session(
      app_name=app_name,
      user_id=user_id,
      state={'key': 'value'})

print(session.state['key'])
# value

ランナーの外部でセッション状態を更新するには、state_delta を使用して新しいイベントをセッションに追加します。

from google.adk.events import Event, EventActions
import time

# Define state changes
state_changes = {'key': 'new_value'}

# Create event with actions
actions_with_update = EventActions(state_delta=state_changes)
system_event = Event(
  invocation_id="invocation_id",
  author="system", # Or 'agent', 'tool' etc.
  actions=actions_with_update,
  timestamp=time.time()
)

# Append the event
await session_service.append_event(session, system_event)

# Check updated state
updated_session = await session_service.get_session(
      app_name=app_name,
      user_id=user_id,
      session_id=session.id)
# State is updated to new value
print(updated_session.state['key'])
# new_value

セッションを削除する

ユーザー ID に関連付けられている特定のセッションを削除します。

await session_service.delete_session(app_name=app_name, user_id=user_id, session_id=session.id)

エージェントを Vertex AI Agent Engine にデプロイする

エージェントをローカルでテストしたら、Vertex AI Agent Engine インスタンスをパラメータで更新して、エージェントを本番環境にデプロイできます。

agent_engines.update(resource_name=agent_engine.name, agent_engine=AGENT, requirements=REQUIREMENTS)

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

  • AGENT: query / stream_query メソッドを実装するアプリケーション（ADK エージェントの AdkApp など）。詳細については、デプロイに関する考慮事項をご覧ください。

クリーンアップ

このプロジェクトで使用したすべてのリソースをクリーンアップするには、Vertex AI Agent Engine インスタンスとその子リソースを削除します。

agent_engine.delete(force=True)

次のステップ