企业官网建设流程全解析

ADK 2.0并行Fan-out/Join与嵌套Workflow

多路并行怎么汇、子图怎么封、状态哪里通哪里隔——ADK 2.0的Fan-out/Join与嵌套Workflow，把第35节的Diamond拓扑写成三行edges

本文导航

• 并行为什么是刚需，又是重灾区
• Fan-out：多起点并行的声明式写法
• JoinNode：汇聚语义与卡死风险
• 嵌套Workflow：把子图变成积木
• 状态隔离与共享：一张表说清
• 实战：多源数据采集→分析→报告
• 小结
• 下节预告
• 互动区

并行为什么是刚需，又是重灾区

上一节的Pipeline是根直线，但你把任何一个真实系统画出来，数据流几乎必然长成菱形：多源采集、并行处理、汇聚分析。第35节讲Diamond拓扑的时候我说过一句话：这是多Agent系统里最黄金的结构，也是最容易写错的结构。错在哪？两个地方：

1. 并行写成了串行。三路采集老老实实for循环跑，延迟直接乘三。表面看是性能问题，本质是图结构问题——你脑子里没有扇出的语法，手上就只能写出循环。
2. 汇聚写丢了。三路并行各自跑完，谁等谁？结果怎么合？中间一路挂了怎么办？没有Join语义的并行就是放烟花，好看但收不了场。

ADK 2.0对这两个问题的回答非常"Google"：都不用你自己写，声明出来就行。这一节就拆这两块：Fan-out/Join的配置化，以及嵌套Workflow的封装复用。老规矩，最后上一个完整实战。

Fan-out：多起点并行的声明式写法

ADK 2.0里做扇出，最朴素的方式是多次使用START。"START"关键字可以用在很多行的开头，每一行就是一条从零出发的路径，框架会让它们并行执行：

root_agent=Workflow(name="parallel_workflow",edges=[("START",parallel_task_A),("START",parallel_task_B),("START",parallel_task_C),],)

三行edges，三路并行，没有任何线程池、asyncio.gather的痕迹——并行是图的属性，不是代码的技巧。这就是声明式的味道。

当然扇出不一定在起点。分支边（router+ 字典路由）加上多行路径的组合，可以在图的任何位置制造分叉：先串行走到router，再按route值分发到多路节点。第57节的分支语法 + 这节的多起点语法，两块积木拼起来就是任意形状的扇出。

有一个官方明确警告要划重点：不是所有节点都能并行。特别是"同一个session里跑多个交互式聊天"这种场景（两个Agent都在等用户回复，session直接精神分裂）。进并行分支的节点，要么是无状态函数，要么是single-turn的Agent——又是上一节那个纪律：进图的Agent不许聊天。

JoinNode：汇聚语义与卡死风险

三行代码的汇聚

并行之后要收网。ADK 2.0给了专用的JoinNode，它干两件事：等齐所有上游，收集所有输出：

fromgoogle.adk.workflowimportJoinNode my_join_node=JoinNode(name="my_join_node")root_agent=Workflow(name="fan_out_join",edges=[("START",parallel_task_A,my_join_node),# A路：跑完进join("START",parallel_task_B,my_join_node),# B路：跑完进join("START",parallel_task_C,my_join_node),# C路：跑完进join(my_join_node,final_task_D),# join等齐后，输出给D],)

对应的图就是教科书级的Diamond：

注意JoinNode传给下游的东西：一个输出集合（list）——每路并行节点各自Event.output的汇总。下游节点拿到的是列表，处理时按列表语义来（遍历、zip、聚合随便你）。

卡死风险：JoinNode的"死等"

官方文档有个Caution框，我第一次没当回事，后来真在生产预发环境撞上了：JoinNode只在"所有上游都产出了Event"之后才放行，任何一个上游没给输出，JoinNode就永远等下去，整张图卡死。

什么情况会"没给输出"？我踩过的三种：

1. 上游节点异常退出，Exception没被接住，Event没产出；
2. 上游Agent多轮聊天倾向，跑完不返回（第57节的坑连锁反应）；
3. 分支条件漏了路由值，router返回了一个字典里没有的route，那条路走到一半没了下文。

解法就一条铁律：所有连向JoinNode的节点，必须保证"必有输出"。惯用姿势是给每个并行节点包一层兜底——异常了也返回一个降级Event（比如Event(output={"source": "api", "error": str(e), "rows": []})），把"这一路失败了"本身当作一种数据传给Join。这不就是第31节失败路由思想吗——失败是数据，不是事故。下游聚合节点看到error字段，自己决定降级策略（跳过、重试、还是计入质量报告）。

嵌套Workflow：把子图变成积木

Workflow当节点用

ADK 2.0里，Workflow对象本身就可以作为另一个Workflow图里的节点。子图有自己的edges、自己的节点序列，在父图眼里它就是一个黑盒节点：

fromgoogle.adkimportWorkflow workflow_B=Workflow(name="collect_web",edges=[("START",fetch,normalize)])workflow_C=Workflow(name="collect_db",edges=[("START",query,normalize)])root_agent=Workflow(name="parent_workflow",edges=[("START",task_A1,router),(router,{"RUN_WORKFLOW_B":workflow_B,# 子图作为一个节点被路由"RUN_WORKFLOW_C":workflow_C,}),],)

子图的数据怎么流

嵌套Workflow的输出规则有个细节，官方描述是"冒泡（bubble up）"：

1. 子图内部，节点之间的数据照常走Event.output，父图看不见；
2. 但子图里每个节点完成时，它的Event会同时冒泡给父图——目的是过程可追溯（traceability），父图层面能看到子图内部的执行轨迹；
3. 子图最后一个节点（叶子节点）的输出，被父图提取出来，作为这个"子图节点"对外的输出。

翻译成人话：子图对父图暴露的，只有入口和出口；中间过程以trace形式上报，但不参与父图数据流。

为什么要嵌套

我在实际项目里用嵌套的三个理由，按重要度排：

1. 复用。"采集一路数据源"这个子图（校验→拉取→规范化，三节点）写一次，六个数据源就是六个实例，各自单独测试；
2. 降噪。父图edges数组保持十几行以内，一眼能看懂顶层结构。第25节画的"让团队秒懂的架构图"，在代码层面的对应物就是分层嵌套——顶层图只画到子系统粒度；
3. 团队分工。子图A归数据组维护、子图B归算法组维护，父图归架构组维护，边界清晰，merge冲突都少了。

Go/TypeScript里的嵌套

Go里嵌套同样是自然组合：workflowagent.New出来的agent用workflow.NewAgentNode（或直接作为可组合单元）塞进父图的workflow.Chain；TypeScript里node()工厂函数明确支持把另一个Workflow包成节点。三语言的嵌套语义一致：子图即节点。

状态隔离与共享：一张表说清

嵌套之后最容易糊涂的就是状态归属。ADK 2.0的规则我总结成一张表：

也就是说ADK 2.0的选择是：数据流隔离，控制状态贯通。

• 隔离的是output：子图中间产物不会污染父图的数据通道，父图的下游节点不需要知道子图内部有几个中间形态——这是好事，符合第19节"节点隔离"的原则；
• 贯通的是state：total、attempt这类流程控制变量跨层共享，子图埋点计数，父图统一决策。

这个设计跟LangGraph的子图方案对照着看很有意思：LangGraph子图可以共享父图state schema（同一套TypedDict），数据天然互通，但要小心reducer冲突；ADK是反过来，数据默认隔离、state默认共享。没有对错，只有取舍：LangGraph赌你更需要数据流动的灵活，ADK赌你更需要边界的干净。第61节对比篇会再回到这个点。

一个实操提醒：既然state是session级贯通的，给state的key加命名空间前缀（如web.total、db.total），不然两个子图都写total，后写的覆盖先写的，排查起来非常痛苦。血泪教训，不谢。

实战：多源数据采集→分析→报告

综合实战来了。需求：竞品情报系统，从三个数据源（官网RSS、工单库、应用商店评论）并行采集，各源独立完成"拉取→规范化"子流程，JoinNode汇合后交给分析Agent产出洞察，最后函数节点生成报告。三个子Workflow + JoinNode + 分层结构，把这一节所有知识点一网打尽。

架构

完整代码

# agent.py —— 多源采集→分析→报告：嵌套Workflow + Fan-out/Join 完整实战fromdatetimeimportdatetimefromgoogle.adkimportAgent,Event,Workflowfromgoogle.adk.workflowimportJoinNodefrompydanticimportBaseModel# ---------- 契约层 ----------classSourceRows(BaseModel):source:strrows:list[str]error:str|None=None# 失败也是数据：JoinNode前的必有输出兜底classInsight(BaseModel):summary:strrisk_level:str# low / medium / high# ---------- 子Workflow ①：RSS采集 ----------deffetch_rss(node_input:str):"""模拟拉RSS（真实项目换成httpx异步请求）"""returnEvent(output=["v2.3发布：新增协作编辑","定价页新增企业版档位"])defnormalize_rss(node_input:list):rows=[f"[rss]{r}"forrinnode_input]returnEvent(output=SourceRows(source="rss",rows=rows),state={"rss.count":len(rows)})collect_rss=Workflow(name="collect_rss",edges=[("START",fetch_rss,normalize_rss)])# ---------- 子Workflow ②：工单采集（带兜底，演示失败也是数据） ----------defquery_tickets(node_input:str):try:# 模拟一次偶发故障：上游接口抖动ifdatetime.now().second%10==0:raiseConnectionError("ticket db timeout")returnEvent(output=["导出功能超时","权限组同步失败"])exceptExceptionase:# 关键：失败时也必须产出Event，否则JoinNode死等returnEvent(output=SourceRows(source="tickets",rows=[],error=str(e)))defnormalize_tickets(node_input):ifisinstance(node_input,SourceRows):# 兜底路径：直接透传returnEvent(output=node_input)rows=[f"[tickets]{r}"forrinnode_input]returnEvent(output=SourceRows(source="tickets",rows=rows),state={"tickets.count":len(rows)})collect_tickets=Workflow(name="collect_tickets",edges=[("START",query_tickets,normalize_tickets)])# ---------- 子Workflow ③：应用商店评论 ----------defscrape_reviews(node_input:str):returnEvent(output=["新版闪退一星差评","深色模式好评如潮"])defnormalize_reviews(node_input:list):rows=[f"[reviews]{r}"forrinnode_input]returnEvent(output=SourceRows(source="reviews",rows=rows),state={"reviews.count":len(rows)})collect_reviews=Workflow(name="collect_reviews",edges=[("START",scrape_reviews,normalize_reviews)])# ---------- 汇聚与分析 ----------join=JoinNode(name="collect_join")analyst=Agent(name="analyst",model="gemini-flash-latest",instruction="""你是竞品情报分析师。输入是多源数据的列表， 每项形如SourceRows（source/rows/error字段）。请： 1. 若某源error非空，在结论中标注该源缺失，并用其余源数据分析 2. 输出一段50字内的跨源洞察 + risk_level（low/medium/high） 只输出JSON：{"summary": "...", "risk_level": "..."}""",)defreport(node_input:Insight,**state):"""state贯通演示：直接读子图写入的计数（带命名空间前缀）"""counts={k:vfork,vinstate.items()ifk.endswith(".count")}lines=["# 竞品情报日报",f"- 洞察:{node_input.summary}",f"- 风险:{node_input.risk_level}",f"- 各源条数:{counts}"]returnEvent(message="\n".join(lines))# ---------- 父图：三路并行子图 → Join → 分析 → 报告 ----------root_agent=Workflow(name="intel_pipeline",edges=[("START",collect_rss,join),# Fan-out：三个子Workflow并行("START",collect_tickets,join),("START",collect_reviews,join),(join,analyst,report),# Join后串行收尾],)

代码里三个细节再点一下：query_tickets的兜底把异常转成了带error字段的正常输出，JoinNode永远能等到它的票；三个子图的state key都带source.count命名空间，report里一句推导式全捞出来；analyst的instruction里明确写了"某源error非空怎么办"——降级逻辑要写进Agent的提示词，而不是指望它自己悟（第32节三层失败处理的老原则）。

控制台输出

Running agent intel_pipeline ... [user]: 采集今日竞品情报 [agent(intel_pipeline)]: # 竞品情报日报 - 洞察: 对手v2.3主打协作编辑直击我方短板，评论区深色模式呼声高，建议两周内立项跟进 - 风险: medium - 各源条数: {'rss.count': 2, 'reviews.count': 2, 'tickets.count': 2}

三路子图并行执行，总耗时约等于最慢那一路（这里都是毫秒级模拟，真实场景下三路HTTP请求并行，延迟从"三路之和"降到"三路最大值"）。如果ticket源那天真超时了，你会看到tickets.count消失、洞察里多一句"工单源缺失"——系统照常出报告，而不是整体卡死。

小结

1. Fan-out是声明不是技巧：多个("START", x)行即并行，分支路由可以制造图中任意位置的扇出；进并行分支的节点必须是无状态函数或single-turn Agent。
2. JoinNode = 等齐 + 收集：下游收到的是输出列表；任何上游"没给输出"都会让JoinNode死等——所有连向Join的节点必须兜底产出Event，失败也要作为数据传出来。
3. 嵌套Workflow即节点：子图对父图只暴露入口和出口，内部Event以trace形式冒泡、不参与父图数据流。
4. 状态规则一句话：数据流（output）隔离，控制状态（state）贯通。给state key加命名空间前缀，否则子图互相覆盖。
5. 嵌套的三大收益：复用、降噪（父图保持可读）、团队分工（子系统边界清晰）。

下节预告

到目前为止我们写的图都是"开机那一刻就定死"的——节点几号、边怎么连，全在代码里写死。但有一类需求压根没法预先画图：数据进来才知道走哪条路、中间结果出来才知道要不要加审、量级看到才知道开几路并行。下一节讲ADK 2.0动态Workflow：DynamicNode、ctx.run_node()、三种触发模式（输入驱动/结果驱动/负载驱动），以及自动Checkpoint带来的断点续跑。图不再是设计期产物，而是运行期决策——这才是图工程的深水区。

互动区

你们的系统里有"伪并行"吗——就是那种for循环顺序调用三个接口、美其名曰异步架构的代码？评论区报个数：三个数据源串行采集，你的系统要跑多久？改成今天这种Fan-out/Join结构，理论上能压到最慢那一路的耗时。下期我挑几个典型场景算笔延迟账。

如果觉得本文对你有帮助，欢迎点赞、收藏、关注三连！
本系列持续更新中，关注不迷路~