用 PocketFlow Node 构建带重试与回退机制的 LLM 文本摘要工具【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址: https://gitcode.com/gh_mirrors/poc/PocketFlow本文围绕 Pocket Flow 仓库中cookbook/pocketflow-node示例讲解如何用 PocketFlow 的Node架构实现一个健壮的 LLM 文本摘要工具通过prep/exec/exec_fallback/post四个阶段组织 LLM 任务借助max_retries自动重试、exec_fallback优雅降级并利用共享存储shared store在步骤间传递数据。读完本文你将掌握 PocketFlow 的核心 Node 抽象、重试与异常回退的底层实现原理以及如何把这个最小示例扩展成多节点的工作流。示例概览一个小而健壮的摘要工具cookbook/pocketflow-node是一个实战导向的入门示例它把 PocketFlow 的核心概念放进一个真实场景——用 LLM 对一段文本做 10 字摘要。示例刻意保持单节点、单流程的最小形态却完整覆盖了生产级 LLM 应用最常见的两个痛点自动重试调用 LLM API 失败时自动重试最多 3 次max_retries3优雅回退重试耗尽后不直接崩溃而是通过exec_fallback()返回降级文案关注点分离数据读取、LLM 调用、结果回写分别由prep/exec/post承载符合 PocketFlow 的 Node 架构共享存储输入文本与输出摘要都存放在同一个shared字典中作为节点间的数据总线。整个工程只有 5 个文件是理解 PocketFlow 最小可行形态的绝佳起点。项目结构cookbook/pocketflow-node/ ├── utils/ # 工具函数LLM API 封装 │ └── call_llm.py ├── flow.py # PocketFlow 实现Summarize Node 与 Flow ├── main.py # 应用入口准备输入、运行 Flow、打印摘要 ├── requirements.txt # 依赖清单 └── README.md # 项目文档其中 flow.py 是核心实现main.py 是可直接运行的入口call_llm.py 负责与 LLM API 通信。核心实现Summarize Node 的四阶段生命周期flow.py 中定义了一个继承自pocketflow.Node的Summarize节点完整展示了 PocketFlow 的 Node 生命周期from pocketflow import Node, Flow from utils.call_llm import call_llm class Summarize(Node): def prep(self, shared): 从共享存储中读取并预处理数据。 return shared[data] def exec(self, prep_res): 使用 LLM 执行摘要。 if not prep_res: return Empty text prompt fSummarize this text in 10 words: {prep_res} summary call_llm(prompt) # 可能失败 return summary def exec_fallback(self, shared, prep_res, exc): 提供简单的降级响应而不是崩溃。 return There was an error processing your request. def post(self, shared, prep_res, exec_res): 将摘要写回共享存储。 shared[summary] exec_res # 不返回值即默认返回 None - 走 default 转移 # 创建流程 summarize_node Summarize(max_retries3) flow Flow(startsummarize_node)四个阶段各司其职构成一次完整的节点执行阶段签名职责本示例中的行为prepprep(self, shared)从共享存储读取并预处理输入取出shared[data]作为待摘要文本execexec(self, prep_res)执行核心计算通常是 LLM 调用构造 Summarize this text in 10 words 提示词并调用 LLMexec_fallbackexec_fallback(self, shared, prep_res, exc)重试耗尽后的兜底逻辑返回固定降级文案There was an error processing your request.postpost(self, shared, prep_res, exec_res)把结果写回共享存储并决定下一步动作将摘要写入shared[summary]不返回 action值得注意的两点细节exec的防御性处理if not prep_res: return Empty text对空输入做了前置校验避免把空字符串拼进提示词浪费 token。post的返回值即动作actionPocketFlow 中post返回的字符串决定 Flow 下一步走哪条转移边。这里不返回任何值即返回None对应默认转移default边在单节点流程中意味着流程自然结束。节点间连线与条件转移虽然本示例只有单个节点但 PocketFlow 的连接语法值得了解便于后续扩展。查看 pocketflow/init.py 可以看到node1 node2等价于node1.next(node2)建立默认转移action 为defaultnode1 - action_name node2等价于node1.next(node2, action_name)建立条件转移——当node1.post()返回action_name时走向node2。因此若想让摘要失败时走一条重试/告警分支只需让post依据执行结果返回不同的 action 字符串再用- action 连出多条路径即可。底层原理max_retries 重试与 exec_fallback 回退是如何工作的示例中Summarize(max_retries3)的重试行为并非 LLM 封装层实现而是 PocketFlow 框架内建的节点能力。查看 pocketflow/init.py 中Node类的实现class Node(BaseNode): def __init__(self, max_retries1, wait0): 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)关键语义如下max_retries是总尝试次数而非额外重试次数for self.cur_retry in range(self.max_retries)意味着max_retries3时exec最多执行 3 次第 1 次 2 次重试。cur_retry从 0 开始计数因此最后一次尝试的索引是max_retries - 1。exec_fallback默认行为是抛出原异常基类默认exec_fallback直接raise exc。只有像示例这样主动覆写exec_fallback才能把崩溃变成降级返回。wait参数支持指数退避前的等待默认wait0即失败后立即重试生产场景可设置wait秒在每次失败后time.sleep(wait)降低对 API 的瞬时冲击。这一行为在仓库测试 tests/test_fall_back.py 中有完整验证test_retry_before_fallbackshould_failTrue, max_retries3时节点恰好尝试 3 次最终返回fallback证明先重试、后回退的顺序test_no_fallback_implementation未覆写exec_fallback的节点在失败时直接抛出异常印证基类默认行为test_fallback_in_flow带 fallback 的节点连入Flow后回退结果可以继续传递给下游节点证明回退机制在 Flow 编排中同样生效。从源码结构可以推断Node.run(shared)的执行链路是prep - _exec含重试/回退- post其中_exec被Node覆写以注入重试逻辑而Flow._orch则循环执行curr._run(shared)并依据post返回的 action 查找后继节点见 pocketflow/init.py 中get_next_node与_orch。这正是单节点即最小 Flow的框架设计。环境搭建与运行按 README.md 的步骤即可复现1. 创建虚拟环境python -m venv venv source venv/bin/activate # Windows 下: venv\Scripts\activate2. 安装依赖pip install -r requirements.txtrequirements.txt 内容极简仅两项pocketflow openai1.0.0pocketflow即框架本体约 100 行核心实现见 pocketflow/init.pyopenai1.0.0是 LLM 调用的 SDK。如果希望把源码直接拷贝进项目而不走 pip 安装也可以把pocketflow/__init__.py复制到你的工程目录。3. 配置 LLM API Key示例的 LLM 封装位于 call_llm.py它创建了一个 OpenAI 客户端并调用gpt-4ofrom openai import OpenAI def call_llm(prompt): client OpenAI(api_keyYOUR_API_KEY_HERE) r client.chat.completions.create( modelgpt-4o, messages[{role: user, content: prompt}] ) return r.choices[0].message.content运行时请把YOUR_API_KEY_HERE替换为真实 Key或改为读取环境变量例如os.environ[OPENAI_API_KEY]。call_llm.py同时提供了__main__入口可直接python utils/call_llm.py单独验证 LLM 连通性。4. 运行示例python main.py预期输出大致为先打印输入文本再打印Summary:后的 10 字摘要。若 API 调用持续失败则会打印Summary: There was an error processing your request.——这正是exec_fallback在起作用。使用示例替换成你自己的文本main.py 内置了一段关于 PocketFlow 自身的示例文本。要摘要你自己的内容只需修改shared字典中的data键shared {data: Your text to summarize here...} flow.run(shared) print(Summary:, shared[summary])整个数据流一目了然main.py构造shared {data: text}flow.run(shared)启动执行Summarize.prep读取文本 →exec调用 LLM →post写回流程结束后shared[summary]即为结果。由于shared是普通字典你可以自由地向其中塞入任意键值原文、元信息、中间结果多节点流程正是依靠这个共享存储 各节点各取所需的模式完成数据交换。从本例中学会的 PocketFlow 核心概念结合仓库文档 docs/core_abstraction/node.md 与 docs/core_abstraction/flow.md本例覆盖了以下框架核心Node 架构用prep/exec/post模式组织 LLM 任务数据读取、计算、回写三阶段解耦便于单独测试和复用异步场景则对应prep_async/exec_async/post_async见 pocketflow/init.py 中的AsyncNode。错误处理max_retries控制总尝试次数exec_fallback决定失败兜底策略——这是把脆弱的 LLM 调用变成可靠服务的最小手段。共享存储Shared Store以字典为数据总线在步骤间传递数据无任何状态管理负担。Flow 创建Flow(startnode)一行即可建立流程多节点时用连接默认转移、用- action 连接条件分支详见 tests/test_flow_basic.py 中的分支与循环用例。进一步扩展的方向单节点示例演示了最小闭环你可以沿以下方向自然演进仓库中均有对应 cookbook 可参考多节点流水线拆分为清洗文本 → 摘要 → 格式校验多个节点用node1 node2 node3串联条件分支在post中依据摘要质量返回不同 action让摘要失败的文本走人工复核分支批处理将节点换成BatchNode对多篇文档批量摘要参考 cookbook/pocketflow-batch-node异步化改用AsyncNodeAsyncFlow并发调用 LLM参考 cookbook/pocketflow-async-basic生产化 LLM 封装在call_llm.py中补充超时、流式输出、模型可配置等能力并让wait配合指数退避策略。总结cookbook/pocketflow-node用约 30 行业务代码把 PocketFlow 最核心的 Node 生命周期、重试回退机制和共享存储范式完整落地到一个可运行的文本摘要工具中。它既是 PocketFlow 框架100 行抽象理念的最佳注脚也是一个可以立即复制、替换文本即可上手的模板。深入阅读 flow.py、pocketflow/init.py 与 tests/test_fall_back.py即可把会用提升为理解其所以然。【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址: https://gitcode.com/gh_mirrors/poc/PocketFlow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 用 PocketFlow Node 构建带重试与回退机制的 LLM 文本摘要工具
用 PocketFlow Node 构建带重试与回退机制的 LLM 文本摘要工具【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址: https://gitcode.com/gh_mirrors/poc/PocketFlow本文围绕 Pocket Flow 仓库中cookbook/pocketflow-node示例讲解如何用 PocketFlow 的Node架构实现一个健壮的 LLM 文本摘要工具通过prep/exec/exec_fallback/post四个阶段组织 LLM 任务借助max_retries自动重试、exec_fallback优雅降级并利用共享存储shared store在步骤间传递数据。读完本文你将掌握 PocketFlow 的核心 Node 抽象、重试与异常回退的底层实现原理以及如何把这个最小示例扩展成多节点的工作流。示例概览一个小而健壮的摘要工具cookbook/pocketflow-node是一个实战导向的入门示例它把 PocketFlow 的核心概念放进一个真实场景——用 LLM 对一段文本做 10 字摘要。示例刻意保持单节点、单流程的最小形态却完整覆盖了生产级 LLM 应用最常见的两个痛点自动重试调用 LLM API 失败时自动重试最多 3 次max_retries3优雅回退重试耗尽后不直接崩溃而是通过exec_fallback()返回降级文案关注点分离数据读取、LLM 调用、结果回写分别由prep/exec/post承载符合 PocketFlow 的 Node 架构共享存储输入文本与输出摘要都存放在同一个shared字典中作为节点间的数据总线。整个工程只有 5 个文件是理解 PocketFlow 最小可行形态的绝佳起点。项目结构cookbook/pocketflow-node/ ├── utils/ # 工具函数LLM API 封装 │ └── call_llm.py ├── flow.py # PocketFlow 实现Summarize Node 与 Flow ├── main.py # 应用入口准备输入、运行 Flow、打印摘要 ├── requirements.txt # 依赖清单 └── README.md # 项目文档其中 flow.py 是核心实现main.py 是可直接运行的入口call_llm.py 负责与 LLM API 通信。核心实现Summarize Node 的四阶段生命周期flow.py 中定义了一个继承自pocketflow.Node的Summarize节点完整展示了 PocketFlow 的 Node 生命周期from pocketflow import Node, Flow from utils.call_llm import call_llm class Summarize(Node): def prep(self, shared): 从共享存储中读取并预处理数据。 return shared[data] def exec(self, prep_res): 使用 LLM 执行摘要。 if not prep_res: return Empty text prompt fSummarize this text in 10 words: {prep_res} summary call_llm(prompt) # 可能失败 return summary def exec_fallback(self, shared, prep_res, exc): 提供简单的降级响应而不是崩溃。 return There was an error processing your request. def post(self, shared, prep_res, exec_res): 将摘要写回共享存储。 shared[summary] exec_res # 不返回值即默认返回 None - 走 default 转移 # 创建流程 summarize_node Summarize(max_retries3) flow Flow(startsummarize_node)四个阶段各司其职构成一次完整的节点执行阶段签名职责本示例中的行为prepprep(self, shared)从共享存储读取并预处理输入取出shared[data]作为待摘要文本execexec(self, prep_res)执行核心计算通常是 LLM 调用构造 Summarize this text in 10 words 提示词并调用 LLMexec_fallbackexec_fallback(self, shared, prep_res, exc)重试耗尽后的兜底逻辑返回固定降级文案There was an error processing your request.postpost(self, shared, prep_res, exec_res)把结果写回共享存储并决定下一步动作将摘要写入shared[summary]不返回 action值得注意的两点细节exec的防御性处理if not prep_res: return Empty text对空输入做了前置校验避免把空字符串拼进提示词浪费 token。post的返回值即动作actionPocketFlow 中post返回的字符串决定 Flow 下一步走哪条转移边。这里不返回任何值即返回None对应默认转移default边在单节点流程中意味着流程自然结束。节点间连线与条件转移虽然本示例只有单个节点但 PocketFlow 的连接语法值得了解便于后续扩展。查看 pocketflow/init.py 可以看到node1 node2等价于node1.next(node2)建立默认转移action 为defaultnode1 - action_name node2等价于node1.next(node2, action_name)建立条件转移——当node1.post()返回action_name时走向node2。因此若想让摘要失败时走一条重试/告警分支只需让post依据执行结果返回不同的 action 字符串再用- action 连出多条路径即可。底层原理max_retries 重试与 exec_fallback 回退是如何工作的示例中Summarize(max_retries3)的重试行为并非 LLM 封装层实现而是 PocketFlow 框架内建的节点能力。查看 pocketflow/init.py 中Node类的实现class Node(BaseNode): def __init__(self, max_retries1, wait0): 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)关键语义如下max_retries是总尝试次数而非额外重试次数for self.cur_retry in range(self.max_retries)意味着max_retries3时exec最多执行 3 次第 1 次 2 次重试。cur_retry从 0 开始计数因此最后一次尝试的索引是max_retries - 1。exec_fallback默认行为是抛出原异常基类默认exec_fallback直接raise exc。只有像示例这样主动覆写exec_fallback才能把崩溃变成降级返回。wait参数支持指数退避前的等待默认wait0即失败后立即重试生产场景可设置wait秒在每次失败后time.sleep(wait)降低对 API 的瞬时冲击。这一行为在仓库测试 tests/test_fall_back.py 中有完整验证test_retry_before_fallbackshould_failTrue, max_retries3时节点恰好尝试 3 次最终返回fallback证明先重试、后回退的顺序test_no_fallback_implementation未覆写exec_fallback的节点在失败时直接抛出异常印证基类默认行为test_fallback_in_flow带 fallback 的节点连入Flow后回退结果可以继续传递给下游节点证明回退机制在 Flow 编排中同样生效。从源码结构可以推断Node.run(shared)的执行链路是prep - _exec含重试/回退- post其中_exec被Node覆写以注入重试逻辑而Flow._orch则循环执行curr._run(shared)并依据post返回的 action 查找后继节点见 pocketflow/init.py 中get_next_node与_orch。这正是单节点即最小 Flow的框架设计。环境搭建与运行按 README.md 的步骤即可复现1. 创建虚拟环境python -m venv venv source venv/bin/activate # Windows 下: venv\Scripts\activate2. 安装依赖pip install -r requirements.txtrequirements.txt 内容极简仅两项pocketflow openai1.0.0pocketflow即框架本体约 100 行核心实现见 pocketflow/init.pyopenai1.0.0是 LLM 调用的 SDK。如果希望把源码直接拷贝进项目而不走 pip 安装也可以把pocketflow/__init__.py复制到你的工程目录。3. 配置 LLM API Key示例的 LLM 封装位于 call_llm.py它创建了一个 OpenAI 客户端并调用gpt-4ofrom openai import OpenAI def call_llm(prompt): client OpenAI(api_keyYOUR_API_KEY_HERE) r client.chat.completions.create( modelgpt-4o, messages[{role: user, content: prompt}] ) return r.choices[0].message.content运行时请把YOUR_API_KEY_HERE替换为真实 Key或改为读取环境变量例如os.environ[OPENAI_API_KEY]。call_llm.py同时提供了__main__入口可直接python utils/call_llm.py单独验证 LLM 连通性。4. 运行示例python main.py预期输出大致为先打印输入文本再打印Summary:后的 10 字摘要。若 API 调用持续失败则会打印Summary: There was an error processing your request.——这正是exec_fallback在起作用。使用示例替换成你自己的文本main.py 内置了一段关于 PocketFlow 自身的示例文本。要摘要你自己的内容只需修改shared字典中的data键shared {data: Your text to summarize here...} flow.run(shared) print(Summary:, shared[summary])整个数据流一目了然main.py构造shared {data: text}flow.run(shared)启动执行Summarize.prep读取文本 →exec调用 LLM →post写回流程结束后shared[summary]即为结果。由于shared是普通字典你可以自由地向其中塞入任意键值原文、元信息、中间结果多节点流程正是依靠这个共享存储 各节点各取所需的模式完成数据交换。从本例中学会的 PocketFlow 核心概念结合仓库文档 docs/core_abstraction/node.md 与 docs/core_abstraction/flow.md本例覆盖了以下框架核心Node 架构用prep/exec/post模式组织 LLM 任务数据读取、计算、回写三阶段解耦便于单独测试和复用异步场景则对应prep_async/exec_async/post_async见 pocketflow/init.py 中的AsyncNode。错误处理max_retries控制总尝试次数exec_fallback决定失败兜底策略——这是把脆弱的 LLM 调用变成可靠服务的最小手段。共享存储Shared Store以字典为数据总线在步骤间传递数据无任何状态管理负担。Flow 创建Flow(startnode)一行即可建立流程多节点时用连接默认转移、用- action 连接条件分支详见 tests/test_flow_basic.py 中的分支与循环用例。进一步扩展的方向单节点示例演示了最小闭环你可以沿以下方向自然演进仓库中均有对应 cookbook 可参考多节点流水线拆分为清洗文本 → 摘要 → 格式校验多个节点用node1 node2 node3串联条件分支在post中依据摘要质量返回不同 action让摘要失败的文本走人工复核分支批处理将节点换成BatchNode对多篇文档批量摘要参考 cookbook/pocketflow-batch-node异步化改用AsyncNodeAsyncFlow并发调用 LLM参考 cookbook/pocketflow-async-basic生产化 LLM 封装在call_llm.py中补充超时、流式输出、模型可配置等能力并让wait配合指数退避策略。总结cookbook/pocketflow-node用约 30 行业务代码把 PocketFlow 最核心的 Node 生命周期、重试回退机制和共享存储范式完整落地到一个可运行的文本摘要工具中。它既是 PocketFlow 框架100 行抽象理念的最佳注脚也是一个可以立即复制、替换文本即可上手的模板。深入阅读 flow.py、pocketflow/init.py 与 tests/test_fall_back.py即可把会用提升为理解其所以然。【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址: https://gitcode.com/gh_mirrors/poc/PocketFlow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考