跳到主要内容

数据截至 (上游 commit f74d023f9360)

02 · Flow 编排:主循环、动作跳转与嵌套

上一章讲了「节点」和「图」是静态结构。本章讲运行时:Flow 怎么拿着起点节点,一步步跳到终点。核心只有一个 while 循环 Flow._orch,但里面藏着两个巧妙决定:每步 copy.copy 节点、Flow 自己也是一个节点。

3.1 编排主循环:_orch​

思路​

有了「后继表」,跑一次流程就是一个循环:从起点开始,反复「执行当前节点 → 用它报的动作名查下一个 → 换成下一个」,直到查不到下一个为止。

真实实现​

def _orch(self,shared,params=None):
    curr,p,last_action =copy.copy(self.start_node),(params or {**self.params}),None
    while curr: curr.set_params(p); last_action=curr._run(shared); curr=copy.copy(self.get_next_node(curr,last_action))
    return last_action

见 pocketflow/__init__.py:46-49。一行 while 干完全部编排:

  1. curr._run(shared)——跑当前节点的 prep/exec/post(第 01 章),拿到 last_action。
  2. self.get_next_node(curr, last_action)——按动作名查后继。
  3. 把 curr 换成后继(且 copy.copy 一份),循环。查不到时后继是 None,while curr 退出。
  4. 返回最后一个动作名——嵌套时外层会拿它继续跳(见 3.4)。

查后继:get_next_node​

def get_next_node(self,curr,action):
    nxt=curr.successors.get(action or "default")
    if not nxt and curr.successors: warnings.warn(f"Flow ends: '{action}' not found in {list(curr.successors)}")
    return nxt

见 pocketflow/__init__.py:42-45。两个细节:

  • action or "default":节点 post 若返回 None(最常见,表示「没什么特别动作」),就退回查 "default" 这条边——所以 a >> b 的默认边对应「post 返回 None」的情况。
  • 「有后继但没匹配上」才告警:如果节点压根没连任何后继(not curr.successors),那是正常终点,不告警;但如果连了后继、却报了一个没连线的动作名,就 warnings.warn("Flow ends: '...' not found in [...]")——这是最常见的调试线索。test_flow_basic.py:165-220 专门测了这两种告警。

3.2 为什么每步都 copy.copy?​

现象​

注意 _orch 里两处 copy.copy:起点 copy.copy(self.start_node),以及每次跳转 copy.copy(self.get_next_node(...))。跑的从来不是你建图时那个节点对象,而是它的浅拷贝。

为什么​

关键是循环 / 复用。看 chat 示例的自环 chat - "continue" >> chat:同一个 ChatNode 对象在图里被反复经过。如果直接在原对象上跑,像 cur_retry(重试计数,pocketflow/__init__.py:30)这种运行期临时状态会在多次经过间串味。copy.copy 每次给一个干净的浅拷贝当「运行实例」,让图里的节点定义保持无状态、可安全复用。

是浅拷贝,不是深拷贝(重要)​

copy.copy 只复制节点这一层。successors 字典、params 字典是共享引用——所以拷贝出来的实例仍指向同一张图、能继续跳转。而 shared 从头到尾根本没被拷贝(它只作参数传给 _run),所以所有节点实例读写的是同一份数据。一句话:拷贝的是「执行者」,共享的是「图」和「数据」。

3.3 Flow 也是节点:_run 与 post​

Flow 继承自 BaseNode(pocketflow/__init__.py:39),所以它自己也有 prep/exec/post 那套外壳。它覆写了 _run,把「跑节点」换成「跑整张图」:

def _run(self,shared): p=self.prep(shared); o=self._orch(shared); return self.post(shared,p,o)
def post(self,shared,prep_res,exec_res): return exec_res

见 pocketflow/__init__.py:50-51。对比第 01 章 BaseNode._run 的 e=self._exec(p),这里换成了 o=self._orch(shared)——Flow 的「exec」就是跑一遍内部子图。而 Flow.post 默认原样返回子图的最后动作名,这一步是嵌套能工作的关键(下一节)。

注:self.start_node 在 __init__(self,start=None) 里设置,也可事后 flow.start(node)(pocketflow/__init__.py:40-41)。

3.4 嵌套:把一个 Flow 当节点塞进另一个 Flow​

思路​

既然 Flow 是 BaseNode,它就能出现在别的图里当一个节点——inner_flow >> next_node、甚至 inner_flow - "done" >> X。这让你把一段子流程封装成一个黑盒节点,复用/组合。

动作怎么穿透边界​

嵌套能按动作分支,靠的是这条链:内层子图跑完 → _orch 返回最后一个动作名 → Flow.post 原样返回它 → 对外层而言,这个「内层 Flow 节点」的 post 就返回了那个动作名 → 外层据此查自己的后继。

  外层 Flow 的循环里,当前节点 = inner_flow
        │ inner_flow._run(shared)
        ▼
   inner_flow._orch(...)  跑完内部所有节点
        │ 返回内部最后节点的动作名,如 "inner_done"
        ▼
   inner_flow.post(...) 原样返回 "inner_done"
        │
        ▼
   外层 get_next_node(inner_flow, "inner_done")
        └▶ 命中 inner_flow - "inner_done" >> path_b  → 跳到 path_b

test_flow_composition.py:110-148(test_composition_with_action_propagation)正是验证这条:内层最后节点报 "inner_done",外层据此走到 path_b。同文件 test_flow_as_node、test_nested_flow 验证多层嵌套时 shared 全程共享、数据正确累积。

关键细节 / 坑​

  • Flow 自己的 _orch 用它自己的 prep 结果吗? 不。Flow._orch 只用 params,不吃 prep 的返回值——Flow.prep 的返回值只在 _run 里传给 Flow.post 当 prep_res,不影响子图。子图靠 shared 拿数据。
  • params 的传递: _orch 起手 params or {**self.params}——外层调 _orch 时可注入一批 params,否则用 Flow 自己的。节点跑前 curr.set_params(p) 把这批 params 装进每个节点的 self.params。批处理(第 03 章)就靠给每个批次注入不同 params 来复用同一张子图。

代码地图(导航索引)​

主题文件符号
编排主循环pocketflow/__init__.pyFlow._orch
动作路由与终止告警pocketflow/__init__.pyFlow.get_next_node
Flow 作为节点pocketflow/__init__.pyFlow._run、Flow.post、Flow.start
节点复用的浅拷贝pocketflow/__init__.py_orch 内 copy.copy、BaseNode.set_params
嵌套/组合行为tests/test_flow_composition.pytest_flow_as_node、test_nested_flow、test_composition_with_action_propagation
终止告警行为tests/test_flow_basic.pytest_flow_ends_warning_default_missing、test_flow_ends_warning_specific_missing