数据截至 (上游 commit f74d023f9360)
PocketFlow — 架构与原理
30 秒导读: PocketFlow 是一个只有 100 行 Python、零依赖的 LLM 应用框架。它不给你「Agent 类」「RAG 类」这些成品,而只给一套最小积木——把程序画成一张有向图:每个节点做「读数据 → 干活 → 写结果并决定下一步」三段动作,节点之间靠一个字符串「动作名」跳转。Agent、Workflow、RAG、多智能体全都是这张图的不同连法。
1. 这是什么(零基础也能懂)
-
一句话定义: PocketFlow 是一个极简的「有向图编排」框架——你把任务拆成一个个节点(Node),用箭头把它们连成图,框架负责按图一个节点接一个节点地跑。
-
解决什么问题 / 给谁用: 假设你要写一个「会上网查资料再回答」的 AI 助手。它的逻辑是个循环:先让大模型决定「该继续搜、还是可以回答了」,如果搜就去查、查完再回来问,直到能回答为止。用一堆
if/while手写会越写越乱;PocketFlow 让你把「决策」「搜索」「回答」各写成一个节点,再画三条箭头把循环连起来。给的是想自己掌控 Agent 逻辑、又不想被 LangChain 这类重框架绑架的工程师。 -
它能做什么(功能):
- 把任务拆成节点,用
>>(默认跳转)和- "动作" >>(条件跳转)连成图。 - 节点自带重试 + 兜底(
max_retries/exec_fallback)。 - 支持批处理(对一批数据重复跑)、异步、并行。
- Flow 本身也能当成一个节点塞进更大的 Flow(嵌套/组合)。
- 把任务拆成节点,用
-
用起来什么样: 下面是一个真实可跑的「自我循环」聊天机器人(来自
cookbook/pocketflow-chat/main.py),核心就三步——继承Node,连一条自环箭头,run:from pocketflow import Node, Flowclass ChatNode(Node):def prep(self, shared): # 读:拿用户输入 + 历史user = input("You: ")shared.setdefault("messages", []).append({"role": "user", "content": user})return shared["messages"]def exec(self, messages): # 干活:调大模型return call_llm(messages)def post(self, shared, prep_res, reply): # 写 + 决定下一步shared["messages"].append({"role": "assistant", "content": reply})return "continue" # 返回动作名 → 触发跳转chat = ChatNode()chat - "continue" >> chat # "continue" 这个动作跳回自己 = 循环Flow(start=chat).run({}) -
一句话直觉/类比: 把它想成流程图会自己执行。你在纸上画的「框 + 箭头」流程图,在这里就是可运行的代码——框是
Node,箭头是>>,箭头上的标签("search"/"answer")是节点post返回的动作名。
本节不碰底层。目标:你现在应能对别人讲清「PocketFlow 是拿有向图来编排 LLM 任务的极简框架」。
2. 顶层全景(它大概怎么转)
2.1 整个框架就三块
PocketFlow 的全部代码在 pocketflow/__init__.py(100 行)。它只有三个概念:
| 部件 | 干什么 | 在哪 |
|---|---|---|
| Node(节点) | 一个工作单元,分 prep→exec→post 三段执行 | pocketflow/__init__.py:3-37(BaseNode/Node) |
| Flow(流程) | 一个「编排器」:从起点节点开始,按动作名一路跳到底 | pocketflow/__init__.py:39-51(Flow._orch) |
| shared(共享字典) | 一个普通 dict,贯穿整场运行,节点靠它读写数据、互相传值 | 由调用方传入 run(shared),见 _run pocketflow/__init__.py:13 |
2.2 一次运行的数据流
怎么读这张图:从上往下是一个节点内部的三段;post 返回的动作名被 Flow 拿去查「下一 个节点」,命中就继续,查不到就结束。
┌──────────────── 一个 Node 的一生 ────────────────┐
│ │
shared ─▶│ prep(shared) 读:从共享字典取出这步要用的数据 │
(字典) │ │ prep_res │
│ ▼ │
│ exec(prep_res) 干活:调模型/查库/算东西(可重试) │
│ │ exec_res │
│ ▼ │
shared ◀─│ post(shared, 写:把结果写回字典 │
(被改写) │ prep_res, 并 return 一个「动作名」字符串 │
│ exec_res) │
└───────┬───────────────────────────────────────────┘
│ action = post 的返回值(如 "search" / None)
▼
Flow.get_next_node(当前节点, action)
│
┌─────────┴──────────┐
▼ ▼
找到后继节点 没找到
→ 跳过去,重复 → Flow 结束
2.3 主线走一遍(不进代码)
- 你
Flow(start=首节点).run(shared)。 - Flow 从首节点开始,对它依次 调
prep→exec→post。 post返回一个动作名(比如"search",或什么都不返回即None)。- Flow 拿这个动作名去当前节点的后继表里查:连过
- "search" >> X就跳到 X。 - 查不到后继 → 整场结束,
run返回最后那个动作名。
关键洞察: 节点自己只关心「我干完该报哪个动作」,完全不知道下一个是谁;「谁接谁」是在建图时用 >> 连线决定的。这层解耦就是为什么同一套原语能拼出 Agent、Workflow、RAG——换连法而已。
3. 往下读哪一章(阅读地图)
建议按顺序读,由浅入深:
| 章节 | 讲什么 | 读完你会 |
|---|---|---|
| 01-core-abstraction.md | 三段式节点、>> / - 运算符建图、shared 字典——逐行读那 100 行里最基础的部分 | 看懂一个节点怎么定义、一张图怎么连 |
| 02-flow-orchestration.md | Flow._orch 主循环:动作跳转、copy.copy 的作用、Flow 为何能当节点嵌套 | 讲清「运行时到底怎么一步步跳的」 |
| 03-reliability-batch-async.md | 重试/兜底、BatchNode/BatchFlow、Async* 与 AsyncParallel* 的实现 | 知道容错和批量/并行是怎么用几行 MRO 技巧堆出来的 |
| 04-patterns-and-boundaries.md | 用原语搭 Agent 循环/Workflow/RAG;框架刻意不做什么、坑在哪 | 会自己拼模式,也知道边界 |
4. 代码地图(导航索引)
| 主题 | 文件 | 符号 |
|---|---|---|
| 节点基类、建图运算符 | pocketflow/__init__.py | BaseNode、BaseNode.next、__rshift__、__sub__、_ConditionalTransition |
| 三段式生命周期 | pocketflow/__init__.py | BaseNode.prep、exec、post、_run |
| 重试与兜底 | pocketflow/__init__.py | Node._exec、Node.exec_fallback |
| 编排主循环 | pocketflow/__init__.py | Flow._orch、Flow.get_next_node |
| 批处理 | pocketflow/__init__.py | BatchNode._exec、BatchFlow._run |
| 异步与并行 | pocketflow/__init__.py | AsyncNode、AsyncFlow._orch_async、AsyncParallelBatchNode._exec、AsyncParallelBatchFlow._run_async |
| 类型契约(阅读友好) | pocketflow/__init__.pyi | 各类的 Generic[_PrepResult, _ExecResult, _PostResult] 签名 |
| Agent 循环示例 | cookbook/pocketflow-agent/flow.py、nodes.py | create_agent_flow、DecideAction、SearchWeb、AnswerQuestion |