【Pocket Flow】源码剖析（二）：批量与异步——BatchNode、AsyncNode 与并行执行
写在前面：第一篇我们拆解了 Pocket Flow 的三大核心抽象——Node、Flow 和 Shared Store，理解了 100 行代码的骨架。今天，我们进入这 100 行代码中最"有料"的部分——批量（Batch）与异步（Async）体系。Pocket Flow 只有 100 行代码，却定义了10 个类，其中 6 个与批量或异步相关。这些类不是"凑数"的——它们用极少的代码，覆盖了从"逐项串行"到"并行并发"的所有执行模式。理解了它们，你就理解了 Pocket Flow 为什么能用 100 行代码实现 MapReduce、RAG 和 Multi-Agent。
📑 文章目录
- 📌 一、为什么需要批量与异步？
- 📦 二、BatchNode：逐项执行的"数据并行"
- 🔄 三、BatchFlow：重跑流程的"任务并行"
- ⚡ 四、AsyncNode：异步 I/O 的优雅封装
- 🔀 五、AsyncFlow：兼容同步与异步的编排器
- 🚀 六、并行执行：asyncio.gather 的威力
- 🔮 七、系列预告
📌 一、为什么需要批量与异步？
1.1 两个真实场景
场景一：RAG 文档处理。你有 10 个 PDF 文件，每个需要：加载文本 → 分块 → 生成 Embedding → 存入向量库。如果逐个处理，每个文件耗时 5 秒，总共 50 秒。但如果 10 个文件并行处理，理论上只需 5 秒。
场景二：Multi-Agent 协作。两个 Agent 需要实时对话——一个出题，一个猜词。出题 Agent 等待猜词 Agent 的输入，猜词 Agent 等待出题 Agent 的提示。这是典型的异步 I/O 场景——如果用同步代码，一个 Agent 的input()会阻塞整个线程，另一个 Agent 永远得不到执行机会。
1.2 Pocket Flow 的解法
Pocket Flow 用6 个类覆盖了这两个场景：
| 类 | 场景 | 核心机制 |
|---|---|---|
| BatchNode | 同步逐项处理 | 列表推导[exec(i) for i in items] |
| BatchFlow | 同步重跑流程 | for bp in pr: _orch(shared, params) |
| AsyncNode | 异步 I/O | prep_async / exec_async / post_async |
| AsyncFlow | 异步编排 | await _orch_async()+ 兼容同步节点 |
| AsyncBatchNode | 异步逐项 | [await exec(i) for i in items] |
| AsyncParallelBatchNode | 异步并行 | asyncio.gather(*[exec(i) for i in items]) |
这 6 个类的源码加起来不到30 行，但覆盖了从"串行"到"并行"的完整光谱。
📦 二、BatchNode：逐项执行的"数据并行"
2.1 BatchNode 是什么？
BatchNode 继承自 Node，但改变了prep和exec的语义：
- prep(shared)：返回一个可迭代对象（如列表），而不是单个值
- exec(item)：对列表中的每个 item 调用一次，而不是对整个 prep 结果调用一次
- post(shared, prep_res, exec_res_list)：接收所有 exec 结果的列表
2.2 源码解读
classBatchNode(Node):def_exec(self,items):return[super(BatchNode,self)._exec(i)foriin(itemsor[])]就这一行。_exec接收prep返回的列表，对每个 item 调用父类 Node 的_exec（含重试逻辑），收集结果到新列表。items or []处理了 prep 返回 None 的情况——优雅的防御性编程。
2.3 实战示例：文档分块摘要
classMapSummaries(BatchNode):defprep(self,shared):content=shared["data"]chunk_size=10000return[content[i:i+chunk_size]foriinrange(0,len(content),chunk_size)]defexec(self,chunk):prompt=f"Summarize this chunk in 10 words:{chunk}"returncall_llm(prompt)defpost(self,shared,prep_res,exec_res_list):shared["summary"]="\n".join(exec_res_list)return"default"这个模式就是经典的Map阶段——把大任务拆成小任务，逐个处理。如果加上一个汇总节点，就是完整的MapReduce。
2.4 BatchNode 的局限
BatchNode 是顺序执行的——[exec(i) for i in items]是列表推导，一个接一个。如果每个 item 需要调用 LLM（耗时 2 秒），10 个 item 就需要 20 秒。要加速？需要异步并行——我们第六节再讲。
🔄 三、BatchFlow：重跑流程的"任务并行"
3.1 BatchFlow 是什么？
BatchNode 是"一个节点处理多个 item"。BatchFlow 是"一个流程跑多遍，每遍参数不同"。听起来相似，但本质不同：
| 维度 | BatchNode | BatchFlow |
|---|---|---|
| 粒度 | 单个 Node 的 exec | 整个 Flow 的 _orch |
| 输入 | prep 返回的 item 列表 | prep 返回的参数字典列表 |
| 状态 | 写入同一个 shared | 写入同一个 shared（注意 key 冲突） |
| 子节点 | 不涉及 | 通过 self.params 访问参数 |
| 适合 | 数据并行（同逻辑不同数据） | 任务并行（同流程不同参数） |
3.2 源码解读
classBatchFlow(Flow):def_run(self,shared):pr=self.prep(shared)or[]forbpinpr:self._orch(shared,{**self.params,**bp})returnself.post(shared,pr,None)三行代码。prep返回参数字典列表，for bp in pr逐组执行，{**self.params, **bp}合并全局参数和当前批次参数。每次_orch都是从 start_node 开始跑一遍完整的 Flow。
3.3 实战示例：多文件摘要
classSummarizeFiles(BatchFlow):defprep(self,shared):return[{"fn":f}forfinshared["files"]]summarize=SummarizeFiles(start=load_node>>summarize_node)prep返回[{"fn": "a.txt"}, {"fn": "b.txt"}, {"fn": "c.txt"}]，BatchFlow 会跑 3 遍子流程，每次self.params中有当前文件名。子节点通过self.params["fn"]访问。
3.4 BatchNode vs BatchFlow 的选择
简单规则：如果只需要一个节点处理多个 item，用 BatchNode；如果需要一串节点处理多组参数，用 BatchFlow。BatchNode 是"一把刀切多块肉"，BatchFlow 是"整条流水线跑多轮"。
⚡ 四、AsyncNode：异步 I/O 的优雅封装
4.1 为什么需要异步？
LLM 调用是 I/O 密集型操作——你发一个请求，等 2 秒收到响应。如果用同步代码，这 2 秒线程完全空闲。如果用异步代码，这 2 秒可以同时发 10 个请求，等它们一起回来。
4.2 AsyncNode 的生命周期
AsyncNode 与 Node 的生命周期完全对称，只是每个方法都加了_async后缀：
| 同步 | 异步 | 说明 |
|---|---|---|
prep(shared) | prep_async(shared) | 准备数据 |
exec(prep_res) | exec_async(prep_res) | 执行核心逻辑（LLM 调用） |
post(shared, p, e) | post_async(shared, p, e) | 后处理 + 返回 action |
_run(shared) | _run_async(shared) | 完整生命周期 |
4.3 源码解读
classAsyncNode(BaseNode):asyncdefprep_async(self,shared):passasyncdefexec_async(self,prep_res):passasyncdefpost_async(self,shared,prep_res,exec_res):passasyncdef_exec_async(self,prep_res):forself.cur_retryinrange(self.max_retries):try:returnawaitself.exec_async(prep_res)exceptExceptionase:ifself.cur_retry==self.max_retries-1:returnawaitself.exec_fallback_async(prep_res,e)ifself.wait>0:awaitasyncio.sleep(self.wait)asyncdef_run_async(self,shared):p=awaitself.prep_async(shared)e=awaitself._exec_async(p)returnawaitself.post_async(shared,p,e)注意_exec_async完整复刻了 Node 的重试逻辑，只是加了await和asyncio.sleep。同步和异步的行为完全一致——这是 Pocket Flow "极简但不简陋"的体现。
4.4 实战示例：异步 LLM 调用
classAsyncLLMNode(AsyncNode):asyncdefprep_async(self,shared):returnshared["question"]asyncdefexec_async(self,question):response=awaitasync_call_llm(question)returnresponseasyncdefpost_async(self,shared,prep_res,exec_res):shared["answer"]=exec_resreturn"default"🔀 五、AsyncFlow：兼容同步与异步的编排器
5.1 最精巧的一行代码
AsyncFlow 的_orch_async方法中有一行代码，是整个异步体系最精巧的设计：
last_action=awaitcurr._run_async(shared)ifisinstance(curr,AsyncNode)elsecurr._run(shared)这行代码让 AsyncFlow可以混用同步和异步节点——遇到 AsyncNode 就 await，遇到普通 Node 就直接 run。不需要两套图定义，一个 AsyncFlow 搞定一切。
5.2 为什么这很重要？
在实际项目中，你不可能把所有节点都改成异步——有些节点是纯计算（不需要 await），有些节点是异步 I/O（必须 await）。如果框架不支持混用，你就得维护两套图，或者把所有节点都改成异步（即使不需要）。Pocket Flow 用一行isinstance检查，优雅地解决了这个问题。
5.3 _orch_async 完整源码
asyncdef_orch_async(self,shared,params=None):curr,p,last_action=copy.copy(self.start_node),(paramsor{**self.params}),Nonewhilecurr:curr.set_params(p)last_action=awaitcurr._run_async(shared)ifisinstance(curr,AsyncNode)elsecurr._run(shared)curr=copy.copy(self.get_next_node(curr,last_action))returnlast_action与同步版_orch唯一的区别就是那一行isinstance检查。其余逻辑——while 循环、copy.copy、get_next_node——完全相同。同步和异步共享同一套路由逻辑，只是执行方式不同。
🚀 六、并行执行：asyncio.gather 的威力
6.1 三种执行模式
Pocket Flow 提供了三种批量执行模式，覆盖从"串行"到"并行"的完整光谱：
模式一：BatchNode（顺序执行）
classBatchNode(Node):def_exec(self,items):return[super()._exec(i)foriin(itemsor[])]列表推导，逐个执行。简单可靠，但无法并发。
模式二：AsyncBatchNode（异步顺序执行）
classAsyncBatchNode(AsyncNode,BatchNode):asyncdef_exec(self,items):return[awaitsuper(AsyncBatchNode,self)._exec(i)foriinitems]逐个 await，但每个 item 是异步的。兼顾顺序和异步 I/O。
模式三：AsyncParallelBatchNode（异步并行执行）
classAsyncParallelBatchNode(AsyncNode,BatchNode):asyncdef_exec(self,items):returnawaitasyncio.gather(*(super(AsyncParallelBatchNode,self)._exec(i)foriinitems))asyncio.gather同时执行所有 item。速度最快，但需注意并发限制。
6.2 性能对比
假设每个 item 耗时 2 秒（LLM 调用），10 个 item：
| 模式 | 耗时 | 适用场景 |
|---|---|---|
| BatchNode | ~20s | API 限速，必须排队 |
| AsyncBatchNode | ~20s | 异步 I/O 但需保序 |
| AsyncParallelBatchNode | ~2s | 无并发限制，追求速度 |
6.3 选型决策
简单规则：能并行就并行，不能并行就顺序，需要异步 I/O 就异步。如果你的 LLM API 有并发限制（比如 OpenAI 的 RPM），用 BatchNode 或 AsyncBatchNode；如果没有限制，用 AsyncParallelBatchNode。
6.4 BatchFlow 的异步变体
BatchFlow 也有异步和并行变体：
classAsyncBatchFlow(AsyncFlow,BatchFlow):asyncdef_run_async(self,shared):pr=awaitself.prep_async(shared)or[]forbpinpr:awaitself._orch_async(shared,{**self.params,**bp})returnawaitself.post_async(shared,pr,None)classAsyncParallelBatchFlow(AsyncFlow,BatchFlow):asyncdef_run_async(self,shared):pr=awaitself.prep_async(shared)or[]awaitasyncio.gather(*(self._orch_async(shared,{**self.params,**bp})forbpinpr))returnawaitself.post_async(shared,pr,None)AsyncBatchFlow逐组 await，AsyncParallelBatchFlow用asyncio.gather并行跑所有组。同样的模式，同样的极简实现。
🔮 七、系列预告
第三篇（也是最终篇），我们将用 Pocket Flow实战构建两个完整的 LLM 应用：
| 应用 | 用到的类 | 设计模式 |
|---|---|---|
| ReAct Agent | Node + Flow | Agent Loop + 条件路由 |
| RAG 系统 | BatchNode + Flow | MapReduce + 向量检索 |
关注我，不要错过最终篇！
🎁 总结速查卡
批量体系速查
| 类 | 执行方式 | 核心源码 | 适用场景 |
|---|---|---|---|
| BatchNode | 同步顺序 | [exec(i) for i in items] | 数据并行，API 限速 |
| BatchFlow | 同步重跑 | for bp in pr: _orch(shared, params) | 任务并行，同流程不同参数 |
| AsyncBatchNode | 异步顺序 | [await exec(i) for i in items] | 异步 I/O，需保序 |
| AsyncParallelBatchNode | 异步并行 | asyncio.gather(*[exec(i) for i in items]) | 无并发限制，追求速度 |
| AsyncBatchFlow | 异步重跑 | for bp in pr: await _orch_async() | 异步任务并行 |
| AsyncParallelBatchFlow | 异步并行重跑 | asyncio.gather(*[_orch_async() for bp in pr]) | 异步并行任务 |
异步 vs 同步速查
| 维度 | 同步 | 异步 |
|---|---|---|
| 生命周期 | prep → exec → post | prep_async → exec_async → post_async |
| 执行入口 | _run(shared) | _run_async(shared) |
| 编排循环 | _orch(shared) | _orch_async(shared) |
| 重试 | time.sleep(wait) | asyncio.sleep(wait) |
| 混用 | 不支持 | ✅ isinstance 检查兼容同步节点 |
一句话总结
Pocket Flow 的批量与异步体系用不到 30 行代码，覆盖了从"逐项串行"到"并行并发"的完整执行光谱。BatchNode 用列表推导实现数据并行，BatchFlow 用 for 循环实现任务并行，AsyncNode 用 async/await 封装异步 I/O，AsyncParallelBatchNode 用 asyncio.gather 实现真正的并行执行。最精巧的设计是 AsyncFlow 的 isinstance 检查——一行代码，让同步和异步节点可以在同一个图中混用。极简，但不简陋。
参考链接：
- Pocket Flow GitHub 仓库
- Pocket Flow Batch 文档
- Pocket Flow Async 文档
- Pocket Flow Parallel 文档
- Pocket Flow Multi-Agent 文档
