ADK 2.0动态Workflow运行时决定图结构静态图是设计期画好的地铁线路图动态Workflow是你打上车才决定下一站去哪——DynamicNode、三种触发模式、自动Checkpoint断点续跑本文导航静态图的三种僵硬动态Workflow把图写进代码里DynamicNode三件套node、ctx.run_node、rerun_on_resume三种触发模式输入驱动、结果驱动、负载驱动实战动态评论审核系统自动Checkpoint断点续跑的白送福利静态还是动态我的判断标准小结下节预告互动区静态图的三种僵硬前两节的Workflow(edges[...])有个隐含前提图在代码写完那一刻就定死了。节点是谁、边怎么连、并行几路全在edges数组里焊死。绝大多数场景这没问题——确定性流程本来就该焊死焊得越死越可靠第22节讲的边类型里静态图覆盖了Sequential/Conditional/Parallel的全部需求。但有三类需求静态图写起来会非常别扭循环。“生成→评分→不够好就重新生成最多十轮”——edges数组表达不了while你只能把图展开成固定深度的重复片段或者拆成多次外部调用自己写循环未知深度的递归。任务分解类场景子任务要不要再拆子任务跑起来才知道结构依赖数据。一千条评论走快速通道一次批量审核十万条要走分批抽样多级复核——图结构本身是数据的函数。我在第40节讲Self-Routing Graphs的时候提过这个矛盾Agent自己决定下一步和图框架声明死结构天然打架。ADK 2.0的解法不是让你在edges里硬凑而是给你第二种工作流形态动态Workflow——把图结构从声明式数组搬回程序式代码用原生语言的控制流循环、条件、递归现场决定执行路径。动态Workflow把图写进代码里先看官方最小示例感受一下画风突变fromgoogle.adkimportContext,Workflowfromgoogle.adk.workflowimportnodefromtypingimportAnynode(namehello_node)defmy_node(node_input:Any):returnHello World# 这不是普通异步函数——这是一个DynamicNode动态编排节点node(rerun_on_resumeTrue)asyncdefmy_workflow(ctx:Context,node_input:str)-str:# ctx.run_node 执行一个节点并返回其输出resultawaitctx.run_node(my_node,node_inputhello)returnresult# 动态工作流仍然挂在Workflow的edges里——只是图上只有一个大节点root_agentWorkflow(nameroot_agent,edges[(START,my_workflow)],)看明白这个结构了吗动态Workflow没有抛弃图而是把一整段动态逻辑封装成一个节点外面还是那张静态edges表。所以ADK 2.0的真实用法是混合式的内部执行低风险高风险START静态节点校验/预处理DynamicNode 动态编排节点内部用while/if现场决定路径静态节点格式化/输出run_node 分类器run_node 快速通过run_node 人工审核外层静态图管骨架内层DynamicNode管现场决策。这个分层跟第64节要讲的CrewAI Flows异曲同工——确定性外壳 动态内核正在成为各家框架的收敛方向。对比一下第51节Claude Code的Dynamic Workflows那边是纯JS编排脚本图完全沉在代码里ADK这边是静态图和动态代码可以在一张图里混排动态部分还白送Checkpoint。定位差异记一下第61节对比要用。DynamicNode三件套node、ctx.run_node、rerun_on_resume①node装饰器把函数升级为可追踪节点普通函数加个node(name...)就成了动态图里的注册节点——每次执行会被框架记录、可被Checkpoint。Python用装饰器Go和TypeScript换成构造函数// GoNewFunctionNode 替代 nodeNewDynamicNode 替代 node(rerun_on_resumeTrue)classifyNode:workflow.NewFunctionNode(classifyFn)orchestrator:workflow.NewDynamicNode(func(ctx context.Context,instring)(string,error){// workflow.RunNode 是 ctx.run_node 的Go等价物out,err:workflow.RunNode(ctx,classifyNode,in)returnout,err})// TypeScript没有装饰器node()工厂函数一视同仁地包函数/Agent/子WorkflowconstmyNodenode(()Hello World,{name:hello_node});constmyWorkflownode(async(ctx:NodeContext,_input:string){constresultawaitctx.runNode(myNode,hello);returnresult.output;// 注意runNode返回节点结果对象取.output拿值},{name:my_workflow,rerunOnResume:true},);②ctx.run_node()动态图里的执行下一跳编排节点里不直接调用子函数而是await ctx.run_node(某节点, node_input...)。这一步是动态Workflow的灵魂每次run_node都被记录进执行历史从而获得Checkpoint能力。如果你直接await my_node(...)裸调函数也能跑但断点续跑时框架不知道它执行过会重跑或漏跑——又是一个官方文档没加粗但绝对值钱的细节。TypeScript/Go的注意点一样run_node返回的是节点结果对象要取.output才是值。③rerun_on_resumeTrue续跑语义的开关凡是内部调用了ctx.run_node的编排节点必须设置rerun_on_resumeTrue。逻辑是恢复执行时编排节点的函数体会整个重跑一遍但已经执行过的子节点直接从Checkpoint回放结果、不再真正执行。于是while循环里跑过的轮次天然跳过未完成的继续——断点续跑对动态逻辑默认可用这是声明式静态图很难白送的东西LangGraph要靠Checkpointer显式thread管理才能达到同等效果第61节对比再展开。三种触发模式输入驱动、结果驱动、负载驱动动态Workflow什么时候该动态我把两年多Agent项目里见过的动态化需求归纳成三种触发模式。这个分类不是官方术语是我自己的实战笔记但拿去套你的需求基本一抓一个准。触发模式图结构由什么决定典型场景动态化的收益输入驱动输入数据的类型/字段文本走语义链路、表格走统计链路一套系统吃多种输入不用维护多张图结果驱动中间结果的置信度/类别低置信度追加Verifier节点复核质量自适应好数据快跑烂数据多查负载驱动数据量级1千条单路处理10万条分批并行成本和延迟随规模弹性伸缩三种模式的代码骨架高度一致差异只在决策依据从哪来node(rerun_on_resumeTrue)asyncdeforchestrator(ctx:Context,job:dict)-dict:# ---- 输入驱动看输入决定走哪条链路 ----ifjob[type]table:routeawaitctx.run_node(stat_chain,node_inputjob)# 统计链路else:routeawaitctx.run_node(semantic_chain,node_inputjob)# 语义链路# ---- 结果驱动看中间结果决定加不加复核 ----ifroute[confidence]0.7:routeawaitctx.run_node(verifier,node_inputroute)# 低置信→复核节点# ---- 负载驱动看量级决定并行度 ----batch_size500iflen(route[items])50_000else5_000results[]forbatchinchunked(route[items],batch_size):results.append(awaitctx.run_node(batch_worker,node_inputbatch))returnmerge(results)第37节Verifier拓扑、第38节Pipeline vs Parallel的选择问题、第40节Self-Routing在这里全部从设计期决策变成了运行期决策。你不再需要预判数据的形状只需要预写应对的节点。实战动态评论审核系统上完整实战。需求UGC评论审核——文本长度和风险等级不同走完全不同的处理深度批量大小按当日总量伸缩低置信结果自动加一轮复核。三种触发模式一次打全。# agent.py —— 动态评论审核三种触发模式综合实战importmathfromgoogle.adkimportAgent,Context,Event,Workflowfromgoogle.adk.workflowimportnodefrompydanticimportBaseModel# ---------- 契约 ----------classComment(BaseModel):id:strtext:strclassVerdict(BaseModel):id:straction:str# approve / reject / reviewconfidence:float# 0~1reason:str# ---------- 静态原子节点被动态编排调用 ----------node(nametriage)deftriage(node_input:list[Comment]):预分类规则引擎先过一遍高危词直接拒HIGH_RISK(诈骗,代开发票,外挂)out[]forcinnode_input:ifany(winc.textforwinHIGH_RISK):out.append(Verdict(idc.id,actionreject,confidence0.99,reason规则命中高危词))else:out.append(Verdict(idc.id,actionreview,confidence0.5,reason待LLM判定))returnEvent(outputout)classifierAgent(namecomment_classifier,modelgemini-flash-latest,instruction你是评论审核员。对每条评论判定 - approve正常内容 - reject辱骂/广告/违规 - review拿不准的边界内容 输出JSON列表每项含id/action/confidence/reason。,)node(nameverifier)defhuman_queue(node_input:list[Verdict]):复核兜底低置信一律转人工真实项目接入Human input节点hardened[v.model_copy(update{action:review,reason:低置信转人工: v.reason})forvinnode_inputifv.confidence0.7]returnEvent(outputhardened,state{escalated:len(hardened)})node(namefinalize)deffinalize(node_input:list[Verdict],escalated:int0):stats{a:sum(1forvinnode_inputifv.actiona)forain(approve,reject,review)}returnEvent(message(f审核完成: 通过{stats[approve]}| 拒绝{stats[reject]}| f转人工{stats[review]}(其中低置信加严{escalated}条)),outputnode_input,)# ---------- DynamicNode三种触发模式的编排核心 ----------node(rerun_on_resumeTrue)asyncdefaudit_orchestrator(ctx:Context,comments:list[dict])-list:items[Comment(**c)forcincomments]# 【负载驱动】总量决定批大小10万级用小批保内存千级用大批省开销batch_size200iflen(items)50_000else(2_000iflen(items)1_000elselen(items))all_verdicts:list[Verdict][]foriinrange(0,len(items),batch_size):batchitems[i:ibatch_size]# 【输入驱动】批次特征决定链路短文本走规则预筛LLM长文本直接LLMifsum(len(c.text)forcinbatch)/len(batch)30:preawaitctx.run_node(triage,node_inputbatch)pending[bforbinpreifb.actionreview]else:pendingbatch aiawaitctx.run_node(classifier,node_inputpending)# 【结果驱动】置信度决定是否加严LLM结果低置信的送复核节点low[vforvinaiifv.confidence0.7]iflow:escalatedawaitctx.run_node(verifier,node_inputlow)ai[vforvinaiifv.confidence0.7]escalated# 规则引擎已判的 LLM判的合并入账rule_done[bforbin(preiflen(batch)andsum(len(c.text)forcinbatch)/len(batch)30else[])ifb.action!review]all_verdictsrule_doneaireturnall_verdicts# ---------- 外层静态图校验 → 动态编排 → 汇总 ----------node(nameintake)defintake(node_input:str):importjsonreturnEvent(outputjson.loads(node_input))root_agentWorkflow(namecomment_audit,edges[(START,intake,audit_orchestrator,finalize)],)跑一批混合评论控制台输出Running agent comment_audit ... [user]: [{id:c1,text:这波更新真不错},{id:c2,text:加V代开发票},{id:c3,text:说好的深色模式呢有点失望但理解},{id:c4,text:作者写得太好了学到了}] [agent(comment_audit)]: 审核完成: 通过3 | 拒绝1 | 转人工0(其中低置信加严0条)c2被规则引擎直接击杀输入驱动选择了规则LLM混合链路其余短评走LLM判定全部高置信通过。同样的edges数组喂十万条评论时batch_size自动切到200、走多轮批处理喂四条时一把梭——图没改一行行为随负载变形。自动Checkpoint断点续跑的白送福利动态Workflow最被低估的能力每个节点的每次执行都被自动记录恢复时已完成的子节点直接回放、跳过重跑。想象审核十万条评论跑到第七万条时服务挂了。静态图方案里你要么整个重跑要么自己在外部维护处理到哪了的账本。动态Workflow里重启后前六万多条已完成的run_node调用直接从Checkpoint读结果接着第七万条跑——复杂控制流默认就是durable和resumable的官方原话。三个工程注意点编排节点必须rerun_on_resumeTrue否则恢复时函数体不重跑子节点状态对不上子节点内部要是有随机性LLM输出、随机采样回放的是历史结果而不是重新生成——这是特性不是bug别惊讶ADK 2.0的Checkpoint存储跟1.0不兼容老项目别共享数据库第57节那个警告的第三遍重要的事说三遍。静态还是动态我的判断标准写到这里给出我的决策清单按顺序问自己图结构在写代码时能画出来吗能→静态edges简单可靠优先不能但只是有限几条备选路径→静态图分支路由第57节的router字典别上动态需要循环/递归/未知深度→动态Workflow一个DynamicNode包掉结构依赖数据量或运行时状态本节实战这种→静态外壳DynamicNode内核的混合结构。一句话版本能用静态绝不动态必须动态就整段封装。动态Workflow的代码是命令式的可读性和可测试性天然低于一张edges表把它压缩在一个编排函数里其余部分保持声明式——这是我在两个项目里反复验证过的平衡点。小结动态Workflow是静态edges的补充而非替代用node(rerun_on_resumeTrue)把一段现场决定路径的逻辑封装成DynamicNode挂回静态图混合结构是最佳实践。三件套node注册节点、ctx.run_node()执行并记录、rerun_on_resumeTrue保证续跑语义——裸调函数会丢Checkpoint。三种触发模式输入驱动类型决定链路、结果驱动置信度决定复核、负载驱动量级决定并行度——动态化的收益分别是多态输入、质量自适应、弹性伸缩。自动Checkpoint是白送的durable断点续跑对循环逻辑默认可用比自建账本省一整层状态管理。决策口诀能静态绝不动态必须动态就整段封装。下节预告图写好了、动态逻辑也封装了接下来回答工程师的终极问题怎么部署到生产、怎么监控、怎么算钱。下一节讲ADK 2.0与Vertex AI的集成实战Agent Engine托管、模型端点配置、Cloud Logging结构化日志、Monitoring告警策略、BigQuery成本追踪一套企业级部署方案从零搭起来。图工程的最后一公里比画图本身更决定成败。互动区你的系统里有没有数据量变了、代码结构却焊死的设计比如大促期间量级翻十倍批处理参数还是平时那套。评论区说说你遇到的场景我看看三种触发模式里哪种的需求最普遍——如果负载驱动呼声高后面加一篇专门讲动态并行度的调参实战。如果觉得本文对你有帮助欢迎点赞、收藏、关注三连本系列持续更新中关注不迷路~