显式图、状态机与条件边

图由节点和连接节点的边组成。在工作流里，节点是一项工作，边规定做完后去哪里。比如办理退款：读取订单后，已付款的订单继续核对金额，尚未付款的订单结束处理。这种把步骤和连接关系写出来的流程叫显式图。

日常办事也有这样的路线：先取号，去窗口核对材料，材料齐全就办理，缺材料就补交。工作流把这些路线写进程序，执行时按当前数据选择分支。Agent 可以在其中一个节点生成建议，其他节点负责查资料、检查建议或等待确认。

用一条退款流程认识三个概念

状态机描述的是“现在处于什么状态，满足什么条件后变成什么状态”。例如 running 表示处理中，等待确认时变为 waiting_human。图侧重工作怎样连接，状态机侧重进度怎样变化；同一个程序可以同时表达这两件事。

先拆开三个容易混在一起的词：节点是正在做的事，例如核对审批；路线是已经走过哪些节点，例如 route → review；状态是目前的进度，例如等待主管确认。就像办理证件，“核对材料”是窗口的工作，“取号→核对材料”是办事路线，“待补材料”是办理状态。

代码中的 trace 保存路线，status 保存进度。读输出时先看 trace，沿图找到做过的工作，再看 status，判断任务停在哪里。

下面使用一套固定规则：已付款且金额不超过100元，得到可退款结论；超过100元，查看审批结果；尚未付款，结束处理。100元是例子的分界值，便于观察分支。

保存完整程序，观察四条路线

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-graph" -Force
cd "F:\学习\l4-graph"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

新建 F:\学习\l4-graph 文件夹，将下面代码保存为 graph_demo.py。打开 PowerShell，运行：

cd "F:\学习\l4-graph"
ls
python -X utf8 graph_demo.py

cd 进入这个文件夹，dir 查看文件；应能看到刚保存的 graph_demo.py。如果学习文件夹放在别处，把命令中的路径换成实际位置。

先在 PowerShell 创建并进入学习目录：

mkdir "F:\学习\l4-graph" -Force
cd "F:\学习\l4-graph"

保存下面的文件后，在这个终端运行 dir 核对文件名，再执行本节给出的运行命令。

"""用状态和条件边组织一条退款处理流程。"""
from dataclasses import dataclass, field


@dataclass
class State:
    amount: int
    paid: bool
    approval: bool | None
    status: str = "running"
    trace: list[str] = field(default_factory=list)


def check(state):
    # 入口先检查字段，后面的节点就可以使用明确的整数和布尔值。
    if type(state.amount) is not int or state.amount <= 0:
        raise ValueError("金额应为正整数")
    if type(state.paid) is not bool:
        raise ValueError("付款状态应为布尔值")
    if state.approval is not None and type(state.approval) is not bool:
        raise ValueError("审批应为 True、False 或 None")


def review(state):
    if state.approval is True:
        return "finish"
    state.status = "waiting_human" if state.approval is None else "rejected"
    return "end"


def finish(state):
    # 保存这次流程的处理结论；此处没有接入支付接口。
    state.status = "ready_to_refund"
    return "end"


def reject(state):
    state.status = "rejected"
    return "end"


def route(state):
    if not state.paid:
        return "reject"
    return "finish" if state.amount <= 100 else "review"


NODES = {"route": route, "review": review, "finish": finish, "reject": reject}


def run(state, max_steps=5):
    check(state)
    if type(max_steps) is not int or max_steps < 1:
        raise ValueError("最大步数应为正整数")
    node = "route"
    for _ in range(max_steps):
        # 当前节点做完工作，返回下一节点的名字。
        state.trace.append(node)
        node = NODES[node](state)
        if node == "end":
            return state
    state.status = "step_limit"
    return state


if __name__ == "__main__":
    cases = [
        ("小额已付款", State(80, True, None)),
        ("大额待审批", State(200, True, None)),
        ("大额已批准", State(200, True, True)),
        ("尚未付款", State(80, False, None)),
    ]
    for label, state in cases:
        result = run(state)
        print(f"{label}：{' → '.join(result.trace)}；{result.status}")

程序输出：

小额已付款：route → finish；ready_to_refund
大额待审批：route → review；waiting_human
大额已批准：route → review → finish；ready_to_refund
尚未付款：route → reject；rejected

route 是分流节点，review 是审批检查，finish 保存可退款结论，reject 保存拒绝结论。每行箭头前后的词来自实际执行记录 trace，分号后面是结束这次运行时的状态。

注意第二行：没有审批结果时，流程停在等待人工确认的位置。第四行在付款检查后直接结束，根本没有进入金额审批。把四行放在一起，就能看到同一张图如何走出不同路线。

沿着第三行读代码

第三个样本是 State(200, True, True)：金额200元、已付款、已批准。

1. run 从 route 开始，将它记进 trace。
2. route 发现已付款，金额又大于100，返回字符串 "review"。
3. 下一轮通过 NODES["review"] 找到 review 函数。审批为 True，它返回 "finish"。
4. finish 将状态改为 ready_to_refund，返回 "end"；执行器据此结束。

这里有两个容易混淆的变量。node 保存下一步要执行的函数名，state.status 保存业务进度。比如最后的 node 是 end，业务状态仍是 ready_to_refund。保存业务状态，客服才能知道任务停在哪里。

NODES 是一张“名字到函数”的查找表。调用 NODES[node](state) 就是找到对应函数，把同一份状态交给它。这使节点之间只需要约定输入状态和下一步名称，执行循环负责调度。

为什么审批有三种值

True 表示批准，False 表示拒绝，None 表示还没有决定。等待和拒绝的后续处理不同，因此要分别保存。

bool | None 是类型标注，表达这个字段的预期类型。实际运行时，check 仍会检查传入的值。Python 字符串 "yes" 和布尔值 True 是两种数据；入口检查可以及早发现错误。

每次运行都要有自己的记录

field(default_factory=list) 为每份状态新建一个列表。如果多笔订单共用一份列表，第二笔订单可能混入第一笔的路线。后面的测试会检查这一点。

按固定改法观察分界点

在主程序的第一个样本中，把 State(80, True, None) 改为 State(101, True, None)，保存后执行同一条运行命令。第一行变为：

小额已付款：route → review；waiting_human

金额跨过分界值后，路线转入审批；样本标签只是打印文字，不参与判断。再把同一行的最后一个 None 改为 False，运行后第一行状态变为 rejected。恢复原来的80和 None，四行输出就回到上面的结果。

这种对照可以检查：程序是按数据走分支，还是把样本名称当成判断条件。一个展示得很像流程的图，最终也需要能在代码中找到对应的条件。

加入模型节点时，怎样保持路线清楚

可以在审批前加入“生成退款建议”节点。模型负责返回建议金额和理由；程序把它们解析成字段，再检查金额是否符合订单记录。条件边读取检查后的状态来选择下一步。

对于模型可能提出的多种动作，先定义允许的下一步名称。例如只接受 review、finish 或 reject，再检查所选动作需要的资料是否齐全。这样阅读流程时，可以分别定位“建议在哪里生成”和“下一步在哪里决定”。

节点越来越多时，显式图更容易展示分支、汇合和重复执行。只有两个固定步骤时，普通函数按顺序调用就很清楚；有多条条件路线时，再把连接关系抽出来。框架的图 API 还会提供状态定义、节点注册和条件边等接口，可对照 LangGraph 图接口文档阅读。

检查各条分支

保存下面代码为 test_graph_demo.py，与主程序放在同一文件夹。它覆盖四条演示路线、拒绝审批、字段类型、步数上限和不同任务的记录隔离。

"""检查四条分支、输入类型和步数上限。"""
import unittest
from graph_demo import State, run


class GraphTests(unittest.TestCase):
    def test_small_paid_order(self):
        result = run(State(80, True, None))
        self.assertEqual(result.status, "ready_to_refund")
        self.assertEqual(result.trace, ["route", "finish"])

    def test_waiting_for_human(self):
        result = run(State(200, True, None))
        self.assertEqual(result.status, "waiting_human")

    def test_approved_order(self):
        result = run(State(200, True, True))
        self.assertEqual(result.trace, ["route", "review", "finish"])

    def test_rejected_approval(self):
        self.assertEqual(run(State(200, True, False)).status, "rejected")

    def test_unpaid_order(self):
        result = run(State(80, False, None))
        self.assertEqual(result.trace, ["route", "reject"])

    def test_invalid_inputs(self):
        for state in (State(True, True, None), State(0, True, None),
                      State(80, "yes", None), State(200, True, "yes")):
            with self.subTest(state=state), self.assertRaises(ValueError):
                run(state)

    def test_step_limit(self):
        result = run(State(200, True, True), max_steps=1)
        self.assertEqual(result.status, "step_limit")
        self.assertEqual(result.trace, ["route"])

    def test_traces_are_separate(self):
        first, second = State(80, True, None), State(80, True, None)
        run(first)
        self.assertEqual(second.trace, [])


if __name__ == "__main__":
    unittest.main()

cd "F:\学习\l4-graph"
ls
python -X utf8 -m unittest -v test_graph_demo

结果应显示8项测试通过。这里的 max_steps 限制一次运行最多执行多少个节点。如果图中加入循环，这个上限能让程序在规定步数后返回 step_limit，便于继续检查路线。

现在已经能从执行记录反推出每一步为什么发生。下一篇并发、子图与结果汇聚会让两项独立的查询同时运行，再把结果接回主流程。