跳到主要内容

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, Flow

    class 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 主线走一遍(不进代码)​

  1. 你 Flow(start=首节点).run(shared)。
  2. Flow 从首节点开始,对它依次调 prep→exec→post。
  3. post 返回一个动作名(比如 "search",或什么都不返回即 None)。
  4. Flow 拿这个动作名去当前节点的后继表里查:连过 - "search" >> X 就跳到 X。
  5. 查不到后继 → 整场结束,run 返回最后那个动作名。

关键洞察: 节点自己只关心「我干完该报哪个动作」,完全不知道下一个是谁;「谁接谁」是在建图时用 >> 连线决定的。这层解耦就是为什么同一套原语能拼出 Agent、Workflow、RAG——换连法而已。

3. 往下读哪一章(阅读地图)​

建议按顺序读,由浅入深:

章节讲什么读完你会
01-core-abstraction.md三段式节点、>> / - 运算符建图、shared 字典——逐行读那 100 行里最基础的部分看懂一个节点怎么定义、一张图怎么连
02-flow-orchestration.mdFlow._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__.pyBaseNode、BaseNode.next、__rshift__、__sub__、_ConditionalTransition
三段式生命周期pocketflow/__init__.pyBaseNode.prep、exec、post、_run
重试与兜底pocketflow/__init__.pyNode._exec、Node.exec_fallback
编排主循环pocketflow/__init__.pyFlow._orch、Flow.get_next_node
批处理pocketflow/__init__.pyBatchNode._exec、BatchFlow._run
异步与并行pocketflow/__init__.pyAsyncNode、AsyncFlow._orch_async、AsyncParallelBatchNode._exec、AsyncParallelBatchFlow._run_async
类型契约(阅读友好)pocketflow/__init__.pyi各类的 Generic[_PrepResult, _ExecResult, _PostResult] 签名
Agent 循环示例cookbook/pocketflow-agent/flow.py、nodes.pycreate_agent_flow、DecideAction、SearchWeb、AnswerQuestion