ADK Python 工作流动态节点实战：用 ctx.run_node 实现运行时决定的执行路径

ADK Python 工作流动态节点实战用 ctx.run_node 实现运行时决定的执行路径【免费下载链接】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本文围绕 ADKAgent Development KitPython 版的 Workflow 引擎讲解动态节点执行Dynamic Node Execution机制当工作流的执行路径或节点运行次数无法在静态edges中确定时如何用ctx.run_node配合原生 Python 控制流如while循环在运行时动态调度节点。基于contributing/samples/workflows/dynamic_nodes官方示例完整走读实现并结合src/google/adk/workflow与src/google/adk/agents/context.py的源码说明其底层调度、可恢复执行rerun_on_resume原理与事件轨迹验证方法。读完后你将掌握如何声明动态编排节点、run_node各参数的实际影响、动态节点被中断后恢复的机制以及如何用会话事件 JSON 验证多轮动态执行。为什么需要动态节点执行在 ADK Workflow 中标准的执行路径由edges静态定义哪些节点先跑、谁触发谁、走哪个分支都在建图时写死。但真实场景中存在一类需求——具体的执行节点集合、或者某个节点要运行几次只有在运行时才能确定例如循环生成内容直到通过 LLM 评审评审轮数未知根据用户输入的数量动态扇出子任务fan-out 数量未知动态拼接多个 Agent 的调用顺序。dynamic_nodes示例处理的正是第一种动态循环场景。一个名为orchestrate的 Python 节点充当驱动器driver在while True:循环中先执行generate_headlineAgent 基于给定主题生成头条再执行evaluate_headlineAgent 对其评分。若评分为tech-related则返回该头条若为unrelated则把评审反馈写回 state循环继续。值得注意的是这个示例是标准loop示例的重写版本它没有使用复杂的图边路由例如在edges里配置条件路由函数而是完全依靠原生 Python 控制流while循环加异步ctx.run_node调用来实现。执行拓扑如下图中只有START → orchestrate一条静态边其余执行路径全部是orchestrate在运行时通过ctx.run_node动态调度出来的。示例可运行的输入见 READMEflowerquantum mechanicsrenewable energy完整示例代码解读下面给出 agent.py 的完整实现它是可独立运行的最小示例from typing import AsyncGenerator from typing import Literal from google.adk import Agent from google.adk import Context from google.adk import Event from google.adk import Workflow from google.adk.workflow import node from pydantic import BaseModel from pydantic import Field class Feedback(BaseModel): grade: Literal[tech-related, unrelated] Field( description( Decide if the headline is related to technology or software engineering. ), ) feedback: str Field( description( If the headline is unrelated to technology, provide feedback on how to make it more tech-focused. ), ) generate_headline Agent( namegenerate_headline, instruction Write a headline about the topic {topic}. If feedback is provided, take it into account. The feedback: {feedback?} , ) evaluate_headline Agent( nameevaluate_headline, instruction Grade whether the headline is related to technology or software engineering. , output_schemaFeedback, output_keyfeedback, ) node(rerun_on_resumeTrue) async def orchestrate( ctx: Context, node_input: str ) - AsyncGenerator[Event | str, None]: yield Event(state{topic: node_input}) while True: headline await ctx.run_node(generate_headline) feedback Feedback.model_validate( await ctx.run_node(evaluate_headline, node_inputheadline) ) if feedback.grade tech-related: yield headline break root_agent Workflow( nameroot_agent, edges[(START, orchestrate)], )代码分为四个关键部分结构化评审输出Feedback用 Pydantic 模型约束评审结果grade字段取值限定为Literal[tech-related, unrelated]保证循环终止条件可被可靠判断。生成 Agentgenerate_headlineinstruction 中用{topic}从 state 读取主题{feedback?}表示可选字段——首轮没有反馈时为空后续轮次自动带上上一轮的改进建议。评审 Agentevaluate_headlineoutput_schemaFeedback使其输出被解析为结构化对象output_keyfeedback把结果写入会话 state 的feedback键从而成为下一轮generate_headline可用的上下文。编排节点orchestrate 根 Workfloworchestrate是唯一的静态节点Workflow只声明了(START, orchestrate)一条边其余全部动态发生。关键机制一node(rerun_on_resumeTrue)启用可恢复执行README 的第一步要点要让 Python 节点使用ctx.run_node必须声明为node(rerun_on_resumeTrue)。它告诉引擎如果任何被动态调度的子节点被中断例如等待 human-in-the-loop 输入工作流引擎应当暂停并在恢复时重新运行re-run编排节点让编排节点从子节点拿到续跑后的结果。这不是可选优化而是硬性要求。在源码 src/google/adk/workflow/_dynamic_node_executor.py 的run_node_internal入口处就有显式校验if not ctx._node_rerun_on_resume: raise ValueError( A node must have rerun_on_resumeTrue. Reason is that dynamically scheduled nodes might be interrupted, and the workflow wakes-up/re-runs the parent node, so it can get the child node response. )即若父节点没有开启rerun_on_resume却调用ctx.run_node会在运行期直接抛出ValueError。这个设计的原理是动态子节点可能长时间挂起HITL、长时工具编排协程不能无限等待引擎选择暂停父节点 → 子节点完成/恢复 → 重跑父节点的恢复模型父节点重跑时从已完成子节点拿到缓存输出去重从而安全地继续循环。node装饰器本身还支持更多参数完整签名见 src/google/adk/workflow/_node.pyname覆盖节点名、rerun_on_resume、retry_config重试策略、timeout、parallel_worker/max_parallel_workers并行工作器、auth_config运行前请求用户认证要求配合rerun_on_resumeTrue、parameter_bindingstate默认从ctx.state绑定函数参数node_input则从node_input绑定并按函数签名推断输入/输出 schema用于节点充当 Agent 工具的场景。动态节点场景中与本例直接相关的是rerun_on_resume。关键机制二ctx.run_node从上下文运行节点README 的第二步要点向 Python 节点注入ctx: Contextawait ctx.run_node(node_to_run)运行目标节点返回值就是该次执行的最终输出在循环的每次迭代之间还可以yield Event(...)更新 state。示例中await ctx.run_node(generate_headline)—— 不传node_inputAgent 直接从 state 读取{topic}/{feedback?}生成头条返回值即头条文本await ctx.run_node(evaluate_headline, node_inputheadline)—— 把本轮头条作为node_input显式传入评审 Agent返回值经Feedback.model_validate解析为结构化对象yield Event(state{topic: node_input})—— 在第一轮循环前把用户输入写入 state让下游 Agent 的 instruction 模板能取到topic。Context.run_node的公开 API 定义在 src/google/adk/agents/context.py完整参数与语义如下均可用于动态节点编排参数说明node要执行的节点BaseNode实例或任何可被构建成节点的可调用对象/Agentnode_input传给该节点的输入数据默认Noneuse_as_output若为True子节点输出直接作为调用节点的输出调用节点自身的输出事件被抑制以避免重复run_id自定义本次动态执行的 run ID便于跨运行关联事件不提供则自动生成use_sub_branch若为True子节点在子分支中执行隔离其 state 与事件override_branch覆盖父节点默认分支override_isolation_scope覆盖父节点默认的隔离作用域raise_on_wait若为True子节点处于 WAITING 时抛出NodeInterruptedError而非返回None该方法文档中有一条重要的使用约束必须直接await它不要用asyncio.create_task()包裹——那样任务将失去监督错误会被静默吞掉父节点被中断如 HITL时任务也不会被取消。本示例中两次调用都是直接await符合这一要求。底层调度动态节点如何被引擎接管从源码结构看ctx.run_node最终进入 run_node_internal其核心逻辑分三种路径前置校验见上文强制父节点rerun_on_resumeTrue。双模式执行Workflow 模式当父节点运行在工作流图内ctx._workflow_scheduler存在执行委托给工作流调度器ScheduleDynamicNode由调度器处理图依赖与状态。这里还有一个细节校验——调用方显式传入的run_id如果是纯数字会被拒绝must contain non-numeric characters to prevent collision with auto-generated IDs因为数字 run_id 由调度器顺序分配显式数字会与之冲突。Standalone 模式节点独立于工作流运行时直接通过NodeRunner执行。Agent Transfer 循环run_node_internal用一个while True循环处理节点执行中又请求转移到另一个 Agent的情况例如子 Agent A 把执行转给 Agent B循环内更新目标节点与父上下文指针并继续若子节点报错则抛DynamicNodeFailError若子节点被中断存在interrupt_ids则把中断 ID 传播回父节点上下文并抛NodeInterruptedError让上层 NodeRunner 把父节点记录为等待状态而非误判完成——这正是rerun_on_resume恢复模型能工作的前提。也就是说示例中看起来只是两行await ctx.run_node(...)的背后引擎完成了 run_id 分配、分支/作用域隔离、事件溯源子节点输出事件带上output_for指向父节点路径、中断传播与恢复重放这一整套机制。用事件轨迹验证多轮动态执行示例目录附带了一个真实的会话事件文件 tests/flower.json记录了输入flower时的一次完整运行可以把它当作动态节点如何落事件的参照e-2orchestrate产生stateDelta: {topic: flower}路径为root_agent1/orchestrate1e-3第一次动态调度generate_headline输出A World of Petals其nodeInfo.path为root_agent1/orchestrate1/generate_headline1且output_for指回父节点路径——即子节点输出归属于父节点的体现e-4第一次evaluate_headline输出结构化 JSONgrade为unrelatedstateDelta写入feedback含改进建议e-5/e-6第二轮循环generate_headline的路径变为.../generate_headline2run 序号递增结合反馈产出**Digital Petals: Engineering AI-Enhanced Blooms**evaluate_headline2评出tech-relatede-7最终输出即该头条output_for同时指向orchestrate与根root_agent。这段轨迹印证了三个机制state 通过Event(state...)在轮次间传递feedback每轮被覆写动态子节点的 path/run_id 由引擎自动生成并随轮次递增循环的终止yield headline后break最终把父节点输出冒泡为整个 Workflow 的输出。测试覆盖与延伸阅读仓库的单元测试 tests/unittests/workflow/test_workflow_dynamic_nodes.py 系统地覆盖了动态节点调度的三种核心情形——全新执行无历史事件、已完成去重重跑时直接返回缓存输出、中断后恢复携带resume_inputs重跑以及多动态节点、嵌套动态节点、use_as_output输出委托等边界情况。此外还有 test_dynamic_node_executor.py、test_dynamic_node_scheduler.py、test_dynamic_use_as_output.py 等针对执行器与调度器的独立测试可作为理解上述底层机制的对照材料。实践要点小结静态边最少化动态节点场景下Workflow只需声明(START, 编排节点)一条边其余拓扑由 Python 控制流表达可读性和灵活性优于在edges中写条件路由函数必须rerun_on_resumeTrue这是调用ctx.run_node的硬性前提用于支撑子节点中断 → 父节点重跑的恢复模型源码中有显式校验直接await ctx.run_node不要用asyncio.create_task包裹否则错误与中断传播都会失效用 state 做轮次间的通信总线通过yield Event(state...)写入topic、利用output_key写回feedback让多轮循环中的 Agent 自然获得上一轮上下文用output_schema固化循环终止条件结构化的grade字段让while循环的退出判断可靠、可测试用事件 JSON 验证行为动态子节点的 path序号递增与output_for归属关系是核对多轮执行是否正确落地的直接证据。适用前提与限制本机制面向 ADK 的 Workflow 引擎google.adk.Workflow/node需要可被build_node构建的节点对象BaseAgent、BaseTool、BaseNode 或可调用对象显式传入纯数字run_id会被拒绝use_as_output委托在非 Workflow 父节点下每个父节点只能设置一次。以上行为均以当前仓库源码为准。【免费下载链接】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），仅供参考