企业官网建设流程全解析

ADK Python 回调机制与插件体系深度指南：从八个 Agent 回调到全局 Plugin 钩子

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python

导读

本文聚焦于 ADK Python（Agent Development Kit）的**回调（Callbacks）与插件（Plugins）**机制——这是 ADK 中两个最基础、使用频率最高的扩展点：回调负责拦截并改写单个 Agent 的生命周期，插件则将同一组钩子应用到 App 下的每一个 Agent、工具与模型调用。读完本文，你将掌握八种 Agent 回调的签名与短路语义、Plugin 与 Callback 的执行顺序与优先级规则，以及仓库内置的九个常用插件的适用场景，并能在实际项目中组合它们实现内容审核、工具审计修复、故障降级、上下文裁剪与全局日志等能力。

一、回调与插件的统一契约：None放行，返回值覆盖

ADK 的回调与插件遵循同一个核心契约：

返回None表示"让正常流程继续"；返回一个值则表示"用该值替换原有的行为"。

理解这句话是掌握整个机制的前提。无论是 Agent 级回调（Callback）还是 App 级插件（Plugin），它们的钩子函数签名都以三个基础类型为核心：

from google.adk.agents.callback_context import CallbackContext from google.adk.models.llm_request import LlmRequest from google.adk.models.llm_response import LlmResponse from google.adk.tools import BaseTool, ToolContext

其中：

• CallbackContext是回调函数收到的上下文对象；
• LlmRequest是即将发送给大模型的完整请求（包含消息序列contents）；
• LlmResponse是大模型返回（或由钩子伪造）的响应；
• BaseTool与ToolContext是工具调用相关钩子的参数。

从源码看，这一设计经过了刻意收敛：CallbackContext与ToolContext在当前的代码中都是Context的别名——也就是说，回调上下文与工具上下文在运行时是同一个对象，注释中也保留了ReadonlyContext的向后兼容别名。Context继承自ReadonlyContext，封装了会话（session）、状态（state）、事件动作（event_actions）、节点路径（node_path）等完整运行时信息，因此你在回调里可以读写状态、触发事件动作，甚至调用add_session_to_memory等方法（见 context.py）。

回调与插件的区别

二、八大 Agent 回调：签名、参数与覆盖语义

下表完整列出了可在单个 Agent 上注册的八个回调。每一行都标注了参数列表以及"返回值用于覆盖什么"：

从 llm_agent.py 中可以看到，这八个字段在LlmAgent上的类型均为Optional[...]回调类型（如BeforeModelCallback、OnToolErrorCallback等），默认值为None。其中：

• before_model_callback：在请求发送给模型前调用，可修改llm_request.contents（这也是内置ContextFilterPlugin实现上下文裁剪的挂载点）；
• after_model_callback：在模型响应返回后调用，是记录 token 用量、收集指标的理想位置；
• on_model_error_callback：模型调用抛异常时调用，返回LlmResponse可吞掉错误；
• before_tool_callback：工具执行前调用，可校验/改写参数，任何非None返回（包括空 dict）都会短路，直接作为工具结果返回；
• after_tool_callback：工具执行后调用，返回的 dict 会整体替换原始结果；
• on_tool_error_callback：工具抛异常时调用，返回 dict 作为兜底结果。

同步/异步与列表语义

每一个回调都既可以是同步函数，也可以是异步函数；每一个回调字段既可以传单个 callable，也可以传一个 callable 列表。列表按注册顺序依次执行，停在第一个返回非None值的回调处（即"短路"）。这一语义在BaseAgent与LlmAgent中通过_normalize_callbacks统一归一化为回调列表后执行（见 base_agent.py 与 llm_agent.py）。

三、实战示例：从内容审核到故障降级

3.1 在请求到达模型前拦截（内容审核/安全护栏）

下面的guard回调遍历即将发送给模型的每条消息，一旦发现文本中包含'unsafe'就立即返回一条替代响应，从而跳过真实的模型调用：

def guard( callback_context: CallbackContext, llm_request: LlmRequest ) -> LlmResponse | None: for content in llm_request.contents: for part in content.parts or []: if part.text and 'unsafe' in part.text: return LlmResponse(content=types.ModelContent('I cannot process that.')) return None agent = LlmAgent( name='guarded', model='gemini-2.5-flash', before_model_callback=guard )

注意：函数返回类型为LlmResponse | None，命中条件时返回LlmResponse，否则显式return None放行——这正是第一节契约的直接体现。types.ModelContent来自google.genai。

3.2 只观察不干预（日志观测）

如果只想记录日志而不改变任何行为，务必写清return None：

def log_response( callback_context: CallbackContext, llm_response: LlmResponse ) -> LlmResponse | None: logger.info('model said: %s', llm_response.content) return None

这一模式适合埋点、审计与遥测场景：after_model_callback在模型响应产生后执行，天然是记录 token 消耗与延迟的位置。

3.3 审计与修复工具调用

工具调用是最容易被劫持和出错的环节，可以在调用前后分别挂上"审计"与"修复"两个钩子：

def audit(tool: BaseTool, args: dict, tool_context: ToolContext) -> dict | None: logger.info('calling %s with %s', tool.name, args) return None def repair( tool: BaseTool, args: dict, tool_context: ToolContext, tool_response: dict ) -> dict | None: if 'error' in tool_response: return {'result': 'Tool execution failed, please try again.'} return None agent = LlmAgent( name='audited', model='gemini-2.5-flash', tools=[my_tool], before_tool_callback=audit, after_tool_callback=repair, )

audit在工具执行前记录tool.name与args；repair在工具执行后检查返回 dict，若其中包含'error'键，就用一条友好的{'result': ...}替换原始结果，避免错误内容被直接喂回模型。

3.4 模型故障优雅降级

当上游模型不可用时，用on_model_error_callback返回一条静态兜底响应，而不是让异常直接抛给调用方：

def handle_model_error( callback_context: CallbackContext, llm_request: LlmRequest, error: Exception, ) -> LlmResponse | None: return LlmResponse(content=types.ModelContent('Service unavailable.')) agent = LlmAgent( name='resilient', model='gemini-2.5-flash', on_model_error_callback=handle_model_error, )

四、插件：App 级的全局钩子

4.1 编写自定义插件

插件是把同一组钩子应用到每一个Agent、工具与模型调用上的机制，并额外提供一批只能在 App 作用域下才有意义的钩子。与 Agent 回调不同，插件的所有钩子都是 async 且 keyword-only（必须使用关键字参数调用）。

from google.adk.plugins.base_plugin import BasePlugin class MyPlugin(BasePlugin): def __init__(self): super().__init__(name='my_plugin') async def before_agent_callback(self, *, agent, callback_context): return None async def before_model_callback(self, *, callback_context, llm_request): return None

从 base_plugin.py 的类文档可以看到：插件最适合实现日志、监控、缓存、请求/响应改写这类横切关注点；每个插件应当实现一个回调方法一次，不应重复实现同一回调。

4.2 App 级专属钩子（6 个）

除 Agent 级 8 个钩子外，BasePlugin还额外提供了：

此外BasePlugin还定义了close()方法，在 Runner 关闭时被调用，用于释放网络连接等资源。其中on_agent_error_callback与on_run_error_callback是通知型回调：从 plugin_manager.py 的实现可以看到，它们走的是_run_notification_callbacks路径，会尽力通知所有插件，单个插件抛错只记录日志不影响其余插件。

4.3 在 App 上注册插件

插件挂在App的plugins参数上（见 app.py，该字段为list[BasePlugin]，默认空列表）：

from google.adk.apps import App from google.adk.plugins.context_filter_plugin import ContextFilterPlugin app = App( name='my_app', root_agent=root_agent, plugins=[ContextFilterPlugin(num_invocations_to_keep=3)], )

4.4 插件与回调的执行顺序与传播规则

base_plugin.py 的类文档明确了三条关键规则，理解它们才能避免踩坑：

1. 执行顺序：插件按注册顺序执行；插件与 Agent 回调顺序执行，且插件优先于 Agent 回调；
2. 短路规则：当某个插件的回调返回非None值，会短路掉所有剩余插件与 Agent 回调；
3. 变更传播：插件与 Agent 回调都可以修改输入参数（Agent 输入、工具输入、LLM 请求/响应），修改结果对链路上的下一个回调可见——例如某个插件在before_tool_callback中改写了工具参数，改写后的参数会传给下一个插件乃至后续的 Agent 回调（若未被短路）。

这一"早期退出（early exit）"策略在 plugin_manager.py 的_run_callbacks中实现：依次await每个插件的同名回调方法，遇到第一个非None返回值立即返回并停止后续执行；插件抛出的未处理异常会被包装为RuntimeError并链式抛出。此外PluginManager.register_plugin会拒绝同名插件重复注册（抛出ValueError），close()为每个插件设置了默认 5 秒的超时（见 plugin_manager.py）。

五、内置插件速查表

仓库在google.adk.plugins下提供了九个开箱即用的插件（模块目录见 plugins/）：

以ContextFilterPlugin为例看插件落地方式

ContextFilterPlugin是理解"插件如何工作"的最佳范本。它的核心逻辑全部挂载在before_model_callback上（见 context_filter_plugin.py）：

• 构造参数：num_invocations_to_keep（保留最近 N 次 invocation，一次 invocation 以一条或多条连续用户消息开始，直到下一条用户消息为止）、custom_filter（自定义过滤函数）、remove_amount（上下文超限时一次移除的 invocation 数量，默认 1，且要求 ≥ 1）；
• 实现细节：遍历llm_request.contents定位 invocation 起点，计算需要截断的位置；截断时通过_adjust_split_index_to_avoid_orphaned_function_responses向前调整切分点，避免"只保留 function_response 却丢掉与其配对的 function_call"导致的孤儿响应问题（见 context_filter_plugin.py）；
• 结束后返回None，表示不替换请求，只是原地修改llm_request.contents——这恰好演示了"修改输入参数并向下游传播"的插件语义。

这个例子也再次印证：插件与回调共用同一套钩子签名与None/非None契约，只是作用域从"单个 Agent"扩大到了"整个 App"。

六、小结：如何选择 Callback 还是 Plugin

实践建议：

• 回调/插件内需要"放行"时，务必显式return None，避免隐式返回None之外的意外值造成短路；
• 多个回调按列表组合时，记住第一个返回非None值的回调会吞掉后续回调；
• 工具相关回调的短路判定是"非None"，因此返回空 dict{}同样会短路，可作为"假装工具成功"的技巧（见 llm_agent.py 中对before_tool_callback/after_tool_callback的注释）；
• 想深入验证行为，可阅读 base_agent.py 中_handle_before_agent_callback/_handle_after_agent_callback的完整实现，以及单元测试tests/unittests/agents/下与回调、插件相关的测试用例。

通过本文的八种回调与六类 App 级插件钩子，你可以在不修改框架源码的前提下，为 Agent 应用注入审核、观测、容错、上下文管理与分析能力——这正是 ADK 以"code-first"方式提供灵活性与控制力的核心所在。

【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python