コールバック

コールバックは、Python コードを使用して特定のエージェントの実行プロセスにフックするための強力なメカニズムを提供する高度な機能です。コールバックを使用すると、特定の事前定義されたポイントでエージェントの動作を観察、カスタマイズ、制御できます。

利用できるコールバックのタイプはさまざまです。各タイプのコールバックは、会話のターンの特定の時点で実行されます。これらのタイプについては、以下のセクションで説明します。

Python ランタイムとクラス

Python コールバックコードでは、コードの作成に役立つ特定のクラスと関数にアクセスできます。詳細については、Python ランタイムリファレンスをご覧ください。

コールバックの種類

コールバックのタイプに応じて、プライマリコールバック関数は特定の名前である必要があります。これにより、コールバックコード内で任意の名前のヘルパー関数を定義できます。

各タイプのコールバックは、会話ターンの特定の時点で実行されます。

コールバックフロー

特定のタイプのコールバックを複数定義すると、定義した順序で実行されます。

以下のセクションでは、各コールバックタイプについて説明します。各タイプについて、次の情報が説明されています。

X	X
名前	必須のコールバック関数名
実行	会話ターンの実行ポイント。
目的	コールバックを使用する便利なシナリオ。
引数	関数の入力引数。
戻る	関数の戻り値。
ADK コールバック	対応する ADK コールバックドキュメントへのリンク。

エージェントの開始前（before_agent_callback）

X	X
名前	before_agent_callback
実行	エージェントが呼び出される前に呼び出されます。
目的	エージェントに必要なリソースまたは状態の設定、セッション状態の検証チェックの実行、エージェントの呼び出しの回避に役立ちます。
引数	CallbackContext
戻る	コンテンツ（省略可）: 設定されている場合、エージェントは呼び出されず、指定されたレスポンスが使用されます。
ADK コールバック	before agent callback

コードサンプル:

import random

def before_agent_callback(
  callback_context: CallbackContext
) -> Optional[Content]:
  username = callback_context.variables.get("username", None)
  if not username:
    # default user
    final_name = "Default Name"
  else:
    # add a random integer to the username
    final_name = f"{username} {random.randint(1,10)}"
  # update the username variable
  callback_context.variables["username"] = final_name

エージェントの終了後（after_agent_callback）

X	X
名前	after_agent_callback
実行	エージェントの完了後に呼び出されます。
目的	クリーンアップタスク、実行後の検証、最終状態の変更、エージェントレスポンスの更新に役立ちます。
引数	CallbackContext
戻る	コンテンツ（省略可）: 設定されている場合、エージェントの出力を指定された出力に置き換えます。
ADK コールバック	エージェントのコールバック後

コードサンプル:

def after_agent_callback(
  callback_context: CallbackContext
) -> Optional[Content]:
  if callback_context.agent_name == "Routing Agent":
    counter = callback_context.variables.get("counter", 0)
    counter += 1
    # increment the invoked counter for this agent
    callback_context.variables["counter"] = int(counter)

LLM 呼び出し前（before_model_callback）

X	X
名前	before_model_callback
実行	モデルリクエストの前に呼び出されます。
目的	モデルリクエストの検査や変更、モデルの使用の回避に役立ちます。
引数	CallbackContext、LlmRequest
戻る	LlmResponse: 設定されている場合、モデル呼び出しはスキップされ、レスポンスはモデルから返されたかのように使用されます。
ADK コールバック	before モデルコールバック

コードサンプル:

def before_model_callback(
  callback_context: CallbackContext,
  llm_request: LlmRequest
) -> Optional[LlmResponse]:
  """
  This callback executes *before* a request is sent to the LLM.

  By returning an `LlmResponse` object, we are intercepting the call to the
  LLM. The LLM will *not* be called, and the framework will instead use the
  `LlmResponse` we provide as if it came from the model.

  This is the core mechanism for implementing input guardrails, prompt
  validation, or serving responses from a cache. Here, we force the agent to
  call a function instead of thinking with the LLM.
  """
  # Modify the shared session state.
  callback_context.variables['foo'] = 'baz'

  # Skip the LLM call and return a custom response telling the agent to
  # execute a specific function.
  return LlmResponse(
    content=Content(parts=[Part(
      function_call=FunctionCall(
        name="function_name", args={"arg_name": "arg_value"}))],
      role="model"))

LLM 呼び出し後（after_model_callback）

X	X
名前	after_model_callback
実行	モデルレスポンスの受信後に呼び出されます。
目的	モデルのレスポンスの再フォーマット、モデルによって生成された機密情報の検閲、モデルから構造化データを解析して変数で使用する、モデルのエラー処理などに役立ちます。
引数	CallbackContext、LlmResponse
戻る	LlmResponse: 設定されている場合、モデルのレスポンスが指定されたレスポンスに置き換えられます。
ADK コールバック	モデルコールバック後

コードサンプル:

def after_model_callback(
  callback_context: CallbackContext,
  llm_response: LlmResponse
) -> Optional[LlmResponse]:
  """
  This callback executes *after* a response has been received from the LLM,
  but before the agent processes it.

  The `llm_response` parameter contains the actual data from the LLM.
  By returning `None`, we are approving this response and allowing the agent
  to use it as-is.

  If we returned a new `LlmResponse` object, it would *replace* the original,
  which is useful for redacting sensitive information, enforcing output
  formatting, or adding disclaimers.
  """
  # Returning None allows the LLM's actual response to be used.
  return None

ツール呼び出し前（before_tool_callback）

X	X
名前	before_tool_callback
実行	ツールの呼び出しの前に呼び出されます。
目的	ツールに渡される引数の検査と変更、ツール実行前の認可チェック、ツールレベルのキャッシュ保存の実装に役立ちます。
引数	Tool, Dict[str,Any]: ツールの入力, CallbackContext
戻る	Dict[str,Any] : 設定されている場合、ツールの実行はスキップされ、この出力がモデルに提供されます。
ADK コールバック	before tool callback

コードサンプル:

def before_tool_callback(
  tool: Tool,
  input: dict[str, Any],
  callback_context: CallbackContext
) -> Optional[dict[str, Any]]:
  """
  This callback executes *before* a specific tool is called by the agent.

  Here, we modify the input arguments intended for the tool and then return
  a dictionary. By returning a dictionary instead of `None`, we are
  overriding the default behavior. The actual tool function will *not* be
  executed. Instead, the dictionary we return will be treated as the
  llm.tool's result and passed back to the LLM for the next step.

  This is ideal for validating tool inputs, applying policies, or returning
  mocked/cached data for testing.
  """
  # Modify the shared session state.
  callback_context.variables['foo'] = 'baz'

  # Modify the arguments for the tool call in-place.
  input['input_arg'] = 'updated_val1'
  input['additional_arg'] = 'updated_val2'

  # Override the tool call and return a mocked result.
  return {"result": "ok"}

ツール呼び出し後（after_tool_callback）

X	X
名前	after_tool_callback
実行	ツールの完了後に呼び出されます。
目的	モデルに送信する前のツールレスポンスの検査と変更、ツール結果の後処理、ツールレスポンスの特定の部分の変数への保存に役立ちます。
引数	Tool, Dict[str,Any]: ツールの入力、CallbackContext, Dict[str,Any]: ツールのレスポンス
戻る	Dict[str,Any]: 設定されている場合、モデルに提供されるツールレスポンスをオーバーライドします。
ADK コールバック	ツールコールバック後

コードサンプル:

# Previous tool was named `get_user_info`
# Previous tool returned the payload:
# {"username": "Patrick", "fave_food": ["pizza"]}

def after_tool_callback(
  tool: Tool,
  input: dict[str, Any],
  callback_context: CallbackContext,
  tool_response: dict
) -> Optional[dict]:

  if tool.name == "get_user_info":
    tool_response["username"] = "Gary"
    tool_response["pet"] = "dog"

    # Override tool response
    return tool_response

コールバックを作成する

コールバックを作成するには:

エージェント設定を開きます。
[コードを追加] をクリックします。
コールバックの種類を選択します。
Python コードを提供する。
[保存] をクリックします。

カスタムペイロード（custom_payloads）

カスタムペイロードを使用すると、エージェントのレスポンスに補足的な非テキストの構造化データ（通常は JSON 形式）を含めることができます。このペイロードは、エージェントと外部システムまたはクライアントアプリケーションとのやり取りを指示または補強するうえで重要な役割を果たします。

ペイロード値は大規模言語モデル（LLM）には表示されません。最終レスポンスの生成にのみ使用されます。カスタムペイロードは、コールバック（具体的には before_model_callback または after_model_callback）を使用して生成および設定されます。

カスタムペイロードは、一般的にリッチで構造化されたインタラクションを可能にすることを目的として、さまざまな用途で使用できます。

エージェントのエスカレーション/ハンドオフ: ルーティング手順（ルーティング先の特定のキューなど）を指定して、インタラクションを人間のエージェントに転送するために頻繁に使用されます。
リッチコンテンツとクライアントサイドアクション: リッチウィジェットやその他のリッチコンテンツをチャットエクスペリエンスに直接埋め込むことができます。これは、カスタムチャット統合で特に便利です。
- たとえば、コールコンパニオンなどのインターフェースを使用している顧客に対して、画像 URL やクイック返信のチップとオプションを表示する場合などです。
レスポンスの構成: カスタムペイロードは、さまざまな方法で返されるように構成できます。
- 明示的なペイロードのみを決定論的に返します。
- ペイロードと LLM によって生成されたテキストレスポンスを返します。
- 静的なテキストレスポンスを含むペイロードを返す

エージェントの設定

カスタムペイロードは、コールバックを使用してのみ生成および設定できます。ペイロードは、mime_type が application/json の Blob として設定されます。

Part.from_json(data=payload_string)

after_model_callback のサンプル

これは、追加のカスタムペイロードレスポンスとともにモデルレスポンスを返す after_model_callback のサンプルです。

import json

def after_model_callback(callback_context: CallbackContext, llm_response: LlmResponse) -> Optional[LlmResponse]:
 """
 Adds a custom payload to every model response which is a text
 """
 if (llm_response.content.parts[0].text is not None):
   # construct payload
   payload_dict = { "custom_payload_key": "custom_payload_value"}
   payload_json_string = json.dumps(payload_dict)

   new_parts = []
   # Keep the origial agent response part, as model only sees text in the historical context.
   new_parts.append(Part(text=llm_response.content.parts[0].text))

   # Append custom payload
   new_parts.append(Part.from_json(data=payload_string))

   return LlmResponse(content=Content(parts=new_parts))

before_model_callback のサンプル

これは、特定のツールがトリガーされた後に追加のカスタムペイロードを返す before_model_callback のサンプルです。

import json

def has_escalate(llm_request: LlmRequest) -> bool:
  for content in llm_request.contents:
    for part in content.parts:
      if part.function_call and part.function_call.name == 'escalate':
        return True
  return False

def before_model_callback(callback_context: CallbackContext, llm_request: LlmRequest) -> Optional[LlmResponse]:
  # checks if `escalate` tool is being called
  if not has_escalate(llm_request):
    return None
  payload_dict = { "escalate": "user ask for escalation"}
  payload_json_string = json.dumps(payload_dict)

  return LlmResponse(content=Content(parts=[Part(text="ESCALATE!!!"), Part.from_json(data=payload_json_string)]))

レスポンスのペイロードをランタイムで検証

ペイロードは、RunSession と BidiRunSession の両方の payload フィールドの Struct として入力されます。

ペイロード値は LLM に表示されません。

コールバック コレクションでコンテンツを整理 必要に応じて、コンテンツの保存と分類を行います。

Python ランタイムとクラス

コールバックの種類

エージェントの開始前（before_agent_callback）

エージェントの終了後（after_agent_callback）

LLM 呼び出し前（before_model_callback）

LLM 呼び出し後（after_model_callback）

ツール呼び出し前（before_tool_callback）

ツール呼び出し後（after_tool_callback）

コールバックを作成する

カスタム ペイロード（custom_payloads）

エージェントの設定

after_model_callback のサンプル

before_model_callback のサンプル

レスポンスのペイロードをランタイムで検証

コールバック

カスタムペイロード（custom_payloads）