Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

Agent Development Kit でセッションを管理する

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

始める前に

以下の手順では、ADK エージェントとそのサポートランナーとデプロイコードを定義するために、次の基本的なプロジェクトファイル構造を使用します。

my_agent/
    agent.py      # main agent code
    runner.py     # code for interacting with the agent
    deploy.py     # code for deploying the agent to Google Cloud

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

環境変数を設定する

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

import os

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

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

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

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

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

Google Cloud プロジェクト

import vertexai

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

# If you don't have an Agent Engine instance already, create an instance.
agent_engine = client.agent_engines.create()

# Print the agent engine ID, you will need it in the later steps to initialize
# the ADK `VertexAiSessionService`.
print(agent_engine.api_resource.name.split("/")[-1])

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

PROJECT_ID: プロジェクト ID。
LOCATION: リージョン。セッションでサポートされているリージョンをご覧ください。

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

ADK エージェントを作成するには、Agent Development Kitの手順に沿って操作するか、次のコードを使用して、固定の挨拶でユーザーに挨拶するエージェントを作成します。このコードを agent.py という名前のファイルに保存します。

# file: my_agent/agent.py
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 を初期化します。これにより、Agent Platform セッションに接続されます。このコードを runner.py という名前のファイルに保存します。

Google Cloud プロジェクト

# file: my_agent/runner.py
import agent # Import from your agent.py
from google.adk import Runner
from google.adk.sessions import VertexAiSessionService
from google.genai import types

app_name="APP_NAME"
user_id="USER_ID"

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

# Helper method to send query to the runner
async def call_agent(query, session_id, user_id):
  content = types.Content(role='user', parts=[types.Part(text=query)])
  async for event in runner.run_async(
      user_id=user_id, session_id=session_id, new_message=content):
      if event.is_final_response():
          final_response = event.content.parts[0].text
          print("Agent Response: ", final_response)

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

APP_NAME: エージェントアプリケーションの名前。
USER_ID: 文字数制限が 128 の独自のユーザー ID を選択します。例: user-123
AGENT_ENGINE_ID: Vertex AI Agent Engine インスタンスのリソース ID。
デプロイ済みエージェントの場合、リソース ID は GOOGLE_CLOUD_AGENT_ENGINE_ID 環境変数として一覧表示されます。
ローカルエージェントの場合、agent_engine.api_resource.name.split("/")[-1] を使用してリソース ID を取得できます。

エージェントを操作する

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

ADK UI

ADK ユーザーインターフェースでエージェントをテストし、Agent Platform セッションに接続するには、ターミナルで adk web コマンドを実行します。

adk web は通常、.env ファイルから環境変数を読み取りますが、--session_service_uri フラグを使用すると、このファイルは無視されます。代わりに、コマンドを実行する前に、ターミナルセッションで GOOGLE_CLOUD_PROJECT 環境変数と GOOGLE_CLOUD_LOCATION 環境変数を設定する必要があります。

export GOOGLE_CLOUD_PROJECT="PROJECT_ID"
export GOOGLE_CLOUD_LOCATION="LOCATION"
adk web --session_service_uri=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 コードを使用してセッションと状態を管理します。エージェントとやり取りするには、runner.py ファイルの末尾に次のコードを追加します。

次のスニペットには、簡潔にするために最上位の await 呼び出しが含まれています。このコードを Python スクリプトとして実行するには、次の例に示すように、スニペットを async 関数内に配置し、asyncio.run() を使用して実行します。

import asyncio

async def main():
  # Place one or more snippets here.
  # For example:
  session = await session_service.create_session(
         app_name=app_name,
         user_id=user_id)

  await call_agent("Hello!", session.id, user_id)

asyncio.run(main())

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

以下のコードを使用してセッションを作成し、エージェントにクエリを送信します。カスタムセッション ID を指定すると、独自の命名規則を使用してセッションを整理したり、他のサービスからすばやく移行したりできます。セッション ID を指定しない場合、Agent Platform は自動的にセッション ID を生成します。

# file: my_agent/runner.py
# Create a session
session = await session_service.create_session(
       app_name=app_name,
       user_id=user_id,
       session_id=session_id)

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

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

カスタムセッション ID の場合は、システム生成 ID との競合を避けるため、次の制限事項を考慮してください。

最初の文字が文字の場合、ID の長さは 63 文字までです。有効な文字は、小文字、数字、ハイフン（[a-z0-9-]）です。最後の文字は文字または数字にする必要があります。
最初の文字が数字の場合、ID の長さは 9 文字までです。有効な文字は、先頭にゼロが付かない数字（[0-9]）です。

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

セッションの有効期間（TTL）を構成する

すべてのセッションに有効期限が必要です。この有効期限は、セッションの作成時または更新時に定義できます。有効期限が切れると、セッションとその子イベントは自動的に削除されます。有効期限（expire_time）を直接設定することも、有効期間（ttl）を秒単位で設定することもできます。どちらも指定しない場合、デフォルトの TTL（365 日）が適用されます。

有効期間（TTL）

有効期間を設定すると、サーバーは新しく作成されたセッションの有効期限を create_time + ttl、更新されたセッションの有効期限を update_time + ttl として計算します。

  session = await session_service.create_session(
        app_name=app_name,
        user_id=user_id,
        # Session will be deleted 10 days after creation time.
        ttl=f"{24 * 60 * 60 * 10}s"
  )
  ```

有効期限

  import datetime

  expire_time = datetime.datetime.now(
        tz=datetime.timezone.utc) + datetime.timedelta(seconds=24 * 60 * 60 * 10)

  session = await session_service.create_session(
        app_name=app_name,
        user_id=user_id,
        # Session will be deleted at the provided time (10 days after current time).
        expire_time=expire_time.isoformat()
  )

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

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

# List sessions
sessions = await session_service.list_sessions(app_name=app_name,user_id=user_id)
print(sessions)
# 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 を使用して新しいイベントをセッションに追加します。

# file: my_agent/runner.py
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

イベントのカスタムフィールド

ADK は柔軟なイベントスキーマをサポートしているため、セッションイベントに任意のデータを含めることができます。これは、他のエージェントフレームワークとの相互運用や、カスタムイベントデータの保存に役立ちます。

カスタムフィールドを含むイベントを追加すると、ADK はイベントデータを保存されたセッションイベントの raw_event フィールドにシリアル化します。

# Create an event with custom fields
custom_event = Event(
  invocation_id="invocation_id",
  author="user",
  timestamp=time.time(),
  custom_field="custom_value",
  another_field={"nested_key": "nested_value"}
)

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

セッションを取得すると、取得したイベントにこれらのカスタムフィールドが保持されます。

セッションを削除する

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

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

エージェントを Agent Platform にデプロイする

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

Google Cloud プロジェクト

client.agent_engines.update(
    resource_name=agent_engine.api_resource.name,
    agent=AGENT,
    config={
      "display_name": DISPLAY_NAME,      # Optional.
      "requirements": REQUIREMENTS,      # Optional.
      "staging_bucket": STAGING_BUCKET,  # Required.
    },
)

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

AGENT：query / stream_query メソッドを実装するアプリケーション（ADK エージェントの場合は AdkApp など）。
DISPLAY_NAME: エージェントのわかりやすい名前。
REQUIREMENTS: エージェントに必要な pip パッケージのリスト。例: ["google-cloud-storage", "google-cloud-aiplatform[agent_engines,adk]"]
STAGING_BUCKET：gs:// が接頭辞として付いた Cloud Storage バケット。

クリーンアップ

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

agent_engine.delete(force=True)

次のステップ

API 呼び出しを使用してセッションを管理する。

Agent Development Kit でセッションを管理する コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。

始める前に

環境変数を設定する

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

Google Cloud プロジェクト

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

ADK ランナーを設定する

Google Cloud プロジェクト

エージェントを操作する

ADK UI

Python

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

セッションの有効期間（TTL）を構成する

有効期間（TTL）

有効期限

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

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

イベントのカスタム フィールド

セッションを削除する

エージェントを Agent Platform にデプロイする

Google Cloud プロジェクト

クリーンアップ

次のステップ

Agent Development Kit でセッションを管理する

イベントのカスタムフィールド