企业官网建设流程全解析

ADK Python 动态节点调度实战：用ctx.run_node把图控制流变成普通 Python 代码

【免费下载链接】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

导读

本文讲解 Google ADK（Agent Development Kit）Python 版 Workflow 中的核心机制——动态节点调度（Dynamic Node Scheduling）。通过Context.run_node(...)，你可以在一个节点内部直接运行另一个节点并拿到它的输出，从而把静态有向图中的"路由边"改写成普通 Python 的循环、条件分支和提前退出。读完本文，你将掌握ctx.run_node的全部参数语义、框架强制执行的运行规则、命令式 Workflow 的编写方式，以及三个最容易踩的"陷阱"。

什么是动态节点调度

在标准的 Workflow 图模型中，节点之间的流转由预先声明的边（edges）决定，例如文档示例中的Workflow(name='root_agent', edges=[('START', orchestrate)])。这种声明式路由适合结构固定的流程，但当"下一步该运行哪个节点"需要在运行时根据结果动态决定时（例如循环重试、条件跳转、提前退出），边的数量会爆炸式增长，难以维护。

动态节点调度把控制权交还给节点内部：await ctx.run_node(...)从节点内部运行另一个节点并返回其输出。正如参考文档所说，它"把图控制流变成普通 Python：循环、条件、提前退出，写起来就是循环、条件、提前退出"。被动态调用的节点会成为当前节点的一个子运行（child run），其路径形如/wf@1/node_a@1，带有自己的 run_id。

从源码实现看，这一机制由 workflow/_dynamic_node_scheduler.py 中的DynamicNodeScheduler承担，它会根据会话事件对子节点做"去重（dedup）/恢复（resume）/全新执行（fresh）"三种处理，保证中断恢复后父节点能拿到子节点的答案。

一个完整的循环示例

参考文档给出了一个"生成标题 → 评判是否与科技相关 → 不合格则重试"的经典循环：

from google.adk import Agent, Context, Event, Workflow from google.adk.workflow import FunctionNode, node from pydantic import BaseModel class Feedback(BaseModel): grade: str generate_headline = Agent( name='generate_headline', instruction='Write a headline about the topic "{topic}".', ) evaluate_headline = Agent( name='evaluate_headline', mode='single_turn', instruction='Grade whether the headline is tech-related.', output_schema=Feedback, ) @node(rerun_on_resume=True) async def orchestrate(ctx: Context, node_input: str) -> str: 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_input=headline) ) if feedback.grade == 'tech-related': yield headline break root_agent = Workflow(name='root_agent', edges=[('START', orchestrate)])

关键点：

• yield Event(state={'topic': node_input})把入参写入会话状态，供子 Agent（generate_headline的 instruction 引用了{topic}）读取；
• await ctx.run_node(generate_headline)不传node_input，标题 Agent 从ctx.state中取topic；
• await ctx.run_node(evaluate_headline, node_input=headline)把上一步的输出作为入参传给评判 Agent；
• 评判结果通过output_schema=Feedback强类型化，用Feedback.model_validate(...)解析；
• grade == 'tech-related'时yield headline并break，循环结束。

注意@node(rerun_on_resume=True)是父节点调用run_node的硬性前提（详见下文"框架强制执行的规则"），示例中两个 Agent 子节点本身由 Agent 封装、默认可重跑，而编排函数必须显式声明该标志。

ctx.run_node参数详解

ctx.run_node的完整签名（见 agents/context.py）：

await ctx.run_node( node, # a function, Agent, BaseTool, or BaseNode node_input=None, *, use_as_output=False, run_id=None, use_sub_branch=False, override_branch=None, )

在源码的_run_node_internal中还可以看到两个内部/进阶参数：override_isolation_scope（覆盖父节点的隔离域）与raise_on_wait（当子节点处于 WAITING 时抛NodeInterruptedError而不是返回None，用于避免父节点被误判为 COMPLETED）。

参数语义细节：

• use_as_output委托：源码在 context.py 中先校验并设置输出委托标记——调用节点的自身输出事件会被抑制，子节点输出（标注output_for）成为父节点输出；
• run_id自动编号：不传时，框架用父节点的_child_run_counters按节点名累加生成"1"、"2"……（见 context.py），这也是自定义 run_id 必须含非数字字符的原因；
• use_sub_branch让动态子节点跑在独立的子分支上，适合"同一个父节点并行调度多个同名节点"的场景，避免事件互相干扰。

框架强制执行的规则

1. 调用节点必须rerun_on_resume=True

调用run_node的节点如果不带rerun_on_resume=True，会立即抛错。源码在 context.py 中直接校验：

if not self._node_rerun_on_resume: raise ValueError( 'A node must have rerun_on_resume=True. 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.' )

原因在文档中写得很清楚：动态调度的子节点可能因用户输入（HITL）等原因中断，父节点要想拿到答案，唯一的方式是从顶部重新执行（re-run from the top）。rerun_on_resume在 workflow/_base_node.py 中定义：True时节点中断后从零重跑；False时中断后立即视为完成、恢复输入被当作节点输出。

2. 显式run_id必须包含非数字字符

自动生成的 run_id 是纯数字（"1"、"2"……），如果自定义 run_id 也是纯数字，就会与自动编号冲突。源码在 context.py 中校验：

if curr_run_id.isdigit() and not skip_run_id_validation: raise ValueError( f'Explicit run_id "{curr_run_id}" for node "{curr_node.name}"' ' must contain non-numeric characters to prevent collision' ' with auto-generated IDs.' )

抛出的ValueError会明确指出违规的 run_id。

3.use_as_output=True每次父执行至多一次

第二次调用会抛出Node {path} already has a use_as_output delegate.（见 context.py），因为父节点只能有一个输出委托。唯一的例外是Workflow自身调用run_node时不受此限制（源码通过isinstance(self.node, Workflow)判断豁免）。

4. 必须直接await调用

不要把它包进asyncio.create_task()：那样子节点将无人监管——错误被静默吞掉，且父节点被中断时子任务不会被取消（文档与 context.py 的 docstring 均强调这一点）。

命令式 Workflow：用 Python 取代路由边

动态节点调度催生了"命令式 Workflow"写法——不再声明条件边，直接用 Python 分支逻辑决定下一个节点：

async def orchestrator(ctx: Context, node_input: str): res_a = await ctx.run_node(step_a, node_input=node_input) if 'success' in res_a: return await ctx.run_node(step_b, node_input=res_a) return await ctx.run_node(step_c, node_input=res_a)

这种写法与 tests/unittests/workflow/test_workflow_dynamic_nodes.py 中大量端到端测试验证的模式一致：父节点（rerun_on_resume=True）调度子节点、接收输出、合并后yield结果，覆盖了全新执行、中断恢复、嵌套动态节点与use_as_output委托等场景。

这种风格下的三个陷阱

陷阱一：普通函数的参数从 state 绑定，而不是从node_input绑定

节点参数绑定的默认值是'state'（见 workflow/_function_node.py 的parameter_binding参数说明），因此run_node(fn, node_input=x)传过去的值，只有通过一个字面命名为node_input的参数才能接收到：

def my_worker(node_input: str): # 必须叫这个名字，否则值到不了 return f'Done: {node_input}'

若函数参数名不叫node_input，框架会尝试从ctx.state中按参数名取值。需要改变绑定方式时，可在创建节点时设置parameter_binding='node_input'。

陷阱二：会调用run_node的子节点也是"父节点"，同样需要rerun_on_resume=True

普通函数默认rerun_on_resume=False（FunctionNode构造函数默认值，见 workflow/_function_node.py），所以如果某个函数内部还要再调度别的节点，必须显式包装：

inner = FunctionNode(func=inner_orchestrator, rerun_on_resume=True)

否则它作为父节点调用run_node时会触发上文规则 1 的ValueError。

陷阱三：生成器不能return值

在使用了yield的节点里，要产出结果必须用yield Event(output=...)；return value在 async 生成器里是语法错误，在 sync 生成器里会被静默忽略。这也是上面示例中orchestrate用yield headline而不是return headline收尾的原因。

底层原理：调度器的三种执行路径

从源码结构看，DynamicNodeScheduler（workflow/_dynamic_node_scheduler.py）对一次ctx.run_node调用按以下三种情况处理：

1. Fresh（全新执行）：会话中没有该节点路径的历史事件，直接创建asyncio.Task运行子节点；
2. Completed（已完成）：历史事件显示此前已执行完毕，通过懒扫描（lazy rehydration）重建状态并直接返回缓存输出，避免重复执行（check_interception决定是快进还是重跑）；
3. Waiting（等待中断）：历史事件显示子节点因中断处于等待，则解析未解决的中断 ID 并传播给父节点，父节点整体重跑后在resume_inputs中拿到恢复输入（如test_workflow_dynamic_nodes.py中ctx.resume_inputs['fc-1']['answer']的用法）。

父节点在中断恢复后从头重跑、再次走到同一个run_node调用时，调度器会基于已记录的 run 状态做出"快进"或"重跑"决策，这正是rerun_on_resume=True之所以是硬性要求的根本原因——父节点的重跑语义由框架保证，子节点才能正确地被去重或恢复。

适用场景小结

• 循环重试：生成-评估-重试（本文示例）；
• 条件路由：根据中间结果在多个节点间选择下一步；
• 提前退出：条件满足时break结束流程；
• 人机协同（HITL）：动态子节点因等待用户输入而中断时，父节点依赖rerun_on_resume与恢复输入完成续跑；
• 并行隔离：配合use_sub_branch在同一父节点下调度多个同名子运行而不互相污染事件。

需要说明的是，ctx.run_node动态调度的节点会作为当前节点的子运行记录在会话事件中，因此它天然具备可恢复、可去重的特性；但这也意味着调用方必须遵守上述框架规则（rerun_on_resume、run_id 约束、直接 await），才能保证中断与恢复语义的正确性。

【免费下载链接】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