第 5 章 图工作流：ADK 2.0 的核心武器

从第 1 章我们就反复说，ADK 2.0 最大的变化是"图工作流引擎"。这一章，我们终于要正面拆解它。你会发现，图工作流不是加了一个新功能，而是重新定义了"Agent 框架该怎么造"。而这一切，要从 ADK 为什么重写说起。

5.1 为什么 ADK 要从 Agent-Centric 重写为 Graph-Based

5.1.1 纯 Agent 的四个致命问题

ADK 1.x 是一个典型的 "Agent-Centric"（以 Agent 为中心）框架：你定义 Agent、挂工具、写 prompt，然后让 LLM 自己编排执行流程。原型阶段很爽，但一进生产环境，四个问题就暴露了。

问题一：成本爆炸。

想象一个业务流程 A → B → C → D → E，其中 B 和 D 需要 LLM 推理，但 A、C、E 只是确定性逻辑（数据转换、条件判断）。在纯 Agent 模式下，LLM 在每一步都要做"下一步做什么"的决策——即使是 A 到 C 这种确定性转换，也要等 LLM 推理一遍。每多一次推理，就是多一次 token 消耗和延迟。

ADK 官方博客举过一个例子：同样处理一个退货请求，纯 Agent 方案的成本是 Workflow 混合方案的 3-5 倍。原因很简单——你把大量 "if-else" 级别的决策交给了 LLM，而 LLM 的推理成本远高于一行 if condition: do_something()。

问题二：无限循环与幻觉。

纯 Agent 的另一个生产灾难是无限循环。Agent 调用工具后，LLM 要解析工具输出、决定下一步。如果 LLM 产生了幻觉、误读了输出，它可能反复调用同一个工具，或者在几个工具之间死循环。没有确定性的流程约束，这种循环一旦发生就很难自动停止。

问题三：调试像黑盒。

纯 Agent 的执行路径是模型"即兴"决定的——这次走 A→B→D，下次可能走 A→C→D。路径不可复现，出了问题你都不知道该看哪一环。

问题四：测试不可复现。

因为执行路径不确定，同一个输入每次走的路可能不一样，导致测试结果不稳定、无法作为质量门禁。

5.1.2 范式转移：把"可靠的逻辑"和"智能的推理"分开

这四个问题的根源是同一个：纯 Agent 把"确定性"和"智能性"混在一起，全交给 LLM。

ADK 2.0 的回应是一次范式转移：图工作流引擎。

智能体、工具、函数——全都是工作流图上的节点。你可以把业务流程的骨架固化成确定性的图（哪些步骤必须按顺序走、哪些分支按条件走），只在真正需要判断的地方放上 LLM 节点。

这就是我们第 1 章就提到的核心论点：ADK 把"智能的推理"和"可靠的逻辑"分开了。该确定的（流程顺序、条件分支、数据转换）用代码和图确定下来；该智能的（理解意图、生成回复）才交给 LLM。

5.2 图工作流的核心概念

5.2.1 一张图的三要素

ADK 的图工作流有非常清晰的三要素：

1. 节点（Node）：图上的一个步骤。可以是 Agent、函数、或 工具——"each node can be an AI agent, Tool, or your programmed code"
2. 边（Edge）：节点之间的连接，表示执行顺序
3. 路由（Router）：条件分支的分发逻辑

5.2.2 Workflow 类

ADK 2.0 中，图工作流用 Workflow 类创建：

from google.adk import Agent, Workflow, Event
from pydantic import BaseModel

# 节点 1：一个 Agent 节点
city_generator_agent = Agent(
    name="city_generator_agent",
    model="gemini-flash-latest",
    instruction="""Return the name of a random city. Return only the name, nothing else.""",
    output_schema=str,
)

# 节点 2：一个函数节点（确定性的数据转换）
class CityTime(BaseModel):
    time_info: str
    city: str

def lookup_time_function(node_input: str):
    """Simulate returning the current time in the specified city."""
    return CityTime(time_info="10:10 AM", city=node_input)

# 节点 3：另一个 Agent 节点
city_report_agent = Agent(
    name="city_report_agent",
    model="gemini-flash-latest",
    input_schema=CityTime,
    instruction="""Output following line: It is {CityTime.time_info} in {CityTime.city} right now.""",
    output_schema=str,
)

# 节点 4：一个返回 Event 的结束函数
def completed_message_function(node_input: str):
    return Event(message=f"{node_input}\n WORKFLOW COMPLETED.")

# 把节点按执行顺序连成一条边
root_agent = Workflow(
    name="root_agent",
    edges=[
        ("START", city_generator_agent, lookup_time_function,
         city_report_agent, completed_message_function)
    ],
)

注意看这个例子的几个关键点：

第一，Workflow(edges=[...]) 是定义方式。 没有 LangGraph 那种 add_node/add_edge 的链式调用——你用一个 edges 列表声明所有路径。每行（tuple）是一条执行路径，按顺序执行。

第二，数据在节点间流动。 前一个节点的返回值，自动成为下一个节点的输入。city_generator_agent 返回城市名 → lookup_time_function 接收字符串 → 返回 CityTime → city_report_agent 接收 CityTime。用 pydantic BaseModel 定义节点间的数据契约，类型安全、可校验。

第三，("START", ...) 是图的入口。 每个图从 START 开始执行。

5.2.3 节点之间的数据契约

图工作流里，节点之间通过返回值传参。为了让数据流清晰可靠，ADK 用 pydantic 的 BaseModel 定义数据结构。刚才的例子已经展示了：

• Agent 节点可以用 output_schema 约束输出（如 output_schema=str）
• Agent 节点可以用 input_schema 声明期望的输入（如 input_schema=CityTime）
• 函数节点通过返回值类型声明输出（-> CityTime）

这种"契约式"的数据流，是图工作流区别于"自由编排"的关键——每一步的输入输出都有明确的形状，不会出现"模型自由发挥输出一个谁都没想到的结构"。

5.3 条件路由：让图有分支

顺序执行的图只能表达线性流程。真实业务需要分支——"如果 A 走甲分支，如果 B 走乙分支"。这就是路由（Routing）。

5.3.1 路由的工作原理

ADK 的路由分两步：

1. 一个 router 函数：读取前一个节点的输出，决定走哪条路，返回 Event(route=...)
2. 一个分支字典：把路由值映射到对应的处理节点

# 路由函数：把消息分类
def router(node_input: str):
    # node_input 是前一个节点的输出（例如 "BUG,CUSTOMER_SUPPORT"）
    routes = [r.strip() for r in node_input.split(",") if r.strip()]
    return Event(route=routes)

# 分支处理节点
def response_1_bug(node_input: str):
    return Event(message="Handling bug...")

def response_2_support(node_input: str):
    return Event(message="Handling customer support...")

def response_3_logistics(node_input: str):
    return Event(message="Handling logistics...")

root_agent = Workflow(
    name="routing_workflow",
    edges=[
        ("START", process_message, router),
        (
            router,
            {
                "BUG": response_1_bug,
                "CUSTOMER_SUPPORT": response_2_support,
                "LOGISTICS": response_3_logistics,
            }
        ),
    ],
)

5.3.2 路由的本质

路由函数是纯代码——它不用 LLM，就是一个普通的 Python 函数，返回 Event(route=routes)。这非常关键：

• 确定性：BUG 永远走 response_1_bug，不会"有时走错分支"
• 零成本：路由这一步不花 token
• 可测试：路由逻辑可以单测

这就是图工作流"把 if-else 还给代码"的体现。第 6 章我们会看到更复杂的动态路由。

5.4 主线项目：用图工作流重写退货审批流程

5.4.1 业务需求：退货审批

现在云销要上线一个退货审批流程。业务规则是：

1. 用户发起退货申请
2. 检查资格：订单是否在退款窗口内、商品是否可退（电子产品 7 天、家居 15 天）
3. 资格判断：
   • 有资格 → 生成退货单 → 完成
   • 无资格 → 告知用户原因 → 完成
4. 每一步都要有清晰记录

这个流程有明确的分支和顺序——它是"确定性逻辑 + 少量 LLM 判断"的混合体，正是图工作流的典型场景。

5.4.2 流程设计

START
  │
  ▼
┌─────────────────┐
│ intake_agent    │  LLM 节点：理解用户退货申请，提取订单号和商品类别
└────────┬────────┘
         │ 输出: RefundRequest(order_id, category)
         ▼
┌─────────────────┐
│ check_eligibility│  函数节点：查订单+查政策，判断资格（确定性逻辑）
└────────┬────────┘
         │ 输出: EligibilityResult(eligible: bool, reason)
         ▼
      router（函数节点）
      /          \
    eligible      not_eligible
      │              │
      ▼              ▼
┌─────────────┐  ┌──────────────┐
│ create_refund│  │ reject_agent │  告知原因
│  函数节点     │  └──────────────┘
└─────────────┘
      │
      ▼
   DONE（Event 结束）

5.4.3 完整代码

from google.adk import Agent, Workflow, Event
from pydantic import BaseModel


# ---------- 数据契约 ----------
class RefundRequest(BaseModel):
    order_id: str
    category: str  # 'electronics' | 'home'


class EligibilityResult(BaseModel):
    eligible: bool
    reason: str


# ---------- 节点 1：理解退货申请（LLM 节点） ----------
intake_agent = Agent(
    name="intake_agent",
    model="gemini-flash-latest",
    instruction=(
        "你是退货申请受理员。用户会描述他们的退货需求。"
        "请提取订单号（order_id）和商品类别（category，electronics 或 home），"
        "按 RefundRequest 格式输出。"
    ),
    output_schema=RefundRequest,
)


# ---------- 节点 2：检查资格（确定性函数节点） ----------
def check_eligibility(node_input: RefundRequest) -> EligibilityResult:
    """检查退货资格：查订单状态 + 查退款政策。"""
    # 查订单（模拟）
    orders = {
        "ORD-20260901-001": {"status": "已发货", "days_since": 2},
        "ORD-20260901-002": {"status": "待发货", "days_since": 10},
        "ORD-20260903-003": {"status": "已签收", "days_since": 20},
    }
    # 查政策（模拟）
    policies = {
        "electronics": {"window_days": 7, "condition": "商品未拆封"},
        "home": {"window_days": 15, "condition": "商品及包装完好"},
    }

    order = orders.get(node_input.order_id)
    policy = policies.get(node_input.category)

    if not order:
        return EligibilityResult(eligible=False, reason=f"订单 {node_input.order_id} 不存在")
    if not policy:
        return EligibilityResult(eligible=False, reason=f"类别 {node_input.category} 不支持退货")

    if order["days_since"] > policy["window_days"]:
        return EligibilityResult(
            eligible=False,
            reason=f"已超出{policy['window_days']}天退货窗口",
        )
    if order["status"] == "已签收" and node_input.category == "electronics":
        return EligibilityResult(
            eligible=False,
            reason="电子产品签收后已激活，不支持无理由退货",
        )
    return EligibilityResult(eligible=True, reason="符合退货条件")


# ---------- 节点 3：路由（确定性分支） ----------
def eligibility_router(node_input: EligibilityResult):
    """根据资格结果路由到对应分支。"""
    if node_input.eligible:
        return Event(route="eligible")
    return Event(route="not_eligible")


# ---------- 节点 4a：生成退货单（确定性） ----------
def create_refund(node_input: EligibilityResult) -> Event:
    return Event(message=f"退货申请通过！已生成退货单，请将商品寄回。\n原因：{node_input.reason}")


# ---------- 节点 4b：拒绝退货（LLM 节点，生成友好话术） ----------
reject_agent = Agent(
    name="reject_agent",
    model="gemini-flash-latest",
    instruction=(
        "你是退货拒绝通知员。根据给定的拒绝原因，用友好、专业的语气告知用户，"
        "并说明依据的退货政策。"
    ),
)


# ---------- 组装图工作流 ----------
root_agent = Workflow(
    name="refund_approval_workflow",
    edges=[
        ("START", intake_agent, check_eligibility, eligibility_router),
        (
            eligibility_router,
            {
                "eligible": create_refund,
                "not_eligible": reject_agent,
            }
        ),
    ],
)

5.4.4 这个设计好在哪里

我们拆开看这个工作流，体会图工作流的威力：

第一，只有两个 LLM 节点，其余全是确定性代码。

• intake_agent：LLM——因为理解自然语言（"我想退那个电子产品"→ 提取结构化字段）只能靠 LLM
• check_eligibility：纯函数——查订单、比对日期，这些是确定的业务逻辑
• eligibility_router：纯函数——if/else 分支
• create_refund：纯函数——生成结果
• reject_agent：LLM——因为"用友好话术拒绝"需要自然语言生成

对比纯 Agent 方案：纯 Agent 会在"要不要查资格、资格怎么判断、拒绝话术怎么说"每一步都调用 LLM 决策。图工作流把 5 步里的 3 步变成了零成本、确定性、可测试的代码。成本降 3-5 倍，可靠性升一个量级——这正是 5.1 说的那个问题。

第二，流程是"画"出来的，谁都能看懂。

edges 列表就是一张流程图的文字版。业务方、测试、新来的开发，一眼就能看懂退货审批的完整路径。而纯 Agent 的流程藏在模型脑子里，没人能完整复述。

第三，分支是确定性的。

用户 20 天前签收的电子产品，100% 走 not_eligible 分支，绝不会"这次走拒绝、下次走通过"。可复现、可测试——这是生产系统的底线。

5.5 确定性路由 vs LLM 自由编排：一张对比表

这一节把图工作流和纯 Agent 做一个系统对比，这是全书反复出现的主题。

维度	图工作流（确定性）	纯 Agent（LLM 自由编排）
流程控制	开发者定义，显式可见	模型即兴决定，不可见
成本	仅 LLM 节点花 token	每步都可能花 token
可靠性	分支确定，可复现	可能循环、可能幻觉
可测试性	可单测每个节点和分支	输出不稳定，难测试
适合场景	有明确顺序/分支的流程	开放式的探索型任务
调试	事件流 + 图结构清晰	黑盒，路径不可复现

关键认知：这不是"图好还是 Agent 好"的问题，而是"哪个环节该用哪个"的问题。

ADK 2.0 的智慧在于：它让你在同一个框架里同时拥有两者。流程骨架用图（确定性），需要判断的地方放 Agent 节点（智能性）。这就是 5.1 说的"把可靠的逻辑和智能的推理分开"的具体落地。

「为什么 ADK 这样设计」：把 if-else 还给代码

这一章的核心设计哲学，可以浓缩成一句话：把 if-else 还给代码，把判断留给 LLM。

我们来看一个反例——如果不用图工作流，用纯 Agent 写退货审批会怎样：

# 伪代码：纯 Agent 方案的"思考过程"
Agent 收到"我想退电子产品" →
  步骤 1: LLM 想"我需要查订单资格" → 调 check 工具
  步骤 2: LLM 读结果，想"我需要判断是否在窗口内" → 这一步本来是个 if，但 LLM 要"推理"
  步骤 3: LLM 想"应该告诉用户能退" → 生成话术

发现没有——步骤 2 是个 if (days_since > window_days) 的判断，纯 Agent 却要 LLM "推理"一遍。一次推理 = 一次 token 消耗 + 一次延迟 + 一次可能的判断错误。

而图工作流把步骤 2 变成了：

if order["days_since"] > policy["window_days"]:
    return EligibilityResult(eligible=False, reason="超出窗口")

零 token、零延迟、零出错、可单测。

这就是"把 if-else 还给代码"的含义：凡是确定性的逻辑，就用确定性的代码表达；只有真正需要理解力、生成力的地方，才动用 LLM。这个原则，就是 ADK 2.0 图工作流的灵魂，也是它和"让 LLM 编排一切"的框架最根本的区别。

本章小结

• 纯 Agent 的四个致命问题：成本爆炸、无限循环与幻觉、调试黑盒、测试不可复现
• 范式转移：ADK 2.0 从 Agent-Centric 重写为 Graph-Based，把"可靠的逻辑"和"智能的推理"分开
• 三要素：节点（Agent/函数/工具）、边（执行顺序）、路由（条件分支）
• Workflow 定义方式：Workflow(name=..., edges=[...])，节点通过返回值 + pydantic BaseModel 传数据
• 条件路由：router 函数返回 Event(route=...)，配分支字典分发——纯代码、确定性、零成本
• 退货审批实战：5 步流程只有 2 个 LLM 节点，其余全是确定性代码，成本降 3-5 倍
• 核心哲学：把 if-else 还给代码，把判断留给 LLM

练习

1. 加一个分支：给退货审批加"物流异常"分支——如果订单已签收超过 30 天，除了拒绝还要转给人工处理（新增一个分支和节点）。
2. 纯函数节点：把 check_eligibility 改成返回包含"可退金额"字段，验证数据契约（BaseModel）如何流动。
3. 对比实验：用同一批退货请求分别跑"纯 Agent 版"和"图工作流版"，记录 token 消耗和延迟，验证成本差异。

下一章预告：第 6 章，图工作流再进一步——动态工作流。当流程需要"循环直到成功"、"根据复杂条件实时分支"、甚至"暂停等待人工审批"时，静态图就不够用了。我们来看动态工作流和人机协作（HITL）怎么解决这些问题。