跳到主要内容

03 · 容错、批处理与异步并行

这一章看 PocketFlow 最「省」的地方:重试、批处理、异步、并行这四类能力,没有一个是新写的大模块,全是覆写内部钩子 _exec / _run + 多继承叠加堆出来的。理解了「公开 exec 不动、只换内部包装」这条主线,这十几个类就一目了然。

3.1 重试与兜底:Node._exec​

它要解决的小问题​

调 LLM/网络会偶发失败。想要「失败自动重试几次,还不行就走兜底逻辑(比如返回默认值)」,而且重试只重试 exec,不重复 prep/post。

真实实现​

Node(区别于 BaseNode)加了 max_retries / wait,并覆写 _exec 把重试包在 exec 外面:

class Node(BaseNode):
    def __init__(self,max_retries=1,wait=0): super().__init__(); self.max_retries,self.wait=max_retries,wait
    def exec_fallback(self,prep_res,exc): raise exc
    def _exec(self,prep_res):
        for self.cur_retry in range(self.max_retries):
            try: return self.exec(prep_res)
            except Exception as e:
                if self.cur_retry==self.max_retries-1: return self.exec_fallback(prep_res,e)
                if self.wait>0: time.sleep(self.wait)

见 pocketflow/__init__.py:26-34。逻辑:循环 max_retries 次调 exec;最后一次仍失败,就调 exec_fallback(默认是 raise exc——即不写兜底就把异常抛出去);否则 wait 秒后重试。cur_retry 记录当前是第几次(你的 exec 里能读它)。

关键细节​

  • 只有 Node 有重试,BaseNode 没有——第 01 章那个空壳 BaseNode._exec 直接 return self.exec(prep_res)。所以实际用都继承 Node。
  • test_fall_back.py 完整验证:test_retry_before_fallback(max_retries=3 时 exec 被调 3 次后才 fallback)、test_no_fallback_implementation(不写 fallback → 异常抛出)。
  • 这里 for self.cur_retry in ... 把循环变量存成实例属性——正是第 02 章 copy.copy 要每步给干净拷贝的原因:避免跨轮串味。

3.2 批处理:BatchNode 与 BatchFlow​

有两种「批」,别混:

类型批的是什么怎么实现
BatchNode一个节点,对一批 item 各跑一次 exec覆写 _exec,对列表逐个调父类 _exec
BatchFlow一整张子图,对一批参数 各跑一遍覆写 _run,对每份 params 调一次 _orch

BatchNode:对一列数据逐个执行​

class BatchNode(Node):
    def _exec(self,items): return [super(BatchNode,self)._exec(i) for i in (items or [])]

见 pocketflow/__init__.py:36-37。它把 prep 返回的列表里每个元素,交给 Node._exec(即带重试的那版)单独跑,收集成结果列表交给 post。所以你的 exec 只需写「处理一个 item」,prep 负责把整批切成列表。(items or []) 让 prep 返回 None 时安全成空批。

BatchFlow:对一批参数各跑一遍子图​

class BatchFlow(Flow):
    def _run(self,shared):
        pr=self.prep(shared) or []
        for bp in pr: self._orch(shared,{**self.params,**bp})
        return self.post(shared,pr,None)

见 pocketflow/__init__.py:53-57。它的 prep 返回一个参数字典列表;对每份 bp,把 {**self.params, **bp} 注入 _orch——也就是用不同参数把同一张子图重跑 N 遍。test_batch_flow.py 里 prep 返回 [{'key': k} for k in ...],子图节点用 self.params.get('key') 取到自己该处理的那份数据。注意 shared 是同一个,各批次写进同一个共享字典(通常写不同键位)。

3.3 异步:AsyncNode 与 AsyncFlow​

思路​

把三段换成 async 版:prep_async / exec_async / post_async。内部钩子 _exec、_run_async 相应地 await。

class AsyncNode(Node):
    async def _exec(self,prep_res):
        for self.cur_retry in range(self.max_retries):
            try: return await self.exec_async(prep_res)
            except Exception as e:
                if self.cur_retry==self.max_retries-1: return await self.exec_fallback_async(prep_res,e)
                if self.wait>0: await asyncio.sleep(self.wait)

见 pocketflow/__init__.py:59-74。这是 3.1 重试逻辑的 async 镜像(time.sleep→await asyncio.sleep)。AsyncNode._run 被特意改成 raise RuntimeError("Use run_async.")(:74)——防止你误用同步入口跑异步节点。

AsyncFlow 的混合编排(巧妙处)​

async def _orch_async(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=await curr._run_async(shared) if isinstance(curr,AsyncNode) else curr._run(shared); curr=copy.copy(self.get_next_node(curr,last_action))
    return last_action

见 pocketflow/__init__.py:82-88。一张图里可以同步节点和异步节点混着连:循环里对每个节点判断 isinstance(curr, AsyncNode)——是就 await curr._run_async,不是就普通 curr._run。这让你不必把每个节点都改成异步。

3.4 并行:AsyncParallelBatch* 与 MRO 技巧​

顺序批 vs 并行批​

AsyncBatchNode 是顺序 await 每个 item;AsyncParallelBatchNode 用 asyncio.gather 同时跑:

class AsyncBatchNode(AsyncNode,BatchNode):
    async def _exec(self,items): return [await super(AsyncBatchNode,self)._exec(i) for i in items]

class AsyncParallelBatchNode(AsyncNode,BatchNode):
    async def _exec(self,items): return await asyncio.gather(*(super(AsyncParallelBatchNode,self)._exec(i) for i in items))

见 pocketflow/__init__.py:76-80。两者都多继承 (AsyncNode, BatchNode),靠 super() 沿 MRO(方法解析顺序)找到 AsyncNode._exec(带 await 重试的那版)来跑单个 item。区别只在外面:一个 [... for ...] 顺序、一个 asyncio.gather(...) 并发。这就是整个框架的复用哲学——新能力 = 选对基类组合 + 覆写内部钩子,几行搞定。

并行流:AsyncParallelBatchFlow​

class AsyncParallelBatchFlow(AsyncFlow,BatchFlow):
    async def _run_async(self,shared):
        pr=await self.prep_async(shared) or []
        await asyncio.gather(*(self._orch_async(shared,{**self.params,**bp}) for bp in pr))
        return await self.post_async(shared,pr,None)

见 pocketflow/__init__.py:96-99。对比 3.2 的 BatchFlow(顺序 for),这里把「每份参数跑一遍子图」全部 gather 起来并发——cookbook 里的「8x speedup」并行图像处理就用它。

并行的隐患(诚实提醒)​

所有并行分支共享同一个 shared 字典,且各 _orch_async 交错执行。若两个分支写同一个键,就是数据竞争。安全用法是每个分支写各自不同的键(靠 params 区分,如 shared["results"][key])。框架不提供任何锁——并发安全是你的责任。

3.5 全家福:11 个类怎么组合出来的​

怎么读这张图:纵向是「同步 → 异步」,横向是「单个 → 批 → 并行批」,每格都只是覆写 _exec/_run 再多继承叠加。

                 单个              批(顺序)            批(并行)
  同步   Node ──────────▶ BatchNode
  Flow   Flow ──────────▶ BatchFlow
           │                 │
           ▼(加 async)       ▼
  异步   AsyncNode ──────▶ AsyncBatchNode ────▶ AsyncParallelBatchNode
  Flow   AsyncFlow ──────▶ AsyncBatchFlow ────▶ AsyncParallelBatchFlow
           (AsyncFlow, BatchFlow 多继承)   (gather 并发)

代码地图(导航索引)​

主题文件符号
重试 + 兜底pocketflow/__init__.pyNode._exec、Node.exec_fallback、cur_retry
批节点pocketflow/__init__.pyBatchNode._exec
批流程pocketflow/__init__.pyBatchFlow._run
异步节点 + 重试pocketflow/__init__.pyAsyncNode._exec、AsyncNode._run、exec_fallback_async
同异步混合编排pocketflow/__init__.pyAsyncFlow._orch_async
并行批(MRO)pocketflow/__init__.pyAsyncParallelBatchNode._exec、AsyncParallelBatchFlow._run_async
容错测试tests/test_fall_back.pytest_retry_before_fallback、test_no_fallback_implementation
批流程测试tests/test_batch_flow.pyTestBatchFlow、test_basic_batch_processing