PocketFlow Hello World 实战：用 Node 与 Flow 搭建你的第一个 LLM 应用

PocketFlow Hello World 实战用 Node 与 Flow 搭建你的第一个 LLM 应用【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址: https://gitcode.com/gh_mirrors/poc/PocketFlow本篇技术指南以 PocketFlow 仓库中cookbook/pocketflow-hello-world示例为核心讲解如何从零创建一个基于 PocketFlow 的最小 LLM 问答应用你将掌握 Node 的prep → exec → post三步生命周期、共享内存shared store的数据流转方式以及如何用 Flow 编排节点并驱动一次真实的 LLM 调用。读完本文你将能独立复制这套最小骨架并在此基础上扩展出分支、循环等多节点流程。示例概览一个最小可运行的 LLM 问答应用cookbook/pocketflow-hello-world是 PocketFlow 官方仓库中难度最低的入门示例它的全部逻辑只有两个文件flow.py定义节点与流程和main.py程序入口。整个应用做的事情非常直观从共享内存shared中读取一个问题例如 In one sentence, whats the end of universe?把问题交给 LLM默认gpt-4o生成答案把答案写回共享内存并打印出来。它不包含任何复杂的多节点编排但完整覆盖了 PocketFlow 的全部核心抽象——Node、Flow、shared共享存储——因此是理解整个框架的最佳起点。项目结构该示例的目录结构如下cookbook/pocketflow-hello-world/ ├── docs/ # 设计文档design.md ├── utils/ # 工具函数call_llm.py ├── flow.py # PocketFlow 实现定义 Node 与 Flow ├── main.py # 主应用入口 └── README.md # 项目文档各文件职责flow.py定义AnswerNode节点并组装成qa_flow流程是本例的业务核心main.py初始化共享内存、调用qa_flow.run(shared)并输出结果utils/call_llm.py封装对 OpenAI Chat Completions API 的调用供节点在exec阶段使用docs/design.md设计文档模板描述需求、工具函数、流程设计与节点设计是官方推荐的先设计后编码Agentic Coding工作流产物。环境搭建与运行官方 README 给出的三步运行方式如下1. 创建虚拟环境python -m venv venv source venv/bin/activate # Windows 下使用venv\Scripts\activate2. 安装依赖pip install -r requirements.txt需要说明的是截至当前仓库快照cookbook/pocketflow-hello-world目录下并未附带requirements.txt该示例目录只包含docs/、utils/、flow.py、main.py、README.md。结合 utils/call_llm.py 的导入语句实际运行所需依赖为pocketflow框架本体本仓库中对应源码为 pocketflow/init.py也可通过pip install pocketflow安装openai用于调用 LLM。因此可直接执行pip install pocketflow openai3. 运行示例python main.py运行前务必在 utils/call_llm.py 中将api_keyYOUR_API_KEY_HERE替换为真实的 OpenAI API Key。程序输出类似Question: In one sentence, whats the end of universe? Answer: 模型生成的回答逐行拆解 flow.py定义节点与流程flow.py 是理解 PocketFlow 抽象的关键完整代码如下from pocketflow import Node, Flow from utils.call_llm import call_llm # An example node and flow # Please replace this with your own node and flow class AnswerNode(Node): def prep(self, shared): # Read question from shared return shared[question] def exec(self, question): return call_llm(question) def post(self, shared, prep_res, exec_res): # Store the answer in shared shared[answer] exec_res answer_node AnswerNode() qa_flow Flow(startanswer_node)Node最小构建单元AnswerNode继承自pocketflow.Node并实现了框架约定的三个生命周期方法。根据官方文档 docs/core_abstraction/node.md每个 Node 都按prep → exec → post三步执行prep(shared)从共享内存shared中读取并预处理数据。本例中只做一件事——取出shared[question]并返回作为prep_res传给后续步骤。exec(prep_res)执行真正的计算逻辑通常是 LLM 调用、远程 API 或工具调用。本例中把问题交给call_llm返回模型的回答。⚠️ 按框架设计原则exec应当只做计算、不要访问shared以便保持数据读写与计算分离且便于重试要求幂等。post(shared, prep_res, exec_res)把结果写回shared并可返回一个字符串作为动作Action来决定流程下一步走向。本例把回答存入shared[answer]不显式返回任何值等价于返回默认动作default。官方文档强调三个步骤都是可选的如果你只需要处理数据可以只实现prep和post。Flow编排节点图qa_flow Flow(startanswer_node)创建了一个以answer_node为起点的流程。Flow负责从一个节点出发依据每个节点post()返回的动作字符串沿边跳转直到没有下一个节点为止。关于 Flow 的编排原理_orch、get_next_node、与-运算符等下文源码级原理一节会深入展开。逐行拆解 main.py共享内存与流程驱动main.py 展示了 PocketFlow 的典型用法——共享内存shared store是节点之间传递数据的唯一通道from flow import qa_flow # Example main function # Please replace this with your own main function def main(): shared { question: In one sentence, whats the end of universe?, answer: None } qa_flow.run(shared) print(Question:, shared[question]) print(Answer:, shared[answer]) if __name__ __main__: main()关键点shared是一个普通 dict入口处预置了question与占位的answer: None。PocketFlow 没有为共享内存引入任何特殊类型任何可变容器推荐 dict都可以这保证了框架的极简性。qa_flow.run(shared)是唯一的驱动入口流程启动后shared被传入prep经exec计算再由post写回全程数据都驻留在同一个 dict 中。运行结束后读取结果flow.run()返回的是最后一个节点post()的动作字符串本例中为None即default真正的业务结果要从shared[answer]中取出。逐行拆解 utils/call_llm.py把 LLM 调用封装成工具utils/call_llm.py 将 LLM 调用独立成工具函数便于节点在exec阶段复用from 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 if __name__ __main__: prompt What is the meaning of life? print(call_llm(prompt))该文件同时提供了独立运行方式python utils/call_llm.py方便在接入 Flow 之前单独验证 LLM 连通性与返回格式。对应地docs/design.md 的 Utility Functions 一节也把call_llm列为该应用唯一的工具函数。源码级原理PocketFlow 的 100 行核心Hello World 背后的全部机制都浓缩在 pocketflow/init.py 约 100 行的源码中。理解它你就能完全掌控这个示例的每一步执行。Node 的三步调度与重试BaseNode._run实现了最基本的调用链def _run(self, shared): p self.prep(shared) e self._exec(p) return self.post(shared, p, e)而Node类在_exec中注入了重试机制class Node(BaseNode): def __init__(self, max_retries1, wait0): super().__init__() self.max_retries, self.wait max_retries, wait 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)也就是说Node构造时支持两个容错参数max_retries默认 1即不重试和wait重试前等待的秒数默认 0。当exec()抛出异常时节点会自动重试直到成功或耗尽重试次数最后一次失败会调用exec_fallback(prep_res, exc)默认行为是直接重新抛出异常也可覆写为返回兜底结果。self.cur_retry记录当前重试次数从 0 开始。例如AnswerNode(max_retries3, wait10)可以在遭遇 LLM 限流时自动退避重试。Hello World 示例未显式传参因此采用默认值不重试。Flow 的图编排Flow同样继承自BaseNode其编排逻辑在_orch中def _orch(self, shared, paramsNone): curr copy.copy(self.start_node) p params or {**self.params} last_action None while curr: curr.set_params(p) last_action curr._run(shared) curr copy.copy(self.get_next_node(curr, last_action)) return last_action流程从start_node出发反复执行当前节点并用其post()返回的动作字符串在get_next_node中查找后继节点直到找不到下一个节点为止。注意copy.copy的使用节点会被浅拷贝后执行避免共享状态污染。get_next_node的实现是def get_next_node(self, curr, action): nxt curr.successors.get(action or default) if not nxt and curr.successors: warnings.warn(fFlow ends: {action} not found in {list(curr.successors)}) return nxt这里揭示了两个重要事实动作查找规则若post()返回None会按default动作查找后继若当前节点声明了后继但找不到匹配动作流程会结束并发出UserWarning。连接语法BaseNode定义了默认转移和-命名动作转移两个运算符。node_a node_b等价于node_a.next(node_b, default)node_a - approved node_b则为指定动作建立转移。这正是 docs/core_abstraction/flow.md 中分支与循环语法的底层实现。在 Hello World 中qa_flow只有起始节点且没有后继因此qa_flow.run(shared)执行一次AnswerNode的prep → exec → post后流程即自然结束。测试印证仓库测试 tests/test_flow_basic.py 从多个角度验证了上述机制test_start_method_chaining验证start().next()链式连接与默认动作流转test_sequence_with_rshift验证线性流水线test_branching_positive/test_branching_negative验证- action 条件分支test_cycle_until_negative_ends_with_signal验证通过动作构成循环test_flow_ends_warning_*验证找不到动作时发出 Flow ends: ... not found 警告。这些测试直接佐证了本文对动作驱动编排的描述。从 Hello World 出发如何扩展成多节点流程Hello World 只有一个节点但它已经为你铺好了扩展路径。结合 docs/core_abstraction/flow.md 中给出的语法只需在flow.py中增加节点并用运算符连线即可构造任意复杂的图# 线性流水线默认动作 node_a node_b node_c # 条件分支post 返回不同动作字符串 review - approved payment review - needs_revision revise review - rejected finish revise review # 循环回到 review flow Flow(startreview)扩展时请遵循以下约定默认转移的节点其post()应当不返回任何值隐式None即default需要条件分支的节点post()必须显式返回动作字符串共享内存中预置哪些键、每个节点读写哪些键建议先在设计文档中规划清楚——docs/design.md 就是为此准备的模板它要求依次列出项目需求、工具函数、流程设计含 Mermaid 流程图、数据结构shared {key: value}以及每个节点的目的、类型与数据访问方式。小结通过cookbook/pocketflow-hello-world这个最小示例你实际上已经掌握了 PocketFlow 的全部核心概念概念在本例中的体现NodeAnswerNode实现prep / exec / post三步方法Shared Storemain.py中的shareddict节点间唯一数据通道Flowqa_flow Flow(startanswer_node)负责动作驱动的图编排LLM 工具封装utils/call_llm.py在exec阶段被调用容错机制Node(max_retries, wait)与exec_fallback本例用默认值接下来你可以阅读仓库根目录 README.md 中列出的其他 cookbook 示例如 Workflow、Agent、RAG、Batch 等与 docs/core_abstraction 系列文档把这份最小骨架扩展成真正的多节点 LLM 应用。【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址: https://gitcode.com/gh_mirrors/poc/PocketFlow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考