Skip to content

回调：观察、自定义和控制智能体行为

介绍：什么是回调及为何使用它们？

回调是 ADK 的核心功能，提供了一种强大的机制来挂钩智能体的执行过程。它们允许你在特定的预定义点观察、自定义甚至控制智能体的行为，而无需修改 ADK 框架的核心代码。

什么是回调？ 本质上，回调是你定义的标准 Python 函数。然后在创建智能体时将这些函数与智能体关联。ADK 框架会在智能体生命周期的关键阶段自动调用你的函数，例如：

  • 在智能体的主要处理逻辑运行之前或之后。
  • 在向大型语言模型 (LLM) 发送请求之前，或从 LLM 接收响应之后。
  • 在执行工具 (如 Python 函数或另一个智能体) 之前或完成之后。

intro_components.png

为什么使用它们？ 回调解锁了显著的灵活性并支持高级智能体功能：

  • 观察与调试： 在关键步骤记录详细信息，用于监控和故障排除。
  • 自定义与控制： 根据你的逻辑修改流经智能体的数据 (如 LLM 请求或工具结果)，甚至完全跳过某些步骤。
  • 实现防护机制： 强制执行安全规则，验证输入/输出，或阻止不允许的操作。
  • 管理状态： 在执行期间读取或动态更新智能体的会话状态。
  • 集成与增强： 触发外部操作 (API 调用、通知) 或添加缓存等功能。

如何添加它们？ 你可以在创建Agent或LlmAgent实例时，通过将定义的 Python 函数作为参数传递给智能体的构造函数 (__init__) 来注册回调。

from google.adk.agents import LlmAgent
from google.adk.agents.callback_context import CallbackContext
from google.adk.models import LlmResponse, LlmRequest
from typing import Optional

# --- Define your callback function ---
def my_before_model_logic(
    callback_context: CallbackContext, llm_request: LlmRequest
) -> Optional[LlmResponse]:
    print(f"Callback running before model call for agent: {callback_context.agent_name}")
    # ... your custom logic here ...
    return None # Allow the model call to proceed

# --- Register it during Agent creation ---
my_agent = LlmAgent(
    name="MyCallbackAgent",
    model="gemini-2.0-flash", # Or your desired model
    instruction="Be helpful.",
    # Other agent parameters...
    before_model_callback=my_before_model_logic # Pass the function here
)

回调机制：拦截与控制

当 ADK 框架遇到可以运行回调的点 (例如，就在调用 LLM 之前) 时，它会检查你是否为该智能体提供了相应的回调函数。如果提供了，框架会执行你的函数。

上下文至关重要： 你的回调函数不是孤立调用的。框架提供特殊的上下文对象(CallbackContext或ToolContext) 作为参数。这些对象包含关于智能体执行当前状态的重要信息，包括调用详情、会话状态，以及可能对服务 (如 artifacts 或 memory) 的引用。你使用这些上下文对象来了解情况并与框架交互。(详见"上下文对象"专门章节)。

控制流程 (核心机制)： 回调最强大的方面在于其返回值如何影响智能体后续的操作。这就是你拦截和控制执行流程的方式：

  1. return None (允许默认行为)：

    • 这是标准方式，表示你的回调已完成工作 (例如，记录、检查、对可变输入参数如llm_request进行小修改)，并且 ADK 智能体应该继续其正常操作。
    • 对于before_*回调 (before_agent, before_model, before_tool)，返回None意味着序列中的下一步 (运行智能体逻辑、调用 LLM、执行工具) 将会发生。
    • 对于after_*回调 (after_agent, after_model, after_tool)，返回None意味着前一步骤刚产生的结果 (智能体的输出、LLM 的响应、工具的结果) 将按原样使用。

  2. return <特定对象> (覆盖默认行为)：

    • 返回特定类型的对象(而不是None) 是你覆盖ADK 智能体默认行为的方式。框架将使用你返回的对象，并跳过通常会后续执行的步骤或替换刚刚生成的结果。
    • before_agent_callback → types.Content：跳过智能体的主要执行逻辑 (_run_async_impl / _run_live_impl)。返回的Content对象立即被视为智能体在此回合的最终输出。适用于直接处理简单请求或执行访问控制。
    • before_model_callback → LlmResponse：跳过对外部大型语言模型的调用。返回的LlmResponse对象被处理为好像是 LLM 的实际响应。适用于实现输入防护、提示验证或提供缓存响应。
    • before_tool_callback → dict：跳过实际工具函数 (或子智能体) 的执行。返回的dict用作工具调用的结果，通常会传回给 LLM。非常适合验证工具参数、应用策略限制或返回模拟/缓存的工具结果。
    • after_agent_callback → types.Content：替换智能体运行逻辑刚刚产生的Content。
    • after_model_callback → LlmResponse：替换从 LLM 接收到的LlmResponse。适用于清理输出、添加标准免责声明或修改 LLM 的响应结构。
    • after_tool_callback → dict：替换工具返回的dict结果。允许在将工具输出发送回 LLM 之前对其进行后处理或标准化。

概念代码示例 (防护机制)：

此示例演示了使用before_model_callback实现防护机制的常见模式。

from google.adk.agents import LlmAgent
from google.adk.agents.callback_context import CallbackContext
from google.adk.models import LlmResponse, LlmRequest
from google.adk.runners import Runner
from typing import Optional
from google.genai import types 
from google.adk.sessions import InMemorySessionService

GEMINI_2_FLASH="gemini-2.0-flash"

# --- Define the Callback Function ---
def simple_before_model_modifier(
    callback_context: CallbackContext, llm_request: LlmRequest
) -> Optional[LlmResponse]:
    """Inspects/modifies the LLM request or skips the call."""
    agent_name = callback_context.agent_name
    print(f"[Callback] Before model call for agent: {agent_name}")

    # Inspect the last user message in the request contents
    last_user_message = ""
    if llm_request.contents and llm_request.contents[-1].role == 'user':
         if llm_request.contents[-1].parts:
            last_user_message = llm_request.contents[-1].parts[0].text
    print(f"[Callback] Inspecting last user message: '{last_user_message}'")

    # --- Modification Example ---
    # Add a prefix to the system instruction
    original_instruction = llm_request.config.system_instruction or types.Content(role="system", parts=[])
    prefix = "[Modified by Callback] "
    # Ensure system_instruction is Content and parts list exists
    if not isinstance(original_instruction, types.Content):
         # Handle case where it might be a string (though config expects Content)
         original_instruction = types.Content(role="system", parts=[types.Part(text=str(original_instruction))])
    if not original_instruction.parts:
        original_instruction.parts.append(types.Part(text="")) # Add an empty part if none exist

    # Modify the text of the first part
    modified_text = prefix + (original_instruction.parts[0].text or "")
    original_instruction.parts[0].text = modified_text
    llm_request.config.system_instruction = original_instruction
    print(f"[Callback] Modified system instruction to: '{modified_text}'")

    # --- Skip Example ---
    # Check if the last user message contains "BLOCK"
    if "BLOCK" in last_user_message.upper():
        print("[Callback] 'BLOCK' keyword found. Skipping LLM call.")
        # Return an LlmResponse to skip the actual LLM call
        return LlmResponse(
            content=types.Content(
                role="model",
                parts=[types.Part(text="LLM call was blocked by before_model_callback.")],
            )
        )
    else:
        print("[Callback] Proceeding with LLM call.")
        # Return None to allow the (modified) request to go to the LLM
        return None


# Create LlmAgent and Assign Callback
my_llm_agent = LlmAgent(
        name="ModelCallbackAgent",
        model=GEMINI_2_FLASH,
        instruction="You are a helpful assistant.", # Base instruction
        description="An LLM agent demonstrating before_model_callback",
        before_model_callback=simple_before_model_modifier # Assign the function here
)

APP_NAME = "guardrail_app"
USER_ID = "user_1"
SESSION_ID = "session_001"

# Session and Runner
session_service = InMemorySessionService()
session = session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
runner = Runner(agent=my_llm_agent, app_name=APP_NAME, session_service=session_service)


# Agent Interaction
def call_agent(query):
  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)

call_agent("callback example")

通过理解这种返回None与返回特定对象的机制，你可以精确控制智能体的执行路径，使回调成为构建复杂可靠智能体的基本工具。