PocketFlow AsyncParallelBatchFlow 实战用 100 行级框架把图像批处理提速 8 倍【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址: https://gitcode.com/gh_mirrors/poc/PocketFlow导读本文以 PocketFlow 官方 Cookbook 中的 Parallel Image Processor 示例 为主线深入讲解如何使用AsyncParallelBatchFlow对多张图片 × 多个滤镜的批量任务做并发处理并通过与顺序执行的对照实验展示在 I/O 密集型场景下超过 8 倍的加速效果。读完本文你将掌握 AsyncParallelBatchFlow 的核心用法、与顺序版 AsyncBatchFlow 的差异以及用信号量Semaphore控制系统并发资源的工程技巧可直接迁移到 LLM 批量调用、批量文件处理等真实场景。一、示例背景图像批处理场景Cookbook 的cookbook/pocketflow-parallel-batch-flow/目录实现了一个并行图像处理器输入images/目录下的 3 张图片bird.jpg、cat.jpg、dog.jpg滤镜3 种grayscale灰度、blur模糊、sepia复古棕褐任务组合3 张图片 × 3 种滤镜 9 个独立子任务输出output/目录下的{图片名}_{滤镜名}.jpg共 9 张成品图。整个处理流的结构如下来自原文档的 Mermaid 图外层是AsyncParallelBatchFlow按图片×滤镜组合批量调度内层是加载图片 → 应用滤镜 → 保存结果的单任务 AsyncFlow节点依次为 nodes.py 中的LoadImage、ApplyFilter、SaveImage。二、运行方式与预期输出原文档给出两条命令即可跑通整个示例pip install -r requirements.txt python main.py依赖清单见 requirements.txt仅三个包pocketflow Pillow10.0.0 # 图像加载与滤镜处理 numpy1.24.0 # sepia 滤镜的矩阵运算运行后控制台会依次输出检测到的图片列表、任务组合数、每个子任务的执行日志最后给出时序对比结果。原文档记录的典型运行输出如下 Processing Images in Parallel Parallel Image Processor ------------------------------ Found 3 images: - images/bird.jpg - images/cat.jpg - images/dog.jpg Running sequential batch flow... Processing 3 images with 3 filters... Total combinations: 9 Loading image: images/bird.jpg Applying grayscale filter... Saved: output/bird_grayscale.jpg ...etc Timing Results: Sequential batch processing: 13.76 seconds Parallel batch processing: 1.71 seconds Speedup: 8.04x Processing complete! Check the output/ directory for results.说明13.76s与1.71s是该示例在某台机器上的实测值会随硬件与运行环境浮动。其数量级符合下文的分析9 个任务串行约需9 × 1.5s ≈ 13.5s而并行接近单个任务耗时≈ 1.5s。三、核心用法AsyncParallelBatchFlow 的两段式写法3.1 先定义单任务流作为内层执行单元并行批处理复用的不是批量节点而是一个完整的小流程。在 flow.py 的create_base_flow()中先用AsyncNode连接出加载→滤镜→保存三步流水线from pocketflow import AsyncParallelBatchFlow, AsyncBatchFlow from nodes import LoadImage, ApplyFilter, SaveImage def create_base_flow(): Create flow for processing a single image with one filter. load LoadImage() apply_filter ApplyFilter() save SaveImage() load - apply_filter apply_filter # 条件边post 返回 apply_filter apply_filter - save save # 条件边post 返回 save return load # 返回起始节点作为子 Flow 的 start三个节点的职责见 nodes.py都很纯粹节点prep_async取数据exec_async执行post_async落盘/流转LoadImage从self.params[image_path]取路径asyncio.sleep(0.5)模拟 I/O 后Image.open()图片写入shared[image]返回apply_filterApplyFilter取shared[image]与self.params[filter]按滤镜类型处理灰度/模糊/sepia结果写入shared[filtered_image]返回saveSaveImage拼接output/{basename}_{filter}.jpgasyncio.sleep(0.5)后image.save()打印保存路径返回default流结束每个节点exec_async中都通过asyncio.sleep(0.5)模拟真实耗时图像解码、滤镜计算、磁盘写入合计单任务约 1.5 秒这正是时序对比能拉开差距的原因。3.2 再定义批量流prep_async返回参数列表AsyncBatchFlow与AsyncParallelBatchFlow的用法几乎一致——只需要覆写prep_async返回一个参数字典列表每个字典对应一次子 Flow 的运行通过self.params传给子节点而不是放进 shared store。flow.py 中的ImageBatchFlow与ImageParallelBatchFlow因此长得几乎一样class ImageParallelBatchFlow(AsyncParallelBatchFlow): Flow that processes multiple images with multiple filters in parallel. async def prep_async(self, shared): Generate parameters for each image-filter combination. images shared.get(images, []) filters [grayscale, blur, sepia] params [] for image_path in images: for filter_type in filters: params.append({ image_path: image_path, filter: filter_type }) print(fProcessing {len(images)} images with {len(filters)} filters...) print(fTotal combinations: {len(params)}) return paramscreate_flows()把单任务流分别包进顺序版和并行版一次返回两个 Flow 供对照实验使用def create_flows(): Create the complete parallel processing flow. base_flow create_base_flow() return ImageBatchFlow(startbase_flow), ImageParallelBatchFlow(startbase_flow)3.3 关键规则为什么用self.params而不是 shared这是 BatchFlow 家族与 BatchNode 家族最本质的差别参见 batch.md 的Key Differences from BatchNode一节prep_async返回的是要传给子 Flow 的参数不是要处理的数据子节点通过self.params[image_path]、self.params[filter]读取本次组合的参数不经过 shared store每个子 Flow 用各自不同的参数独立运行子节点可以是普通 AsyncNode批量发生在 Flow 层面子 Flow 的每次运行都共享同一个 shared store因此LoadImage写入shared[image]、ApplyFilter写入shared[filtered_image]的模式在任意一次运行中都是安全的——因为每次运行是独立的参数组合且图片对象在SaveImage消费后即被下一次运行覆盖。3.4 入口 main.py对照计时main.py 负责收集图片路径、组装 shared store、并分别对顺序版和并行版计时async def main(): image_paths get_image_paths() # 扫描 images/ 下所有 jpg/jpeg/png shared {images: image_paths} batch_flow, parallel_batch_flow create_flows() start_time time.time() print(\nRunning sequential batch flow...) await batch_flow.run_async(shared) batch_time time.time() - start_time start_time time.time() print(\nRunning parallel batch flow...) await parallel_batch_flow.run_async(shared) parallel_time time.time() - start_time print(fSequential batch processing: {batch_time:.2f} seconds) print(fParallel batch processing: {parallel_time:.2f} seconds) print(fSpeedup: {batch_time/parallel_time:.2f}x)四、底层原理8 倍加速从哪来4.1 顺序版 vs 并行版的核心差异两个 Flow 的prep_async完全相同差异完全在框架内部的_run_async实现pocketflow/init.py顺序版AsyncBatchFlowclass AsyncBatchFlow(AsyncFlow, BatchFlow): async def _run_async(self, shared): pr await self.prep_async(shared) or [] for bp in pr: # 逐个 await一个跑完才跑下一个 await self._orch_async(shared, {**self.params, **bp}) return await self.post_async(shared, pr, None)并行版AsyncParallelBatchFlowclass AsyncParallelBatchFlow(AsyncFlow, BatchFlow): async def _run_async(self, shared): pr await self.prep_async(shared) or [] await asyncio.gather( # 一次性并发启动所有子 Flow *(self._orch_async(shared, {**self.params, **bp}) for bp in pr) ) return await self.post_async(shared, pr, None)从源码可以清晰看到顺序版是一个for循环逐个await总耗时 所有子任务耗时之和并行版用asyncio.gather同时发起全部子协程总耗时 ≈ 最慢的单个子任务耗时。示例中 9 个任务、每个约 1.5 秒串行约 13.5 秒并行约 1.5 秒理论加速比约 9 倍。实测 8.04x 与理论值吻合略有开销与调度损耗说明该示例每个子任务的大部分时间都在asyncio.sleep模拟的 I/O 等待上等待期间事件循环可以无缝切换到其他任务——这正是异步并发收益的来源。4.2 注意异步并行 ≠ CPU 多核并行原文档在 parallel.md 中给出了重要警告Because of Pythons GIL, parallel nodes and flows cant truly parallelize CPU-bound tasks (e.g., heavy numerical computations). However, they excel at overlapping I/O-bound work—like LLM calls, database queries, API requests, or file I/O.也就是说AsyncParallelBatchFlow的本质是asyncio 层面的并发它对 I/O 密集任务LLM 调用、数据库查询、API 请求、文件读写收益巨大但如果任务内部是纯 CPU 密集的数值计算由于 GIL 的存在并不会有真正的多核加速。本示例刻意在exec_async中加入asyncio.sleep(0.5)来模拟 I/O正是为了把这个特性直观地演示出来。4.3 使用前提任务必须相互独立并行化的前提是各子任务之间没有数据依赖。原文档强调如果每个 item 依赖前一个 item 的输出不要并行化并行调用可能很快触发 LLM 服务的速率限制需要节流机制如信号量或 sleep 间隔部分 LLM 提供单次批量推理 API比发起大量并行请求更省事、更不易触发限流。本示例中9 个图片×滤镜组合互不依赖各自读图、各自滤波、各自存盘因此可以安全并行。五、深入细节节点实现与 sepia 矩阵运算5.1 LoadImage / SaveImageI/O 边界LoadImage与SaveImage用asyncio.sleep(0.5)模拟磁盘 I/O 等待随后真正调用 PIL 完成读写。SaveImage的输出路径规则为base_name os.path.splitext(os.path.basename(self.params[image_path]))[0] output_path foutput/{base_name}_{filter_type}.jpg os.makedirs(output, exist_okTrue) # 自动建目录防止目录不存在报错仓库中cookbook/pocketflow-parallel-batch-flow/output/下已存在 9 张结果图即为该示例运行后留下的真实产物可直接对照检查。5.2 ApplyFilter三种滤镜ApplyFilter.exec_async按self.params[filter]分发三种处理if filter_type grayscale: return image.convert(L) # PIL 直接转灰度 elif filter_type blur: return image.filter(ImageFilter.BLUR) # PIL 内置模糊 elif filter_type sepia: img_array np.array(image) sepia_matrix np.array([ [0.393, 0.769, 0.189], [0.349, 0.686, 0.168], [0.272, 0.534, 0.131] ]) sepia_array img_array.dot(sepia_matrix.T) # RGB 线性变换 sepia_array np.clip(sepia_array, 0, 255).astype(np.uint8) return Image.fromarray(sepia_array)sepia 是一种经典的 RGB 线性变换把每个像素的 RGB 三元组与 3×3 权重矩阵做点积再裁剪到[0, 255]并转回uint8。这就是requirements.txt中numpy1.24.0的用途。遇到未知滤镜会抛出ValueError(fUnknown filter: {filter_type})而该异常会顺着_exec的max_retries重试机制与exec_fallback_async向上传播参见 pocketflow/init.py 中AsyncNode._exec的实现。六、测试佐证框架行为已被单元测试锁定仓库的 tests/test_async_parallel_batch_flow.py 从三个维度验证了AsyncParallelBatchFlow的语义正确性对 3 个 batch 并行翻倍后shared_storage[processed_numbers]与顺序聚合结果完全一致并行性3 个 batch、每个含 3 个delay0.1的任务串行约 0.9 秒测试断言总耗时 0.2秒——直接锁定了并行总耗时 ≈ 单个任务耗时的语义嵌套OuterAsyncParallelBatchFlow套InnerAsyncParallelBatchFlow的双层并行流验证了并行批量流支持多层嵌套对应 batch.md 的 Nested Batches 章节。该测试同时给出了一个比图像示例更贴近 LLM 场景的写法内层prep_async通过self.params[group]读取外层传入的参数并继续下发形成参数合并链。七、工程实践建议与延伸7.1 何时选择顺序、何时选择并行原文档 Key Points 给出了清晰的决策准则顺序版总耗时 所有子任务耗时之和。适合被限流的 API避免触发速率限制、对结果顺序有要求的场景并行版总耗时 ≈ 最慢单个任务耗时。适合I/O 密集、任务相互独立的场景。7.2 用信号量控制系统并发度原文档 Features 中特别提到 Manages system resources with semaphores。虽然示例本身规模较小但把asyncio.gather无限制并发迁移到真实 LLM 调用场景时必须引入节流。推荐的实现方式是在子任务内部用共享信号量包裹高并发段import asyncio semaphore asyncio.Semaphore(5) # 最多 5 个并发 async def throttled_call(image_path, filter_type): async with semaphore: return await call_llm(...) # 或执行其他受限资源操作将semaphore放进 shared store或作为模块级对象即可让AsyncParallelBatchFlow在享受并发收益的同时避免打爆 API 配额这与 parallel.md 中Beware of Rate Limits的最佳实践一致。7.3 同类示例对照仓库中另有几个可对照学习的相关示例pocketflow-batch-flow同一图像处理任务但使用顺序版AsyncBatchFlow可作为本文并行版的直接对照pocketflow-parallel-batch并行批处理翻译任务的示例pocketflow-nested-batch嵌套批量流示例配合本文嵌套测试代码一起阅读效果更佳。八、小结通过cookbook/pocketflow-parallel-batch-flow这个 100 行级框架下的图像批处理示例可以看到AsyncParallelBatchFlow的核心价值把对一个流程反复执行升级为对一个流程并发执行代码改动仅在于继承的基类不同AsyncBatchFlow→AsyncParallelBatchFlowprep_async的参数生成逻辑完全复用。框架底层用asyncio.gather一次性调度所有子 Flow使总耗时从所有任务之和降至最慢单任务实测加速约 8 倍。这套单任务 Flow 批量参数生成的模式可以平滑迁移到批量 LLM 摘要、批量文件转换、批量检索增强等任何 I/O 密集且任务独立的场景而顺序版则保留了限流友好的备选路径两者结合即可在吞吐与稳定性之间取得平衡。【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址: https://gitcode.com/gh_mirrors/poc/PocketFlow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 PocketFlow AsyncParallelBatchFlow 实战：用 100 行级框架把图像批处理提速 8 倍
PocketFlow AsyncParallelBatchFlow 实战用 100 行级框架把图像批处理提速 8 倍【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址: https://gitcode.com/gh_mirrors/poc/PocketFlow导读本文以 PocketFlow 官方 Cookbook 中的 Parallel Image Processor 示例 为主线深入讲解如何使用AsyncParallelBatchFlow对多张图片 × 多个滤镜的批量任务做并发处理并通过与顺序执行的对照实验展示在 I/O 密集型场景下超过 8 倍的加速效果。读完本文你将掌握 AsyncParallelBatchFlow 的核心用法、与顺序版 AsyncBatchFlow 的差异以及用信号量Semaphore控制系统并发资源的工程技巧可直接迁移到 LLM 批量调用、批量文件处理等真实场景。一、示例背景图像批处理场景Cookbook 的cookbook/pocketflow-parallel-batch-flow/目录实现了一个并行图像处理器输入images/目录下的 3 张图片bird.jpg、cat.jpg、dog.jpg滤镜3 种grayscale灰度、blur模糊、sepia复古棕褐任务组合3 张图片 × 3 种滤镜 9 个独立子任务输出output/目录下的{图片名}_{滤镜名}.jpg共 9 张成品图。整个处理流的结构如下来自原文档的 Mermaid 图外层是AsyncParallelBatchFlow按图片×滤镜组合批量调度内层是加载图片 → 应用滤镜 → 保存结果的单任务 AsyncFlow节点依次为 nodes.py 中的LoadImage、ApplyFilter、SaveImage。二、运行方式与预期输出原文档给出两条命令即可跑通整个示例pip install -r requirements.txt python main.py依赖清单见 requirements.txt仅三个包pocketflow Pillow10.0.0 # 图像加载与滤镜处理 numpy1.24.0 # sepia 滤镜的矩阵运算运行后控制台会依次输出检测到的图片列表、任务组合数、每个子任务的执行日志最后给出时序对比结果。原文档记录的典型运行输出如下 Processing Images in Parallel Parallel Image Processor ------------------------------ Found 3 images: - images/bird.jpg - images/cat.jpg - images/dog.jpg Running sequential batch flow... Processing 3 images with 3 filters... Total combinations: 9 Loading image: images/bird.jpg Applying grayscale filter... Saved: output/bird_grayscale.jpg ...etc Timing Results: Sequential batch processing: 13.76 seconds Parallel batch processing: 1.71 seconds Speedup: 8.04x Processing complete! Check the output/ directory for results.说明13.76s与1.71s是该示例在某台机器上的实测值会随硬件与运行环境浮动。其数量级符合下文的分析9 个任务串行约需9 × 1.5s ≈ 13.5s而并行接近单个任务耗时≈ 1.5s。三、核心用法AsyncParallelBatchFlow 的两段式写法3.1 先定义单任务流作为内层执行单元并行批处理复用的不是批量节点而是一个完整的小流程。在 flow.py 的create_base_flow()中先用AsyncNode连接出加载→滤镜→保存三步流水线from pocketflow import AsyncParallelBatchFlow, AsyncBatchFlow from nodes import LoadImage, ApplyFilter, SaveImage def create_base_flow(): Create flow for processing a single image with one filter. load LoadImage() apply_filter ApplyFilter() save SaveImage() load - apply_filter apply_filter # 条件边post 返回 apply_filter apply_filter - save save # 条件边post 返回 save return load # 返回起始节点作为子 Flow 的 start三个节点的职责见 nodes.py都很纯粹节点prep_async取数据exec_async执行post_async落盘/流转LoadImage从self.params[image_path]取路径asyncio.sleep(0.5)模拟 I/O 后Image.open()图片写入shared[image]返回apply_filterApplyFilter取shared[image]与self.params[filter]按滤镜类型处理灰度/模糊/sepia结果写入shared[filtered_image]返回saveSaveImage拼接output/{basename}_{filter}.jpgasyncio.sleep(0.5)后image.save()打印保存路径返回default流结束每个节点exec_async中都通过asyncio.sleep(0.5)模拟真实耗时图像解码、滤镜计算、磁盘写入合计单任务约 1.5 秒这正是时序对比能拉开差距的原因。3.2 再定义批量流prep_async返回参数列表AsyncBatchFlow与AsyncParallelBatchFlow的用法几乎一致——只需要覆写prep_async返回一个参数字典列表每个字典对应一次子 Flow 的运行通过self.params传给子节点而不是放进 shared store。flow.py 中的ImageBatchFlow与ImageParallelBatchFlow因此长得几乎一样class ImageParallelBatchFlow(AsyncParallelBatchFlow): Flow that processes multiple images with multiple filters in parallel. async def prep_async(self, shared): Generate parameters for each image-filter combination. images shared.get(images, []) filters [grayscale, blur, sepia] params [] for image_path in images: for filter_type in filters: params.append({ image_path: image_path, filter: filter_type }) print(fProcessing {len(images)} images with {len(filters)} filters...) print(fTotal combinations: {len(params)}) return paramscreate_flows()把单任务流分别包进顺序版和并行版一次返回两个 Flow 供对照实验使用def create_flows(): Create the complete parallel processing flow. base_flow create_base_flow() return ImageBatchFlow(startbase_flow), ImageParallelBatchFlow(startbase_flow)3.3 关键规则为什么用self.params而不是 shared这是 BatchFlow 家族与 BatchNode 家族最本质的差别参见 batch.md 的Key Differences from BatchNode一节prep_async返回的是要传给子 Flow 的参数不是要处理的数据子节点通过self.params[image_path]、self.params[filter]读取本次组合的参数不经过 shared store每个子 Flow 用各自不同的参数独立运行子节点可以是普通 AsyncNode批量发生在 Flow 层面子 Flow 的每次运行都共享同一个 shared store因此LoadImage写入shared[image]、ApplyFilter写入shared[filtered_image]的模式在任意一次运行中都是安全的——因为每次运行是独立的参数组合且图片对象在SaveImage消费后即被下一次运行覆盖。3.4 入口 main.py对照计时main.py 负责收集图片路径、组装 shared store、并分别对顺序版和并行版计时async def main(): image_paths get_image_paths() # 扫描 images/ 下所有 jpg/jpeg/png shared {images: image_paths} batch_flow, parallel_batch_flow create_flows() start_time time.time() print(\nRunning sequential batch flow...) await batch_flow.run_async(shared) batch_time time.time() - start_time start_time time.time() print(\nRunning parallel batch flow...) await parallel_batch_flow.run_async(shared) parallel_time time.time() - start_time print(fSequential batch processing: {batch_time:.2f} seconds) print(fParallel batch processing: {parallel_time:.2f} seconds) print(fSpeedup: {batch_time/parallel_time:.2f}x)四、底层原理8 倍加速从哪来4.1 顺序版 vs 并行版的核心差异两个 Flow 的prep_async完全相同差异完全在框架内部的_run_async实现pocketflow/init.py顺序版AsyncBatchFlowclass AsyncBatchFlow(AsyncFlow, BatchFlow): async def _run_async(self, shared): pr await self.prep_async(shared) or [] for bp in pr: # 逐个 await一个跑完才跑下一个 await self._orch_async(shared, {**self.params, **bp}) return await self.post_async(shared, pr, None)并行版AsyncParallelBatchFlowclass AsyncParallelBatchFlow(AsyncFlow, BatchFlow): async def _run_async(self, shared): pr await self.prep_async(shared) or [] await asyncio.gather( # 一次性并发启动所有子 Flow *(self._orch_async(shared, {**self.params, **bp}) for bp in pr) ) return await self.post_async(shared, pr, None)从源码可以清晰看到顺序版是一个for循环逐个await总耗时 所有子任务耗时之和并行版用asyncio.gather同时发起全部子协程总耗时 ≈ 最慢的单个子任务耗时。示例中 9 个任务、每个约 1.5 秒串行约 13.5 秒并行约 1.5 秒理论加速比约 9 倍。实测 8.04x 与理论值吻合略有开销与调度损耗说明该示例每个子任务的大部分时间都在asyncio.sleep模拟的 I/O 等待上等待期间事件循环可以无缝切换到其他任务——这正是异步并发收益的来源。4.2 注意异步并行 ≠ CPU 多核并行原文档在 parallel.md 中给出了重要警告Because of Pythons GIL, parallel nodes and flows cant truly parallelize CPU-bound tasks (e.g., heavy numerical computations). However, they excel at overlapping I/O-bound work—like LLM calls, database queries, API requests, or file I/O.也就是说AsyncParallelBatchFlow的本质是asyncio 层面的并发它对 I/O 密集任务LLM 调用、数据库查询、API 请求、文件读写收益巨大但如果任务内部是纯 CPU 密集的数值计算由于 GIL 的存在并不会有真正的多核加速。本示例刻意在exec_async中加入asyncio.sleep(0.5)来模拟 I/O正是为了把这个特性直观地演示出来。4.3 使用前提任务必须相互独立并行化的前提是各子任务之间没有数据依赖。原文档强调如果每个 item 依赖前一个 item 的输出不要并行化并行调用可能很快触发 LLM 服务的速率限制需要节流机制如信号量或 sleep 间隔部分 LLM 提供单次批量推理 API比发起大量并行请求更省事、更不易触发限流。本示例中9 个图片×滤镜组合互不依赖各自读图、各自滤波、各自存盘因此可以安全并行。五、深入细节节点实现与 sepia 矩阵运算5.1 LoadImage / SaveImageI/O 边界LoadImage与SaveImage用asyncio.sleep(0.5)模拟磁盘 I/O 等待随后真正调用 PIL 完成读写。SaveImage的输出路径规则为base_name os.path.splitext(os.path.basename(self.params[image_path]))[0] output_path foutput/{base_name}_{filter_type}.jpg os.makedirs(output, exist_okTrue) # 自动建目录防止目录不存在报错仓库中cookbook/pocketflow-parallel-batch-flow/output/下已存在 9 张结果图即为该示例运行后留下的真实产物可直接对照检查。5.2 ApplyFilter三种滤镜ApplyFilter.exec_async按self.params[filter]分发三种处理if filter_type grayscale: return image.convert(L) # PIL 直接转灰度 elif filter_type blur: return image.filter(ImageFilter.BLUR) # PIL 内置模糊 elif filter_type sepia: img_array np.array(image) sepia_matrix np.array([ [0.393, 0.769, 0.189], [0.349, 0.686, 0.168], [0.272, 0.534, 0.131] ]) sepia_array img_array.dot(sepia_matrix.T) # RGB 线性变换 sepia_array np.clip(sepia_array, 0, 255).astype(np.uint8) return Image.fromarray(sepia_array)sepia 是一种经典的 RGB 线性变换把每个像素的 RGB 三元组与 3×3 权重矩阵做点积再裁剪到[0, 255]并转回uint8。这就是requirements.txt中numpy1.24.0的用途。遇到未知滤镜会抛出ValueError(fUnknown filter: {filter_type})而该异常会顺着_exec的max_retries重试机制与exec_fallback_async向上传播参见 pocketflow/init.py 中AsyncNode._exec的实现。六、测试佐证框架行为已被单元测试锁定仓库的 tests/test_async_parallel_batch_flow.py 从三个维度验证了AsyncParallelBatchFlow的语义正确性对 3 个 batch 并行翻倍后shared_storage[processed_numbers]与顺序聚合结果完全一致并行性3 个 batch、每个含 3 个delay0.1的任务串行约 0.9 秒测试断言总耗时 0.2秒——直接锁定了并行总耗时 ≈ 单个任务耗时的语义嵌套OuterAsyncParallelBatchFlow套InnerAsyncParallelBatchFlow的双层并行流验证了并行批量流支持多层嵌套对应 batch.md 的 Nested Batches 章节。该测试同时给出了一个比图像示例更贴近 LLM 场景的写法内层prep_async通过self.params[group]读取外层传入的参数并继续下发形成参数合并链。七、工程实践建议与延伸7.1 何时选择顺序、何时选择并行原文档 Key Points 给出了清晰的决策准则顺序版总耗时 所有子任务耗时之和。适合被限流的 API避免触发速率限制、对结果顺序有要求的场景并行版总耗时 ≈ 最慢单个任务耗时。适合I/O 密集、任务相互独立的场景。7.2 用信号量控制系统并发度原文档 Features 中特别提到 Manages system resources with semaphores。虽然示例本身规模较小但把asyncio.gather无限制并发迁移到真实 LLM 调用场景时必须引入节流。推荐的实现方式是在子任务内部用共享信号量包裹高并发段import asyncio semaphore asyncio.Semaphore(5) # 最多 5 个并发 async def throttled_call(image_path, filter_type): async with semaphore: return await call_llm(...) # 或执行其他受限资源操作将semaphore放进 shared store或作为模块级对象即可让AsyncParallelBatchFlow在享受并发收益的同时避免打爆 API 配额这与 parallel.md 中Beware of Rate Limits的最佳实践一致。7.3 同类示例对照仓库中另有几个可对照学习的相关示例pocketflow-batch-flow同一图像处理任务但使用顺序版AsyncBatchFlow可作为本文并行版的直接对照pocketflow-parallel-batch并行批处理翻译任务的示例pocketflow-nested-batch嵌套批量流示例配合本文嵌套测试代码一起阅读效果更佳。八、小结通过cookbook/pocketflow-parallel-batch-flow这个 100 行级框架下的图像批处理示例可以看到AsyncParallelBatchFlow的核心价值把对一个流程反复执行升级为对一个流程并发执行代码改动仅在于继承的基类不同AsyncBatchFlow→AsyncParallelBatchFlowprep_async的参数生成逻辑完全复用。框架底层用asyncio.gather一次性调度所有子 Flow使总耗时从所有任务之和降至最慢单任务实测加速约 8 倍。这套单任务 Flow 批量参数生成的模式可以平滑迁移到批量 LLM 摘要、批量文件转换、批量检索增强等任何 I/O 密集且任务独立的场景而顺序版则保留了限流友好的备选路径两者结合即可在吞吐与稳定性之间取得平衡。【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址: https://gitcode.com/gh_mirrors/poc/PocketFlow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考