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 エージェントを作成するには、Agent Development Kit の手順に沿って操作するか、次のコードを使用して、固定の挨拶でユーザーに挨拶するエージェントを作成します。

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 ランタイムは、エージェント、ツール、コールバックの実行を調整し、セッションの読み取りと書き込みの呼び出しを調整します。VertexAiSessionService を使用して Runner を初期化します。これにより、Vertex AI Agent Engine セッションに接続されます。

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: 使用するリージョン。

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

次のステップ