【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 源码解读

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

就这一行。_exec 接收 prep 返回的列表，对每个 item 调用父类 Node 的 _exec（含重试逻辑），收集结果到新列表。items or [] 处理了 prep 返回 None 的情况——优雅的防御性编程。

2.3 实战示例：文档分块摘要

class MapSummaries(BatchNode):
    def prep(self, shared):
        content = shared["data"]
        chunk_size = 10000
        return [content[i:i+chunk_size] for i in range(0, len(content), chunk_size)]

    def exec(self, chunk):
        prompt = f"Summarize this chunk in 10 words: {chunk}"
        return call_llm(prompt)

    def post(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 源码解读

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)

三行代码。prep 返回参数字典列表，for bp in pr 逐组执行，{**self.params, **bp} 合并全局参数和当前批次参数。每次 _orch 都是从 start_node 开始跑一遍完整的 Flow。

3.3 实战示例：多文件摘要

class SummarizeFiles(BatchFlow):
    def prep(self, shared):
        return [{"fn": f} for f in shared["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 源码解读

class AsyncNode(BaseNode):
    async def prep_async(self, shared): pass
    async def exec_async(self, prep_res): pass
    async def post_async(self, shared, prep_res, exec_res): pass

    async def _exec_async(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)

    async def _run_async(self, shared):
        p = await self.prep_async(shared)
        e = await self._exec_async(p)
        return await self.post_async(shared, p, e)

注意 _exec_async 完整复刻了 Node 的重试逻辑，只是加了 await 和 asyncio.sleep。同步和异步的行为完全一致——这是 Pocket Flow "极简但不简陋"的体现。

4.4 实战示例：异步 LLM 调用

class AsyncLLMNode(AsyncNode):
    async def prep_async(self, shared):
        return shared["question"]

    async def exec_async(self, question):
        response = await async_call_llm(question)
        return response

    async def post_async(self, shared, prep_res, exec_res):
        shared["answer"] = exec_res
        return "default"

---

🔀 五、AsyncFlow：兼容同步与异步的编排器

5.1 最精巧的一行代码

AsyncFlow 的 _orch_async 方法中有一行代码，是整个异步体系最精巧的设计：

last_action = await curr._run_async(shared) if isinstance(curr, AsyncNode) else curr._run(shared)

这行代码让 AsyncFlow 可以混用同步和异步节点——遇到 AsyncNode 就 await，遇到普通 Node 就直接 run。不需要两套图定义，一个 AsyncFlow 搞定一切。

5.2 为什么这很重要？

在实际项目中，你不可能把所有节点都改成异步——有些节点是纯计算（不需要 await），有些节点是异步 I/O（必须 await）。如果框架不支持混用，你就得维护两套图，或者把所有节点都改成异步（即使不需要）。Pocket Flow 用一行 isinstance 检查，优雅地解决了这个问题。

5.3 _orch_async 完整源码

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

与同步版 _orch 唯一的区别就是那一行 isinstance 检查。其余逻辑——while 循环、copy.copy、get_next_node——完全相同。同步和异步共享同一套路由逻辑，只是执行方式不同。

---

🚀 六、并行执行：asyncio.gather 的威力

6.1 三种执行模式

Pocket Flow 提供了三种批量执行模式，覆盖从"串行"到"并行"的完整光谱：

模式一：BatchNode（顺序执行）

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

列表推导，逐个执行。简单可靠，但无法并发。

模式二：AsyncBatchNode（异步顺序执行）

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

逐个 await，但每个 item 是异步的。兼顾顺序和异步 I/O。

模式三：AsyncParallelBatchNode（异步并行执行）

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

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 也有异步和并行变体：

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

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)

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 文档