回呼是進階功能,可透過 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 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
建立回呼
如要建立回呼,請按照下列步驟操作:
- 開啟代理程式設定。
- 按一下「新增代碼」。
- 選取回呼類型。
- 提供 Python 程式碼。
- 按一下 [儲存]。
自訂酬載 (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_string))
return LlmResponse(content=Content(parts=new_parts))
before_model_callback 範例
這是 sample_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)]))
在回應中執行階段驗證酬載
酬載會以 Struct 形式填入 RunSession 和 BidiRunSession 的 payload 欄位。
LLM 看不到酬載值。