PocketFlow 聊天应用实战：用单节点自循环 Flow 构建带对话历史的终端 Chatbot

PocketFlow 聊天应用实战用单节点自循环 Flow 构建带对话历史的终端 Chatbot【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址: https://gitcode.com/gh_mirrors/poc/PocketFlow本篇技术指南以 cookbook/pocketflow-chat/README.md 为核心讲解如何在 PocketFlow 这一 100 行级极简 LLM 框架上实现一个完整的终端聊天应用。通过剖析ChatNode的prep → exec → post三步生命周期、Action 驱动的自循环 Flow 以及基于shared存储的对话历史维护读者将掌握 PocketFlow 最核心的图 循环抽象并能独立搭建可扩展的 CLI Chatbot 及其他有状态交互应用。一、示例概览一个最小可用的终端聊天机器人pocketflow-chat是 PocketFlow cookbook 中最基础的入门示例难度标注为 ★☆☆ Dummy它用一个节点加一条自循环边实现了完整的多轮对话体验终端下的对话式聊天界面You:/Assistant:交互完整保留整轮对话历史为 LLM 提供多轮上下文演示 PocketFlow 的 Node 与 Flow 两大核心概念代码总量约 50 行。其核心结构可用下面的 Mermaid 图表示源自原文档 README.md 的 How It Works 一节整个应用只有一个ChatNode当它返回continue动作时不断循环回自身直到用户输入exit时返回None结束对话。这种节点返回动作字符串 → Flow 依据动作路由到下一个节点的机制正是 PocketFlow 全部控制流的根基。二、快速上手安装、配置与运行按照原文档 README.md 的 Run It 一节完整的启动流程分为两步。第 1 步设置 OpenAI API Key。export OPENAI_API_KEYyour-api-key-here示例代码在调用 LLM 时会从环境变量读取密钥如果你不想用环境变量也可以直接编辑 cookbook/pocketflow-chat/utils.py把默认的your-api-key占位符替换为真实密钥。第 2 步安装依赖并运行。pip install -r requirements.txt python main.py依赖声明位于 cookbook/pocketflow-chat/requirements.txt仅有两个包pocketflow0.0.1 openai1.0.0启动后终端会输出Welcome to the chat! Type exit to end the conversation.随后即可开始多轮对话输入exit结束并打印Goodbye!。三、源码剖析ChatNode 的三步生命周期main.py 是全部应用逻辑所在核心是继承自pocketflow.Node的ChatNode。PocketFlow 规定每个 Node 都遵循prep → exec → post三步执行模型详见 docs/core_abstraction/node.mdChatNode恰好把三步职责拆解得非常清晰。3.1prep读取共享存储并获取用户输入def prep(self, shared): if messages not in shared: shared[messages] [] print(Welcome to the chat! Type exit to end the conversation.) user_input input(\nYou: ) if user_input.lower() exit: return None shared[messages].append({role: user, content: user_input}) return shared[messages]prep负责读数据与预处理首次运行时在shared存储中初始化messages列表随后阻塞等待终端输入若用户输入exit则返回None作为终止信号否则把用户消息追加进历史并把完整消息列表作为prep_res交给下一阶段。这里的shared是贯穿整个 Flow 的共享字典本示例即main()末尾创建的shared {}它是节点之间传递状态的唯一通道。文档强调prep阶段可以查询数据库、读文件、序列化数据但不应做重计算。3.2exec调用 LLM纯计算不碰 shareddef exec(self, messages): if messages is None: return None response call_llm(messages) return responseexec阶段只做计算把prep_res完整对话历史交给call_llm返回模型回复。注意这里刻意不访问shared——这是 PocketFlow 的职责分离原则exec只依赖输入参数便于重试、并发与单元测试。如果传入None用户已退出则直接短路返回None。3.3post写回结果并决定下一个动作def post(self, shared, prep_res, exec_res): if prep_res is None or exec_res is None: print(\nGoodbye!) return None print(f\nAssistant: {exec_res}) shared[messages].append({role: assistant, content: exec_res}) return continuepost负责写数据与决策把助手回复写入shared[messages]然后返回动作字符串。这里返回continue表示继续下一轮而返回None则等价于default动作并终止流程。PocketFlow 规定post返回的字符串即 Action若返回None则视为default见 docs/core_abstraction/flow.md。四、LLM 封装utils.py 与 OpenAI 调用cookbook/pocketflow-chat/utils.py 提供了极简的 OpenAI 封装from openai import OpenAI import os def call_llm(messages): client OpenAI(api_keyos.environ.get(OPENAI_API_KEY, your-api-key)) response client.chat.completions.create( modelgpt-4o, messagesmessages, temperature0.7 ) return response.choices[0].message.content三个关键点模型gpt-4o使用 OpenAI 的 Chat Completions 接口参数temperature0.7兼顾回答的连贯性与一定随机性messages直接透传 OpenAI 要求的[{role: ..., content: ...}]结构这也是为什么ChatNode中要维护role为user/assistant的消息列表自测入口if __name__ __main__分支提供了一个最小冒烟测试直接运行python utils.py即可验证 API Key 与网络连通性。五、底层原理Action 驱动与自循环 Flow 是如何运转的聊天应用的核心谜题是一个节点如何循环自己答案藏在 PocketFlow 的运算符重载与 Flow 编排器中。5.1 运算符重载-与pocketflow/init.py 中BaseNode定义了__rshift__和__sub__def __rshift__(self, other): return self.next(other) def __sub__(self, action): if isinstance(action, str): return _ConditionalTransition(self, action) raise TypeError(Action must be a string)node_a node_b等价于node_a.next(node_b)注册default动作的后继节点node_a - continue node_b经_ConditionalTransition中转等价于node_a.next(node_b, continue)注册命名动作的后继节点。于是 main.py 中的两行chat_node ChatNode() chat_node - continue chat_node # Loop back to continue conversation flow Flow(startchat_node)含义是当ChatNode.post()返回continue时下一个节点仍是它自己——自循环边由此建立。Flow(startchat_node)则指定chat_node为整个图的入口。5.2 Flow 编排器while 循环消费 ActionFlow._orch见 pocketflow/init.py是驱动这一切的引擎def _orch(self, shared, paramsNone): curr, p, last_action copy.copy(self.start_node), (params or {**self.params}), 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开始对当前节点执行_run内部即prep → exec → post拿到post返回的 Action调用get_next_node(curr, action)在curr.successors中查找与该 Action 匹配的后继节点若找到则继续循环否则结束。对本示例而言ChatNode.post()返回continue→ 查表命中自身 → 再次执行prep等待下一条输入……如此往复当用户输入exit时prep返回Noneexec短路返回Nonepost返回None此时get_next_node在successors中找不到default动作对应的节点本图只有continue一条边循环自然终止。这正是直到用户输入 exit 才结束的底层机制。get_next_node还会在动作未命中时给出警告Flow ends: {action} not found in [...]tests/test_flow_basic.py 中的test_flow_ends_warning_*两个用例专门验证了这条找不到动作即结束的兜底行为。5.3 循环与分支的通用性虽然本示例只用了一个节点但 Action 驱动的路由可以组合出任意复杂的图。同一份测试文件 tests/test_flow_basic.py 里的test_cycle_until_negative_ends_with_signal展示了一个更典型的多节点循环NumberNode → CheckPositiveNode正数分支-3后跳回检查节点负数分支跳到终止节点——与聊天机器人共享完全相同的返回 Action → 路由 → 再循环机制。也就是说掌握了本文的自循环写法就等于掌握了 PocketFlow 全部控制流顺序、分支、循环、嵌套的基石。六、工程化增强让示例走向生产基于源码事实可以低成本地把这个最小示例升级为更健壮的应用重试与退避pocketflow.Node构造器接受max_retries与wait参数默认max_retries1, wait0。把ChatNode声明为ChatNode(max_retries3, wait2)即可在exec抛异常如触发 OpenAI 限流时自动重试并等待还可覆写exec_fallback返回兜底文案而非直接崩溃参考 docs/core_abstraction/node.md 的 Fault Tolerance 一节。注意exec需保持幂等以配合重试语义。对话长度控制prep中发送的是完整历史长对话会持续膨胀 token 开销可在prep中加入截断/摘要策略或改用pocketflow-chat-memory见 cookbook/pocketflow-chat-memory/README.md中的短期长期记忆方案。消息输入模式本示例用input()阻塞读取适合演示真实产品可参考 cookbook/pocketflow-cli-hitl/README.mdCLI 人机协同或 cookbook/pocketflow-fastapi-websocket/README.mdWebSocket 实时聊天替换交互层。七、小结pocketflow-chat用约 50 行代码完整呈现了 PocketFlow 的三个核心事实Node 的三步生命周期prep读、exec算、post写、shared 共享存储多轮上下文与跨节点状态的载体、Action 驱动的图路由continue自循环实现对话轮转。作为入门第一课它是理解 cookbook 下 Agent、Workflow、RAG、Multi-Agent 等全部进阶示例的基础也是把任意有状态多轮交互需求映射为 PocketFlow 图结构的最直接范式。相关文件索引示例入口与核心逻辑cookbook/pocketflow-chat/main.pyOpenAI 封装cookbook/pocketflow-chat/utils.py依赖声明cookbook/pocketflow-chat/requirements.txt框架核心实现pocketflow/init.py核心抽象文档docs/core_abstraction/node.md、docs/core_abstraction/flow.md循环/分支行为测试tests/test_flow_basic.py【免费下载链接】PocketFlowPocket Flow: 100-line LLM framework. Let Agents build Agents!项目地址: https://gitcode.com/gh_mirrors/poc/PocketFlow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考