Graph Workflows in ADK: Everything You Need to Know

ADK 中的图工作流：你需要知道的一切 | Google Cloud 博客

开发者与实践者

ADK 中的图工作流：你需要知道的一切

2026年9月30日

##### Annie Wang

Google AI Cloud 开发者倡导者

##### Shangjie Chen

Google Cloud AI 软件工程师

图工程是设计工作：将任务分解为节点，通过边连接它们，并决定代码、模型或人员如何控制下一步骤。Agent Development Kit（ADK）的工作流将这种设计转化为可执行过程，由函数和代理执行具体工作。通过退款示例，本文将展示如何并行运行步骤、路由决策、暂停等待人工审核以及处理案例列表。同时解释在什么情况下应静态声明路径，什么情况下应让 Python 根据结果动态安排后续工作。

TL;DR：通过 ADK 的退款工作流，我们将涵盖扇出与扇入、确定性与代理路由、人工介入暂停、并行工作者以及动态编排——并说明何时使用静态图，何时让 Python 决定下一步执行什么。

从单个代理开始

我们可以为一个代理提供工具和指令，让它从头到尾处理退款请求：

lang-py

加载中...

refund_agent = Agent( name="refund_agent", model=MODEL, tools=[fetch_order, fetch_payment, fetch_history], instruction="""你负责处理退款请求。1. 查询订单。2. 检查支付记录。3. 检查客户的退款历史。4. 如果存在未解决的争议或超过30天，拒绝退款。如果金额低于50美元且今年退款次数少于三次，批准退款。5. 给客户发送一封解释决定的邮件。""", )

这使模型负责选择工具、应用策略并撰写回复。但提示信息已经描述了不同的工作环节：三次查询、一个决策和一个响应。将这些环节作为独立节点处理，可以决定每个环节的执行方式。

假设客户已选择订单并点击“申请退款”。应用程序知道订单ID，因此工作流可以从查询开始。

让独立步骤并行执行

订单、支付和退款历史查询都需要订单ID，但都不需要其他查询的结果。尽管提示信息按顺序列出它们，但没有理由让它们相互等待。我们可以并行执行这三个查询。

策略决策则不同：它需要所有三个记录。因此工作流会分成三条路径，收集所有结果后再继续。这两个操作分别称为扇出和扇入。

在ADK中，查询函数可以直接成为节点。嵌套元组可以同时启动它们，JoinNode会等待它们的结果。首先导入和查询签名：

import asyncio from pydantic import BaseModel, Field from google.adk import Agent, Context, Event, Workflow from google.adk.events import RequestInput from google.adk.workflow import JoinNode, START, node async def fetch_order(node_input: str) -> dict: ...

每个查询通过node_input接收订单ID并返回字典。我们可以这样连接它们：

join_case = JoinNode(name="join_case") edges=[ (START, (fetch_order, fetch_payment, fetch_history), join_case), ]

从左到右阅读：同时启动三个查询，然后在它们完成后通过join_case继续执行。

join 操作返回一个以节点名称为键的字典。下一个节点可以直接读取 node_input["fetch_order"]、node_input["fetch_payment"] 和 node_input["fetch_history"] 来收集结果，无需调用模型。JoinNode 指南详细解释了相关实现细节。

如果支付查询需要从订单查询中获取交易 ID，这两个节点将按顺序执行。依赖关系决定了流程图中的边，即使提示信息中已明确列出所有步骤顺序。

你可以在以下视频中了解更多关于扇出和扇入的实现方式：

ADK 图形工程

将工具转换为节点时需要注意一个细节：名为 node_input 的参数会接收前一个节点的输出。其他命名的变量默认绑定到 ctx.state，因此 order_id 会查找 ctx.state["order_id"]，如果找不到则会抛出 ValueError。此处的 str 注解还会将 START 的类型转换为字符串。

将每个请求路由到对应的流程

到目前为止，客户已明确要求退款。在更广泛的客服对话场景中，我们首先需要识别客户意图，再将请求路由到对应的处理流程。"鞋子尺寸不对，能否寄送一双新的？" 应该进入换货流程，而要求退款的请求则应进入退款流程。

这引入了路由节点（router），这是一个决定后续分支执行路径的节点。

• 确定性路由（deterministic router）遵循明确规则，相同输入始终产生相同路由路径。
• 非确定性路由（nondeterministic router）对相同输入可以选择不同分支。
• 代理路由（agent router）使用模型来解析请求，因此选择路径可能具有多样性。

路由节点在工作流定义的路径中进行选择。在我们的客服示例中，可以使用代理来识别客户意图，再通过固定规则应用退款政策。

使用代理路由识别意图

代理可以解析客户消息并分类请求。在 ADK 的一种模式中，它会返回结构化分类结果，然后通过一个小型函数发出对应的 Event(route=...) 事件，将请求路由到指定工作流：

客户消息 → 分类代理 → 路由函数 ├─ REFUND → 退款工作流 ├─ EXCHANGE → 换货工作流 └─ CLARIFY → 提问跟进问题

模型识别意图；图结构定义可用目标路径。如果请求不明确，工作流可以提出跟进问题。ADK 的路由示例展示了这种模式。

这个意图路由节点会位于我们的退款工作流之前。对于已选订单示例，"申请退款"按钮已经明确了意图，因此可以直接进入该工作流。

使用确定性路由应用退款政策

在退款工作流内部，三个查询操作为我们提供了另一个决策的依据：批准、拒绝或要求人工审核。这次政策提供了明确的阈值，因此可以通过函数选择路径：

def refund_policy(amount, days_ago, chargeback_open, priors):
    if chargeback_open or days_ago > 30:
        return "DENY"
    if amount < 50 and priors < 3:
        return "AUTO_APPROVE"
    return "MANUAL_REVIEW"

def route_refund(node_input):
    case = {**node_input["fetch_order"], **node_input["fetch_payment"], **node_input["fetch_history"]}
    route = refund_policy(
        case["amount_usd"],
        case["placed_days_ago"],
        case["chargeback_open"],
        case["prior_refunds_12mo"]
    )
    return Event(output=case, route=route)
# 函数为路径命名

route_refund 会合并记录并返回一个包含三个路由名称之一的案例：AUTO_APPROVE、DENY 或 MANUAL_REVIEW。人工审核处理那些不符合任何自动规则的情况。

这是一个确定性路由器：相同的数据会产生相同的决策，我们可以不调用模型直接测试策略。

确定性路由器

代理路由器

最佳决策来源

代码中的规则

模型对输入的解释

最佳匹配

已知事实和明确策略

难以用规则捕捉的含义

示例

拒绝超过30天的订单

识别换货请求

路由所需的模型调用

无

必需

两种路由器都会在定义的路径中选择。代理路由器可以嵌入静态图中：模型的选择会变化，但可能的连接保持不变。

对于自动批准或拒绝的情况，下一步是向客户解释决策。我们为每个路由分配一个通知代理，从案例数据中生成回复。ADK 会将字典以 Event(output=case, ...) 的形式作为 JSON 用户消息传递给该代理：

approve_notice = Agent( name="approve_notice", model=MODEL, instruction="使用你收到的案例 JSON 告诉客户他们的退款已获批准以及何时能收到款项。简短的邮件，语气友好，不加赘述。", ) denial_notice = Agent( name="denial_notice", model=MODEL, instruction="根据你收到的案例 JSON 告诉客户他们的退款被拒绝以及具体原因。如果包含 reviewer_note，这就是原因。简短的邮件，直接且友好。不要编造政策。", )

现在退款工作流有了从初始查询到决策和回复的路径，并为需要人工处理的案例增加了第三条分支：

workflow = Workflow( name="refund_decision", edges=[ (START, (fetch_order, fetch_payment, fetch_history), join_case, route_refund), (route_refund, {"AUTO_APPROVE": approve_notice, "MANUAL_REVIEW": escalate_to_human, "DENY": denial_notice}), ], )

第一条链获取记录、合并它们并应用策略。第二条链将路由器的决策映射到目标。自动决策直接发送到通知代理；MANUAL_REVIEW 会转到我们下一步要定义的人工审核节点。

第三条分支需要的不只是另一个函数调用。审核人员可能需要几分钟甚至几天才能回复，因此工作流必须暂停并保持案例待处理状态，直到人工决定后继续执行。

审核节点会生成 RequestInput，记录待处理请求并暂停运行。恢复时，rerun_on_resume=True 会重新运行该节点，答案可通过 ctx.resume_inputs 获取：

REVIEW = "refund:review" class ReviewDecision(BaseModel): approve: bool = Field(description="True 表示退款，False 表示拒绝。") note: str = Field("", description="用审核人员的话说明原因。") @node(rerun_on_resume=True) async def escalate_to_human(ctx: Context, node_input: dict): answer = ctx.resume_inputs.get(REVIEW) if answer is None: # 首次运行：提问后停止 yield RequestInput( interrupt_id=REVIEW, message=f"退款 ${node_input['amount_usd']} 的订单 " f"{node_input['order_id']}?", payload=node_input, # 审核人员看到的内容 response_schema=ReviewDecision, ) return # 第二次运行：答案已存在，据此进行路由 yield Event(output={**node_input, "reviewer_note": answer.get("note", "")}, route="AUTO_APPROVE" if answer["approve"] else "DENY")

响应模式为节点提供了用于路由的批准值和需要传递的备注。如果审核者拒绝，通知代理会收到他们的理由和案件信息。另一个边连接审核决策与回复：

edges=[ ..., (escalate_to_human, {"AUTO_APPROVE": approve_notice, "DENY": denial_notice}), ]

ADK 提供了此功能的两种变体。上述是单节点模式：节点重新运行并读取 ctx.resume_inputs ，如 request_input_rerun 示例所示。request_input 示例展示了双节点变体，其中一个节点生成 RequestInput ，审核者的回答作为下一个节点的 node_input 传入——因此在该文件中不会出现 ctx.resume_inputs 。下方链接的配套脚本包含将审核者回答发送回运行的代码。

Workflow 本身也是一个节点。整个退款流程可以成为更大客户服务工作流中的一个步骤。

将相同步骤应用于一批案件

我们现在有一个处理单个退款的流程。假设一批案件已经收集了记录到达。每个案件都需要相同的策略检查，且每批案件的数量可能不同。我们可以使用并行工作者将同一个节点应用于每个项目。

在 ADK 中，设置 parallel_worker=True 会按输入列表中的每个项目运行一次节点，并按原始顺序收集结果。我们可以复用策略函数：

@node(parallel_worker=True) def review_case(node_input): # 每个工作者从输入列表接收一个案件。 case = node_input decision = refund_policy( case["amount_usd"], case["placed_days_ago"], case["chargeback_open"], case["prior_refunds_12mo"], ) return {"order_id": case["order_id"], "decision": decision} def collect_decisions(node_input): # 该节点接收工作者结果列表。 return {"decisions": node_input} batch_review = Workflow( name="batch_review", edges=[(START, review_case, collect_decisions)], )

每个工作者接收一个案件，collect_decisions 接收结果列表。不需要单独的 JoinNode 。该标志也适用于代理——例如，为每个案件编写解释。并行工作者示例展示了这两种形式。

此处的策略计算较为简单。当每个项目需要等待 API 或模型调用时，并行处理会更有价值，但输入和结果的流动方式保持不变。

模式

分布式工作

收集的输出

通过

JoinNode

进行扇出

不同节点执行独立任务

按节点名称键值的字典

并行工作者

每个项目使用相同节点

按输入顺序排列的列表

批量大小可以改变而无需修改图结构。可变工作量仍能适应固定流程。有关执行细节，请参阅并行工作者指南。

让结果决定下一步

我们的退款流程即使在批量包含更多案件或审核者需要更长时间回答时也有已知路径。但有些工作只有在调查时才会变得清晰。

假设一个有争议的退款揭示了第二笔交易。检查它时会引发需要进一步调查的配送问题。现在每个结果都可以生成后续工作。动态节点可以检查这些结果，并在 Python 中安排下一步检查。

ADK 提供 ctx.run_node 来运行另一个节点并等待其结果。为了说明这如何改变编排，我们先以这种方式表达现有的退款流程：

HANDLERS = { "AUTO_APPROVE": approve_notice, "MANUAL_REVIEW": escalate_to_human, "DENY": denial_notice, } @node(rerun_on_resume=True) async def refund_flow(ctx, node_input): # 步骤1：一次性获取所有三条记录。 order, payment, history = await asyncio.gather( ctx.run_node(fetch_order, node_input, use_sub_branch=True), ctx.run_node(fetch_payment, node_input, use_sub_branch=True), ctx.run_node(fetch_history, node_input, use_sub_branch=True), ) case = order | payment | history # 步骤2：应用退款策略——相同的纯函数，不调用LLM。 decision = refund_policy( amount=case["amount_usd"], days_ago=case["placed_days_ago"], chargeback_open=case["chargeback_open"], priors=case["prior_refunds_12mo"], ) # 步骤3：运行选定的处理程序，并将其输出作为当前节点的输出。 await ctx.run_node(HANDLERS[decision], case, use_as_output=True)

asyncio.gather 并行执行查询。Python 合并查询结果，应用策略并执行选定的处理程序。use_as_output=True 使处理程序的输出直接成为父节点的输出，避免重复发送。

动态版本还调整了人工审核节点：在收到回答后，通过 ctx.run_node 调用选定的通知代理。完整的动态脚本（refund_dynamic.py）包含这种变体。

外层流程只需定义入口点：

workflow = Workflow( name="refund_dynamic", edges=[(START, refund_flow)], )

在静态版本中，边列表显示分支和目标节点。在此版本中，我们通过查看 refund_flow 来了解这些信息。动态节点指南涵盖了这种实现方式。

人工审核暂停功能在此依然有效。当回答到达时，refund_flow 会从顶部重新运行，但已完成的 ctx.run_node 调用会从会话历史中返回记录的输出。查询不会重复执行，如同直接函数调用那样。将副作用限制在子节点中，可以让已完成的调用在父节点恢复时重放结果。每个子节点都有独立的追踪跨度，use_sub_branch=True 确保并行子节点的事件记录在不同分支上。

对于这个固定的退款流程，边列表仍然易于检查。Python 版本为我们提供了添加上述调查逻辑的位置：检查结果、选择后续节点、并同时执行独立检查。模型可以建议需要调查的内容，而代码则限制工作量——例如，每个发现最多三个后续步骤，两次调查后进入人工审核。

循环仍然可以嵌入静态图中。固定的草稿 → 检查 → 修订流程可以使用条件回边和限制修订次数的路由器。"静态"描述的是可能的连接方式；实际路径和迭代次数可以变化。图指南涵盖了条件循环的实现。

你也可以在静态工作流中嵌入动态节点，用 Python 实现需要动态处理的阶段，同时保持整体流程的可见性。

确定下一步由谁决策

从一个问题开始：在输入到达前，你能画出可能的流程图吗？包括分支、循环和重复阶段。你不需要预测每个请求的具体路径。

你需要的要素

独立执行的任务，然后收集所有结果

分发任务和汇总结果

根据固定规则选择分支

根据语义解释选择分支

在继续前等待人工决策

RequestInput

对列表中的每个元素执行相同操作

随着结果到达安排工作

动态节点

使用边列表来明确那些连接关系。当结果需要进一步处理或 Python 能更自然地表达控制逻辑时，使用动态编排。一个小型的开放性任务可能只需要一个代理及其工具即可。

在我们的退款工作流中，图结构协调了查询操作，代码应用了策略，人工处理异常情况，模型负责撰写回复。图工程为每个组件赋予了明确的职责——并使整个流程的运作方式更易于理解。

入门指南

• 阅读：[工作流指南](the workflow guide)
• 探索：[工作流示例](the workflow samples)
• 实践：[动手实践教程](the hands-on codelab)，该教程将这些模式应用于马拉松赛日教练场景

发布分类

• Developers & Practitioners