插件¶
什么是插件？¶
智能体开发套件 (ADK) 中的插件是一个自定义代码模块，可以使用回调钩子在智能体工作流生命周期的各个阶段执行。你使用插件来实现适用于整个智能体工作流的功能。插件的一些典型应用如下：
- 日志记录和跟踪：创建智能体、工具和生成式 AI 模型活动的详细日志，用于调试和性能分析。
- 策略执行：实现安全防护措施，例如检查用户是否有权使用特定工具并在他们没有权限时阻止其执行的函数。
- 监控和指标：收集并导出有关令牌使用量、执行时间和调用次数的指标到监控系统，如 Prometheus 或 Google Cloud Observability （前身为 Stackdriver）。
- 响应缓存：检查之前是否已发出请求，以便你可以返回缓存的响应，跳过昂贵或耗时的 AI 模型或工具调用。
- 请求或响应修改：动态地向 AI 模型提示添加信息或标准化工具输出响应。
注意： 插件不受ADK Web 界面支持。如果你的 ADK 工作流使用插件，你必须在没有 Web 界面的情况下运行你的工作流。
插件如何工作？¶
ADK 插件扩展了 BasePlugin 类并包含一个或多个 callback 方法，指示插件应该在智能体生命周期的哪个位置执行。你通过在你的智能体的 Runner 类中注册插件来将插件集成到智能体中。有关如何在你的智能体应用程序中触发插件以及在哪里触发插件的更多信息，请参阅插件回调钩子。
插件功能建立在回调的基础上，这是 ADK 可扩展架构的关键设计元素。虽然典型的智能体回调配置在单个智能体、单个工具上用于特定任务，但插件在 Runner 上注册一次，其回调全局应用于该运行器管理的每个智能体、工具和 LLM 调用。插件让你可以将相关的回调函数打包在一起以在工作流中使用。这使得插件成为实现跨越整个智能体应用程序功能的理想解决方案。
定义和注册插件¶
本节解释如何定义插件类并将其注册为你智能体工作流的一部分。有关完整的代码示例，请参阅在仓库中的插件基础。
创建插件类¶
首先扩展 BasePlugin 类并添加一个或多个 callback 方法，如以下代码示例所示：
# count_plugin.py
from google.adk.agents.base_agent import BaseAgent
from google.adk.agents.callback_context import CallbackContext
from google.adk.models.llm_request import LlmRequest
from google.adk.plugins.base_plugin import BasePlugin
class CountInvocationPlugin(BasePlugin):
"""一个计算智能体和工具调用次数的自定义插件。"""
def __init__(self) -> None:
"""使用计数器初始化插件。"""
super().__init__(name="count_invocation")
self.agent_count: int = 0
self.tool_count: int = 0
self.llm_request_count: int = 0
async def before_agent_callback(
self, *, agent: BaseAgent, callback_context: CallbackContext
) -> None:
"""计算智能体运行次数。"""
self.agent_count += 1
print(f"[插件] 智能体运行计数：{self.agent_count}")
async def before_model_callback(
self, *, callback_context: CallbackContext, llm_request: LlmRequest
) -> None:
"""计算 LLM 请求次数。"""
self.llm_request_count += 1
print(f"[插件] LLM 请求计数：{self.llm_request_count}")
此示例代码实现了 before_agent_callback 和 before_model_callback 的回调，以在智能体生命周期期间计算这些任务的执行次数。
注册插件类¶
通过在智能体初始化期间作为 Runner 类的一部分注册你的插件类，使用 plugins 参数来集成你的插件类。你可以使用此参数指定多个插件。以下代码示例显示了如何注册前一部分中定义的 CountInvocationPlugin 插件与简单的 ADK 智能体。
from google.adk.runners import InMemoryRunner
from google.adk import Agent
from google.adk.tools.tool_context import ToolContext
from google.genai import types
import asyncio
# 导入插件。
from .count_plugin import CountInvocationPlugin
async def hello_world(tool_context: ToolContext, query: str):
print(f'Hello world: 查询是 [{query}]')
root_agent = Agent(
model='gemini-2.0-flash',
name='hello_world',
description='打印 hello world 和用户查询。',
instruction="""使用 hello_world 工具打印 hello world 和用户查询。
""",
tools=[hello_world],
)
async def main():
"""智能体的主入口点。"""
prompt = 'hello world'
runner = InMemoryRunner(
agent=root_agent,
app_name='test_app_with_plugin',
# 在这里添加你的插件。你可以添加多个插件。
plugins=[CountInvocationPlugin()],
)
# 其余部分与启动常规 ADK 运行器相同。
session = await runner.session_service.create_session(
user_id='user',
app_name='test_app_with_plugin',
)
async for event in runner.run_async(
user_id='user',
session_id=session.id,
new_message=types.Content(
role='user', parts=[types.Part.from_text(text=prompt)]
)
):
print(f'** 从 {event.author} 收到事件')
if __name__ == "__main__":
asyncio.run(main())
使用插件运行智能体¶
像通常一样运行插件。以下显示如何运行命令行：
插件不受 ADK Web 界面支持。 如果你的 ADK 工作流使用插件，你必须在没有 Web 界面的情况下运行你的工作流。
前面描述的智能体的输出应该类似于以下内容：
[插件] 智能体运行计数: 1
[插件] LLM 请求计数: 1
** 从 hello_world 收到事件
Hello world: 查询是 [hello world]
** 从 hello_world 收到事件
[插件] LLM 请求计数: 2
** 从 hello_world 收到事件
有关运行 ADK 智能体的更多信息，请参阅快速开始指南。
使用插件构建工作流¶
插件回调钩子是实现拦截、修改甚至控制智能体执行生命周期逻辑的机制。每个钩子都是你插件类中的一个特定方法，你可以实现它在关键时刻运行代码。根据钩子的返回值，你可以在两种操作模式之间进行选择：
- 观察： 实现没有返回值（
None）的钩子。这种方法适用于日志记录或收集指标等任务，因为它允许智能体工作流继续进行到下一步而不中断。例如，你可以使用插件中的after_tool_callback来记录每个工具的结果以进行调试。 - 干预： 实现钩子并返回值。这种方法会使工作流短路。
Runner停止处理，跳过任何后续插件和原始预期操作，如模型调用，并使用插件回调的返回值作为结果。一个常见的用例是实现before_model_callback来返回缓存的LlmResponse，防止冗余和昂贵的 API 调用。 - 修改： 实现钩子并修改上下文对象。这种方法允许你修改要执行的模块的上下文数据，而不会中断该模块的执行。例如，为模型对象执行添加额外的、标准化的提示文本。
注意： 插件回调函数优先于对象级别实现的回调。这种行为意味着任何插件回调代码都在任何智能体、模型或工具对象回调之前执行。此外，如果插件级别的智能体回调返回任何值，而不是空的（None）响应，则智能体、模型或工具级别的回调不会执行（被跳过）。
插件设计建立了代码执行的层次结构，并将全局关注点与本地智能体逻辑分离。插件是你构建的有状态的模块，如 PerformanceMonitoringPlugin，而回调钩子是该模块中执行的具体函数。这种架构在以下关键方面与标准智能体回调根本不同：
- 范围： 插件钩子是全局的。你在
Runner上注册一次插件，其钩子普遍应用于它管理的每个智能体、模型和工具。相比之下，智能体回调是本地的，在特定的智能体实例上单独配置。 - 执行顺序： 插件有优先级。对于任何给定的事件，插件钩子总是在任何相应的智能体回调之前运行。这种系统行为使插件成为实现跨切功能的正确架构选择，如安全策略、通用缓存和整个应用程序的一致日志记录。
智能体回调和插件¶
如前一部分所述，插件和智能体回调之间有一些功能相似性。下表更详细地比较了插件和智能体回调之间的差异。
| 插件 | 智能体回调 | |
|---|---|---|
| 范围 | 全局：应用于 Runner 中的所有智能体/工具/LLM。 |
本地：仅应用于它们配置的特定智能体实例。 |
| 主要用例 | 横向功能：日志记录、策略、监控、全局缓存。 | 特定智能体逻辑：修改单个智能体的行为或状态。 |
| 配置 | 在 Runner 上配置一次。 |
在每个 BaseAgent 实例上单独配置。 |
| 执行顺序 | 插件回调在智能体回调之前运行。 | 智能体回调在插件回调之后运行。 |
插件回调钩子¶
你通过在插件类中定义的回调函数来定义何时调用插件。当收到用户消息时，在调用 Runner、Agent、Model 或 Tool 之前和之后，对于 Events，以及当 Model 或 Tool 发生错误时，回调都可用。这些回调包括并优先于你在智能体、模型和工具类中定义的任何回调。
下图说明了在智能体工作流期间你可以附加和运行插件功能的回调点：
图 1. 带有插件回调钩子位置的 ADK 智能体工作流程图。
以下部分更详细地描述了插件可用的回调钩子。
用户消息回调¶
用户消息回调（on_user_message_callback）在用户发送消息时发生。on_user_message_callback 是第一个运行的钩子，让你有机会检查或修改初始输入。
- 何时运行： 此回调在
runner.run()之后立即发生，在任何其他处理之前。 - 目的： 检查或修改用户原始输入的第一个机会。
- 流程控制： 返回
types.Content对象以替换用户的原始消息。
以下代码示例显示了此回调的基本语法：
async def on_user_message_callback(
self,
*,
invocation_context: InvocationContext,
user_message: types.Content,
) -> Optional[types.Content]:
运行器开始回调¶
运行器开始回调（before_run_callback）在 Runner 对象获取可能已修改的用户消息并准备执行时发生。before_run_callback 在这里触发，允许在任何智能体逻辑开始之前进行全局设置。
- 何时运行： 在调用
runner.run()之后立即，在任何其他处理之前。 - 目的： 检查或修改用户原始输入的第一个机会。
- 流程控制： 返回
types.Content对象以替换用户的原始消息。
以下代码示例显示了此回调的基本语法：
async def before_run_callback(
self, *, invocation_context: InvocationContext
) -> Optional[types.Content]:
智能体执行回调¶
智能体执行回调（before_agent、after_agent）在 Runner 对象调用智能体时发生。before_agent_callback 在智能体的主要工作开始之前立即运行。主要工作包括智能体处理请求的整个过程，这可能涉及调用模型或工具。在智能体完成所有步骤并准备好结果后，after_agent_callback 运行。
注意： 实现这些回调的插件在智能体级别回调执行之前执行。此外，如果插件级别的智能体回调返回除 None 或空响应之外的任何内容，则智能体级别回调不会执行（被跳过）。
有关作为智能体对象一部分定义的智能体回调的更多信息，请参阅 回调类型。
模型回调¶
模型回调（before_model、after_model、on_model_error）在模型对象执行之前和之后发生。插件功能还支持在发生错误时的回调，详细信息如下：
- 如果智能体需要调用 AI 模型，
before_model_callback首先运行。 - 如果模型调用成功，
after_model_callback接下来运行。 - 如果模型调用因异常而失败，则触发
on_model_error_callback，允许优雅恢复。
注意： 实现 before_model 和 after_model 回调方法的插件在模型级别回调执行之前执行。此外，如果插件级别的模型回调返回除 None 或空响应之外的任何内容，则模型级别回调不会执行（被跳过）。
模型错误回调详情¶
模型对象的错误回调仅由插件功能支持，工作方式如下：
- 何时运行： 在模型调用期间引发异常时。
- 常见用例： 优雅的错误处理、记录特定错误，或返回后备响应，如"AI 服务当前不可用。"
- 流程控制：
- 返回
LlmResponse对象以抑制异常并提供后备结果。 - 返回
None以允许引发原始异常。
注意：如果模型对象的执行返回 LlmResponse，系统恢复执行流程，after_model_callback 将被正常触发。
以下代码示例显示了此回调的基本语法：
async def on_model_error_callback(
self,
*,
callback_context: CallbackContext,
llm_request: LlmRequest,
error: Exception,
) -> Optional[LlmResponse]:
工具回调¶
插件的工具回调（before_tool、after_tool、on_tool_error）在工具执行之前或之后发生，或在发生错误时发生。插件功能还支持在发生错误时的回调，详细信息如下：
- 当智能体执行工具时，
before_tool_callback首先运行。 - 如果工具执行成功，
after_tool_callback接下来运行。 - 如果工具引发异常，则触发
on_tool_error_callback，让你有机会处理失败。如果on_tool_error_callback返回字典，after_tool_callback将被正常触发。
注意： 实现这些回调的插件在工具级别回调执行之前执行。此外，如果插件级别的工具回调返回除 None 或空响应之外的任何内容，则工具级别回调不会执行（被跳过）。
工具错误回调详情¶
工具对象的错误回调仅由插件功能支持，工作方式如下：
- 何时运行： 在工具
run方法执行期间引发异常时。 - 目的： 捕获特定的工具异常（如
APIError），记录失败，并向 LLM 提供用户友好的错误消息。 - 流程控制： 返回
dict以抑制异常，提供后备结果。返回None以允许引发原始异常。
注意：通过返回 dict，这会恢复执行流程，after_tool_callback 将被正常触发。
以下代码示例显示了此回调的基本语法：
async def on_tool_error_callback(
self,
*,
tool: BaseTool,
tool_args: dict[str, Any],
tool_context: ToolContext,
error: Exception,
) -> Optional[dict]:
事件回调¶
事件回调（on_event_callback）在智能体产生诸如文本响应或工具调用结果等输出时发生，它将它们作为 Event 对象产生。on_event_callback 为每个事件触发，让你在将其流式传输到客户端之前修改它。
- 何时运行： 在智能体产生
Event之后但发送给用户之前。智能体的运行可能产生多个事件。 - 目的： 用于修改或丰富事件（例如，添加元数据）或基于特定事件触发副作用。
- 流程控制： 返回
Event对象以替换原始事件。
以下代码示例显示了此回调的基本语法：
async def on_event_callback(
self, *, invocation_context: InvocationContext, event: Event
) -> Optional[Event]:
运行器结束回调¶
运行器结束回调（after_run_callback）在智能体完成其整个过程且所有事件都已处理后发生，Runner 完成其运行。after_run_callback 是最终的钩子，非常适合清理和最终报告。
- 何时运行： 在
Runner完全完成请求执行之后。 - 目的： 理想的全局清理任务，如关闭连接或最终确定日志和指标数据。
- 流程控制： 此回调仅用于拆卸，无法更改最终结果。
以下代码示例显示了此回调的基本语法：