ADK BaseAgent 深度解析:Agent 生命周期、回调体系与自定义扩展点 ADK BaseAgent 深度解析Agent 生命周期、回调体系与自定义扩展点【免费下载链接】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-pythonBaseAgent是 Google Agent Development KitADK中所有 Agent 的抽象基类它把工作流节点与带生命周期的 Agent两个概念缝合在一起既继承了BaseNode的节点执行契约又在其上叠加了回调、错误处理、调用观测与 Agent 树管理。本文以仓库文档 .agents/skills/adk-architecture/references/interface-base-agent.md 为主体结合 base_agent.py 源码与 test_base_agent.py 测试用例系统讲解 BaseAgent 的执行链路、可扩展点、关键属性与回调语义。读完本文你将掌握如何自定义一个 Agent、理解为什么不能覆写_run_impl、以及回调与作者归属机制在底层是如何工作的。从 BaseNode 到 BaseAgentAgent 即节点BaseAgent 的类声明位于 base_agent.pyclass BaseAgent(BaseNode, abc.ABC): Base class for all agents in Agent Development Kit.它直接继承工作流运行时中最基本的执行单元BaseNode定义于 workflow/_base_node.py。这意味着一个 Agent 本质上就是一个节点它可以作为独立 Agent 挂在 Runner 下运行也可以作为子节点嵌入 Workflow 图由工作流调度器驱动执行。BaseNode定义了两方法契约详见 interface-base-node.mdrun()— 被标记为final负责把子类_run_impl的每次 yield 规范化为统一的事件流永远不应覆写_run_impl()— 扩展点子类在此实现自己的业务逻辑以异步生成器形式。run()对子类每次 yield 的值应用以下规范化规则yield 的值处理方式None直接跳过Event原样透传RequestInput转换为中断 Event用于 HITL 人工介入其他任意值包装为Event(outputvalue)因此所有内置 AgentLlmAgent、SequentialAgent、LoopAgent、ParallelAgent都天然具备节点的输出、消息、路由语义可以被任意组合进 Workflow 图中。这正是 ADK一切皆节点设计哲学的体现。理解 Agent 生命周期一条不可覆写的桥BaseAgent 的价值在于它把节点执行与Agent 生命周期桥接起来。原文档给出了这条关键调用链Workflow calls node.run() (BaseNode, final) └─ BaseAgent._run_impl (bridge — do not override) └─ BaseAgent.run_async (callbacks, instrumentation, error handling) └─ your _run_async_impl ← override point对应到源码run()继承自BaseNodefinalWorkflow 调度节点时调用的入口BaseAgent._run_impl()base_agent.py标注override作为桥把节点执行转发给run_async并在转发过程中逐事件保留作者信息BaseAgent.run_async()base_agent.pyAgent 生命周期的真正载体——创建 InvocationContext、执行 before/after 回调、记录调用指标_instrumentation.record_agent_invocation、兜底错误回调_run_async_impl(ctx)你的覆写点Agent 的实际逻辑。为什么绝对不能覆写_run_implBaseAgent已经用override覆写了_run_impl作为桥接层。如果子类自行覆写_run_impl会静默丢弃 before/after 回调、错误回调与调用指标——这些能力全部集中在run_async中绕开桥就等于绕开了整个生命周期体系。原文档明确强调Replacing it silently drops all of that.唯一的例外LlmAgentLlmAgent是证明规则的例外它也覆写了_run_implllm_agent.py目的是让 Agent 以工作流节点身份运行时走一个专用的节点包装器run_llm_agent_as_node实现于 workflow/_llm_agent_wrapper.py以适配 LLM Agent 特有的流式输出与事件语义。这是框架内部实现细节自定义 Agent 不应仿效。唯一的扩展点_run_async_impl 与 _run_live_implBaseAgent 把文本对话与实时音视频对话分成两个独立入口扩展点用途签名_run_async_impl(ctx)文本对话text conversationAsyncGenerator[Event, None]接收InvocationContext_run_live_impl(ctx)实时音视频对话live audio/videoAsyncGenerator[Event, None]接收InvocationContext两者的基类默认实现都直接raise NotImplementedErrorbase_agent.py即任何具体 Agent 必须至少实现其中之一。值得强调的是所有内置复合 Agent——LlmAgent、SequentialAgent、LoopAgent、ParallelAgent——都只实现这两个方法不额外触碰_run_impl这从侧面印证了扩展点只有两个的结论。对应入口方法run_async(parent_context)文本会话入口围绕_run_async_impl执行回调、错误处理与调用观测非finalrun_live(parent_context)音视频会话入口标记为final因此只能通过覆写_run_live_impl扩展。自定义一个最小 Agent参照测试中的_TestingAgenttest_base_agent.py自定义 Agent 只需继承 BaseAgent 并实现两个扩展点from collections.abc import AsyncGenerator from google.genai import types from google.adk.agents.base_agent import BaseAgent from google.adk.agents.invocation_context import InvocationContext from google.adk.events.event import Event class HelloAgent(BaseAgent): override async def _run_async_impl( self, ctx: InvocationContext ) - AsyncGenerator[Event, None]: yield Event( authorself.name, branchctx.branch, invocation_idctx.invocation_id, contenttypes.Content(parts[types.Part(textHello, world!)]), ) override async def _run_live_impl( self, ctx: InvocationContext ) - AsyncGenerator[Event, None]: yield Event( authorself.name, invocation_idctx.invocation_id, branchctx.branch, contenttypes.Content(parts[types.Part(textHello, live!)]), )如果子类只实现其中一个而另一个保持未实现由于BaseAgent通过abc.ABC暴露了ABCMeta该类仍是抽象类实例化会直接失败测试test_abstract_subclass_cannot_be_instantiated验证了这一点见 test_base_agent.py。调用未实现的扩展点则会抛出NotImplementedError。核心属性配置详解BaseAgent 暴露的关键配置属性如下每个都有源码级校验逻辑name必填且受三重约束name是 BaseAgent 唯一必填字段Pydantic 必填项其校验器validate_namebase_agent.py强制执行三条规则必须是合法的 Python 标识符——以字母或下划线开头只能包含字母、数字、下划线否则抛ValueErrornot an identifier这类含空格的名字会被拒绝对应测试test_invalid_agent_name必须是合法标识符且在整个 Agent 树内唯一不能是user——该名称保留给终端用户输入。注意校验是大小写敏感的User是允许的对应测试 test_user_agent_name_validation。description模型委派决策依据description是 Agent 能力的自然语言描述默认空字符串模型在决定是否把控制权委派给某个子 Agent 时依据它做判断。原文档与源码都建议用一句话描述简明扼要即可One-line description is enough and preferred.。sub_agents 与 parent_agent分层委派树sub_agents: list[BaseAgent]默认空列表表示子 Agentparent_agent字段在实例化时由model_post_init自动回填__set_parent_agent_for_sub_agentsbase_agent.py无需手动设置重复名称在树内会被拒绝validate_sub_agents_unique_names校验器会扫描同一层子 Agent 的重名并记录 warningbase_agent.py测试test_validate_sub_agents_unique_names_*系列覆盖了单个、多个、三个重复及无重复场景一个 Agent 只能被添加为子 Agent 一次若某个 Agent 已有parent_agent再被另一个 Agent 引用会抛ValueError测试test_set_parent_agent_for_sub_agent_twice。若需要同一配置的 Agent 出现两次应创建两个配置相同但name不同的实例。before/after 回调生命周期钩子before_agent_callback与after_agent_callback都可以是单个可调用对象或可调用对象列表二者语义不同详见下文回调章节。源码中的规范形态通过只读属性canonical_before_agent_callbacks/canonical_after_agent_callbacks暴露内部调用_normalize_callbacks归一化为列表base_agent.py。生命周期回调before/after 钩子的完整语义回调是 BaseAgent 生命周期中最核心的扩展机制其执行逻辑集中在run_async/run_live与_handle_before_agent_callback/_handle_after_agent_callbackbase_agent.py。before_agent_callback短路优先在_run_async_impl执行之前调用。关键语义回调接收一个CallbackContext参数支持关键字或位置绑定多个回调按声明顺序依次执行直到某个回调返回真值即停止_run_callbacks(callbacks, _stop_on_truthy, ...)停止条件来自 utils/_callback_pipeline.py若返回真值一个types.ContentAgent 本轮运行被跳过该内容作为事件直接返回给用户并置ctx.end_invocation True测试test_run_async_before_agent_callback_bypass_agent验证了短路行为若回调只改动了状态callback_context.state.has_delta()也会产出对应的事件。after_agent_callback追加回复在_run_async_impl正常结束后调用。语义与 before 不同同样按列表顺序执行、遇真值停止返回真值时不会跳过任何内容而是把该内容作为额外的 Agent 回复事件追加到事件历史末尾对应测试test_run_async_after_agent_callback_append_reply。与插件体系的协作两个回调处理函数都会先询问插件管理器plugin_manager.run_before_agent_callback/run_after_agent_callback若插件提供了覆盖内容则跳过 canonical 回调。这让回调既可以通过属性注入也可以通过插件机制注入两条路径互不干扰。错误回调通知式、尽力而为_handle_agent_error_callbackbase_agent.py负责把逃逸异常通知所有插件的on_agent_error_callback。它遵循三条原则仅通知原始异常总是会被重新抛出run_async中except后raise回调自身抛出的异常只会被记录日志并抑制绝不允许掩盖原始错误通知是尽力而为的回调实现缺失如测试替身也不会影响主流程。调用观测run_async与run_live都会在_run外层包裹_instrumentation.record_agent_invocation(ctx, self)为每次 Agent 调用记录 OpenTelemetry 观测数据同时通过_with_caller_context在每次 yield 时把调用方的 context 上下文重新挂回保证遥测与日志的上下文正确性。作者归属机制事件不被父节点覆盖当 Agent 作为工作流节点运行时存在一个容易被忽略的细节——事件作者归属。BaseAgent._run_implbase_agent.py在转发每个事件时执行async for event in self.run_async(parent_contextctx.get_invocation_context()): # Preserve author by setting it in context for NodeRunner if event.author: ctx.event_author event.author ...它把每个事件的author复制到ctx.event_author这样外层的NodeRunner就不会用父工作流或父 Agent的名字覆盖它。事件因此始终归属于真正产生它的那个 Agent而不是机械地打上父节点的标签。LlmAgent覆写的_run_impl也做了完全相同的事件作者保留处理llm_agent.py可见这是框架层的一致约定。Agent 树操作查找、根节点与克隆BaseAgent 还提供了一套树操作工具均有对应测试覆盖方法/属性行为测试find_agent(name)在自身及所有后代中查找 Agent优先返回自身即使后代存在同名节点test_find_agent_prefers_self_over_same_named_descendantfind_sub_agent(name)只在后代中查找test_find_sub_agentroot_agent沿parent_agent链向上走到树根test_root_agentclone(updateNone)深拷贝 Agent含递归克隆子 Agent新实例与父节点解耦以自身方法绑定的回调会被重绑到克隆体test_agent_clone.pytests/unittests/agents/test_agent_clone.py需要注意clone()不允许在update中修改parent_agent也不允许更新类中不存在的字段否则都会抛ValueError。测试验证行为有据可查本文描述的所有行为都可以在 tests/unittests/agents/test_base_agent.py 中找到对应用例例如名称校验test_invalid_agent_name、test_user_agent_name_validation抽象约束test_agent_classes_are_abstract、test_abstract_subclass_cannot_be_instantiatedbefore 回调短路test_run_async_before_agent_callback_bypass_agentafter 回调追加test_run_async_after_agent_callback_append_reply回调链test_before_agent_callbacks_chain、test_after_agent_callbacks_chain验证多回调按声明顺序执行、遇真值停止canonical 归一化test_canonical_agent_callbacks_unset_resolves_to_empty_list等三例树结构test_find_agent、test_root_agent、test_validate_sub_agents_unique_names_*系列。相关文档索引interface-base-agent.md本文主体的官方精简版说明interface-agent.mdAgent即LlmAgent的类型别名与BaseAgent的区别及入口方法矩阵interface-base-node.mdBaseNode节点契约、输出/流式/状态/路由规则与配置参考表base_agent.pyBaseAgent 完整实现校验器、回调管线、克隆与树操作llm_agent.pyLlmAgent对_run_async_impl/_run_live_impl/_run_impl的覆写示例workflow/_base_node.py节点基类与run()规范化实现。小结BaseAgent 是 ADK 中节点执行与Agent 生命周期的汇合点。记住三条铁律——只覆写_run_async_impl与_run_live_impl、绝不碰_run_impl、name必须全局唯一且不能是user——你就能安全地构建出自定义 Agent并完整保留回调、错误处理与观测能力。【免费下载链接】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创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考