尧图网站建设 
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 ADKAgent Development KitPython 版 Workflow 中的核心机制——动态节点调度Dynamic Node Scheduling。通过Context.run_node(...)你可以在一个节点内部直接运行另一个节点并拿到它的输出从而把静态有向图中的路由边改写成普通 Python 的循环、条件分支和提前退出。读完本文你将掌握ctx.run_node的全部参数语义、框架强制执行的运行规则、命令式 Workflow 的编写方式以及三个最容易踩的陷阱。什么是动态节点调度在标准的 Workflow 图模型中节点之间的流转由预先声明的边edges决定例如文档示例中的Workflow(nameroot_agent, edges[(START, orchestrate)])。这种声明式路由适合结构固定的流程但当下一步该运行哪个节点需要在运行时根据结果动态决定时例如循环重试、条件跳转、提前退出边的数量会爆炸式增长难以维护。动态节点调度把控制权交还给节点内部await ctx.run_node(...)从节点内部运行另一个节点并返回其输出。正如参考文档所说它把图控制流变成普通 Python循环、条件、提前退出写起来就是循环、条件、提前退出。被动态调用的节点会成为当前节点的一个子运行child run其路径形如/wf1/node_a1带有自己的 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( namegenerate_headline, instructionWrite a headline about the topic {topic}., ) evaluate_headline Agent( nameevaluate_headline, modesingle_turn, instructionGrade whether the headline is tech-related., output_schemaFeedback, ) node(rerun_on_resumeTrue) 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_inputheadline) ) if feedback.grade tech-related: yield headline break root_agent Workflow(nameroot_agent, edges[(START, orchestrate)])关键点yield Event(state{topic: node_input})把入参写入会话状态供子 Agentgenerate_headline的 instruction 引用了{topic}读取await ctx.run_node(generate_headline)不传node_input标题 Agent 从ctx.state中取topicawait ctx.run_node(evaluate_headline, node_inputheadline)把上一步的输出作为入参传给评判 Agent评判结果通过output_schemaFeedback强类型化用Feedback.model_validate(...)解析grade tech-related时yield headline并break循环结束。注意node(rerun_on_resumeTrue)是父节点调用run_node的硬性前提详见下文框架强制执行的规则示例中两个 Agent 子节点本身由 Agent 封装、默认可重跑而编排函数必须显式声明该标志。ctx.run_node参数详解ctx.run_node的完整签名见 agents/context.pyawait ctx.run_node( node, # a function, Agent, BaseTool, or BaseNode node_inputNone, *, use_as_outputFalse, run_idNone, use_sub_branchFalse, override_branchNone, )参数作用node要动态运行的节点可以是普通函数、Agent、BaseTool或任何BaseNode框架内部通过 workflow/utils/_workflow_graph_utils.py 的build_node将其规范化为节点node_input传给被调节点的输入默认Noneuse_as_output为True时子节点的输出成为父节点的输出父节点自身的输出事件被抑制避免重复run_id自定义这次执行的 run_id替代自动编号use_sub_branch为True时在分支路径上追加node_namerun_id把事件与兄弟运行隔离override_branch使用指定的 branch 而不是父节点的 branch在源码的_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_resumeTrue调用run_node的节点如果不带rerun_on_resumeTrue会立即抛错。源码在 context.py 中直接校验if not self._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. )原因在文档中写得很清楚动态调度的子节点可能因用户输入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( fExplicit 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_outputTrue每次父执行至多一次第二次调用会抛出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_inputnode_input) if success in res_a: return await ctx.run_node(step_b, node_inputres_a) return await ctx.run_node(step_c, node_inputres_a)这种写法与 tests/unittests/workflow/test_workflow_dynamic_nodes.py 中大量端到端测试验证的模式一致父节点rerun_on_resumeTrue调度子节点、接收输出、合并后yield结果覆盖了全新执行、中断恢复、嵌套动态节点与use_as_output委托等场景。这种风格下的三个陷阱陷阱一普通函数的参数从 state 绑定而不是从node_input绑定节点参数绑定的默认值是state见 workflow/_function_node.py 的parameter_binding参数说明因此run_node(fn, node_inputx)传过去的值只有通过一个字面命名为node_input的参数才能接收到def my_worker(node_input: str): # 必须叫这个名字否则值到不了 return fDone: {node_input}若函数参数名不叫node_input框架会尝试从ctx.state中按参数名取值。需要改变绑定方式时可在创建节点时设置parameter_bindingnode_input。陷阱二会调用run_node的子节点也是父节点同样需要rerun_on_resumeTrue普通函数默认rerun_on_resumeFalseFunctionNode构造函数默认值见 workflow/_function_node.py所以如果某个函数内部还要再调度别的节点必须显式包装inner FunctionNode(funcinner_orchestrator, rerun_on_resumeTrue)否则它作为父节点调用run_node时会触发上文规则 1 的ValueError。陷阱三生成器不能return值在使用了yield的节点里要产出结果必须用yield Event(output...)return value在 async 生成器里是语法错误在 sync 生成器里会被静默忽略。这也是上面示例中orchestrate用yield headline而不是return headline收尾的原因。底层原理调度器的三种执行路径从源码结构看DynamicNodeSchedulerworkflow/_dynamic_node_scheduler.py对一次ctx.run_node调用按以下三种情况处理Fresh全新执行会话中没有该节点路径的历史事件直接创建asyncio.Task运行子节点Completed已完成历史事件显示此前已执行完毕通过懒扫描lazy rehydration重建状态并直接返回缓存输出避免重复执行check_interception决定是快进还是重跑Waiting等待中断历史事件显示子节点因中断处于等待则解析未解决的中断 ID 并传播给父节点父节点整体重跑后在resume_inputs中拿到恢复输入如test_workflow_dynamic_nodes.py中ctx.resume_inputs[fc-1][answer]的用法。父节点在中断恢复后从头重跑、再次走到同一个run_node调用时调度器会基于已记录的 run 状态做出快进或重跑决策这正是rerun_on_resumeTrue之所以是硬性要求的根本原因——父节点的重跑语义由框架保证子节点才能正确地被去重或恢复。适用场景小结循环重试生成-评估-重试本文示例条件路由根据中间结果在多个节点间选择下一步提前退出条件满足时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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 ADK Python 动态节点调度实战：用 `ctx.run_node` 把图控制流变成普通 Python 代码
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 ADKAgent Development KitPython 版 Workflow 中的核心机制——动态节点调度Dynamic Node Scheduling。通过Context.run_node(...)你可以在一个节点内部直接运行另一个节点并拿到它的输出从而把静态有向图中的路由边改写成普通 Python 的循环、条件分支和提前退出。读完本文你将掌握ctx.run_node的全部参数语义、框架强制执行的运行规则、命令式 Workflow 的编写方式以及三个最容易踩的陷阱。什么是动态节点调度在标准的 Workflow 图模型中节点之间的流转由预先声明的边edges决定例如文档示例中的Workflow(nameroot_agent, edges[(START, orchestrate)])。这种声明式路由适合结构固定的流程但当下一步该运行哪个节点需要在运行时根据结果动态决定时例如循环重试、条件跳转、提前退出边的数量会爆炸式增长难以维护。动态节点调度把控制权交还给节点内部await ctx.run_node(...)从节点内部运行另一个节点并返回其输出。正如参考文档所说它把图控制流变成普通 Python循环、条件、提前退出写起来就是循环、条件、提前退出。被动态调用的节点会成为当前节点的一个子运行child run其路径形如/wf1/node_a1带有自己的 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( namegenerate_headline, instructionWrite a headline about the topic {topic}., ) evaluate_headline Agent( nameevaluate_headline, modesingle_turn, instructionGrade whether the headline is tech-related., output_schemaFeedback, ) node(rerun_on_resumeTrue) 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_inputheadline) ) if feedback.grade tech-related: yield headline break root_agent Workflow(nameroot_agent, edges[(START, orchestrate)])关键点yield Event(state{topic: node_input})把入参写入会话状态供子 Agentgenerate_headline的 instruction 引用了{topic}读取await ctx.run_node(generate_headline)不传node_input标题 Agent 从ctx.state中取topicawait ctx.run_node(evaluate_headline, node_inputheadline)把上一步的输出作为入参传给评判 Agent评判结果通过output_schemaFeedback强类型化用Feedback.model_validate(...)解析grade tech-related时yield headline并break循环结束。注意node(rerun_on_resumeTrue)是父节点调用run_node的硬性前提详见下文框架强制执行的规则示例中两个 Agent 子节点本身由 Agent 封装、默认可重跑而编排函数必须显式声明该标志。ctx.run_node参数详解ctx.run_node的完整签名见 agents/context.pyawait ctx.run_node( node, # a function, Agent, BaseTool, or BaseNode node_inputNone, *, use_as_outputFalse, run_idNone, use_sub_branchFalse, override_branchNone, )参数作用node要动态运行的节点可以是普通函数、Agent、BaseTool或任何BaseNode框架内部通过 workflow/utils/_workflow_graph_utils.py 的build_node将其规范化为节点node_input传给被调节点的输入默认Noneuse_as_output为True时子节点的输出成为父节点的输出父节点自身的输出事件被抑制避免重复run_id自定义这次执行的 run_id替代自动编号use_sub_branch为True时在分支路径上追加node_namerun_id把事件与兄弟运行隔离override_branch使用指定的 branch 而不是父节点的 branch在源码的_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_resumeTrue调用run_node的节点如果不带rerun_on_resumeTrue会立即抛错。源码在 context.py 中直接校验if not self._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. )原因在文档中写得很清楚动态调度的子节点可能因用户输入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( fExplicit 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_outputTrue每次父执行至多一次第二次调用会抛出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_inputnode_input) if success in res_a: return await ctx.run_node(step_b, node_inputres_a) return await ctx.run_node(step_c, node_inputres_a)这种写法与 tests/unittests/workflow/test_workflow_dynamic_nodes.py 中大量端到端测试验证的模式一致父节点rerun_on_resumeTrue调度子节点、接收输出、合并后yield结果覆盖了全新执行、中断恢复、嵌套动态节点与use_as_output委托等场景。这种风格下的三个陷阱陷阱一普通函数的参数从 state 绑定而不是从node_input绑定节点参数绑定的默认值是state见 workflow/_function_node.py 的parameter_binding参数说明因此run_node(fn, node_inputx)传过去的值只有通过一个字面命名为node_input的参数才能接收到def my_worker(node_input: str): # 必须叫这个名字否则值到不了 return fDone: {node_input}若函数参数名不叫node_input框架会尝试从ctx.state中按参数名取值。需要改变绑定方式时可在创建节点时设置parameter_bindingnode_input。陷阱二会调用run_node的子节点也是父节点同样需要rerun_on_resumeTrue普通函数默认rerun_on_resumeFalseFunctionNode构造函数默认值见 workflow/_function_node.py所以如果某个函数内部还要再调度别的节点必须显式包装inner FunctionNode(funcinner_orchestrator, rerun_on_resumeTrue)否则它作为父节点调用run_node时会触发上文规则 1 的ValueError。陷阱三生成器不能return值在使用了yield的节点里要产出结果必须用yield Event(output...)return value在 async 生成器里是语法错误在 sync 生成器里会被静默忽略。这也是上面示例中orchestrate用yield headline而不是return headline收尾的原因。底层原理调度器的三种执行路径从源码结构看DynamicNodeSchedulerworkflow/_dynamic_node_scheduler.py对一次ctx.run_node调用按以下三种情况处理Fresh全新执行会话中没有该节点路径的历史事件直接创建asyncio.Task运行子节点Completed已完成历史事件显示此前已执行完毕通过懒扫描lazy rehydration重建状态并直接返回缓存输出避免重复执行check_interception决定是快进还是重跑Waiting等待中断历史事件显示子节点因中断处于等待则解析未解决的中断 ID 并传播给父节点父节点整体重跑后在resume_inputs中拿到恢复输入如test_workflow_dynamic_nodes.py中ctx.resume_inputs[fc-1][answer]的用法。父节点在中断恢复后从头重跑、再次走到同一个run_node调用时调度器会基于已记录的 run 状态做出快进或重跑决策这正是rerun_on_resumeTrue之所以是硬性要求的根本原因——父节点的重跑语义由框架保证子节点才能正确地被去重或恢复。适用场景小结循环重试生成-评估-重试本文示例条件路由根据中间结果在多个节点间选择下一步提前退出条件满足时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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考