回呼

回呼是進階功能，可提供強大的機制，讓您使用 Python 程式碼掛鉤至特定代理的執行程序。您可以在特定預先定義的時機點，觀察、自訂，甚至是控制代理的行為。

您可以運用各種回呼類型，每種回呼類型都會在對話回合中的特定時間點執行。下文將說明這些類型。

Python 執行階段和類別

在 Python 回呼程式碼中，您可以存取特定類別和函式，協助您編寫程式碼。詳情請參閱 Python 執行階段參考資料。

沙箱和網路限制

Python 回呼程式碼會在沙箱環境中執行。這個環境有下列限制：

• 無法存取私人網路 (PNA)：即使已設定服務目錄，回呼也無法直接存取私人 IP 位址或解析私人 DNS 網域。如要存取私人網路中的資源，請改用支援私人網路存取權的工具類型 (例如 OpenAPI 或 MCP 工具)。
• 公開網際網路存取權：回呼只能存取網際網路上公開的端點。

回呼類型

視回呼類型而定，主要回呼函式必須使用特定名稱。您可以在回呼程式碼中定義任何名稱的輔助函式。

每種回呼都會在對話回合中的特定時間點執行：

如果您定義特定類型的多個回呼，系統會按照定義順序執行這些回呼。

下方各節說明每種回呼類型，並提供下列資訊：

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 model callback

程式碼範例：

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 回呼	after model callback

程式碼範例：

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
執行作業	工具完成後呼叫。
目的	此階段適用於在將結果回傳給模型之前，檢查或修改工具的回覆；也可用於對工具結果進行後處理，或將工具回覆中的特定部分儲存為變數。
引數	工具、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

建立回呼

如要建立回呼，請按照下列步驟操作：

1. 開啟代理程式設定。
2. 按一下「新增代碼」。
3. 選取回呼類型。
4. 提供 Python 程式碼。
5. 按一下 [儲存]。

自訂酬載 (custom_payloads)

自訂酬載可讓您在代理程式的回應中加入補充的非文字結構化資料 (通常為 JSON 格式)。這個酬載有助於引導或擴增代理程式與外部系統或用戶端應用程式的互動。

大型語言模型 (LLM) 無法看到酬載值，該值只會用於生成最終回覆。自訂酬載是使用回呼 (具體來說是 before_model_callback 或 after_model_callback) 產生及設定。

自訂酬載可用於多種用途，通常是為了啟用豐富的結構化互動：

• 轉交給真人服務專員： 提供轉送指示 (例如要轉送的特定佇列)， 將互動轉交給真人服務專員。
• 多媒體內容和用戶端動作： 支援將豐富的小工具和其他多媒體內容直接嵌入聊天體驗， 這對自訂聊天整合功能特別有用。
  • 例如，使用通話助理等介面時，向顧客顯示圖片網址或快速回覆晶片和選項。
• 回應組成： 您可以設定以各種方式傳回自訂酬載：
  • 僅以確定性方式傳回明確的酬載。
  • 連同 LLM 生成的文字回覆，一併傳回酬載。
  • 傳回含有靜態文字回應的酬載

設定代理程式

自訂酬載只能透過回呼產生及設定。 酬載設為 Blob，且 mime_type 為 application/json。

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_json_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)]))

在回應中執行階段驗證酬載

酬載會填入 payload 欄位，做為 RunSession 和 BidiRunSession 的 Struct。

LLM 看不到酬載值。