跳到主要内容

01 · 核心抽象:三段式节点 + 有向图 + 共享字典

本章讲最底部的三块基石:一个节点内部怎么分三段执行、节点之间怎么用运算符连成图、数据怎么在节点间流动。读完你能自己定义节点、连一张图。全部代码在 pocketflow/__init__.py。

3.1 三段式节点:prep / exec / post​

它要解决的小问题​

一个 LLM 步骤天然分三段:准备输入(从状态里取出要用的东西)、真正干活(调模型/查库,这段可能失败要重试)、处理结果(写回状态、决定下一步)。如果把三者混在一个函数里,重试逻辑就会把「准备」和「写状态」也重复执行——很危险。所以 PocketFlow 强制拆开。

思路 / 直觉​

  • prep(shared):只读共享状态,挑出这步要用的数据,返回它。
  • exec(prep_res):只吃 prep 的返回值,不碰共享状态——这样它才能被安全地反复重试。
  • post(shared, prep_res, exec_res):写共享状态,并 return 一个动作名决定跳向哪。

真实实现​

BaseNode 把这三段串起来,默认实现全是空的,留给你覆写:

def prep(self,shared): pass
def exec(self,prep_res): pass
def post(self,shared,prep_res,exec_res): pass
def _exec(self,prep_res): return self.exec(prep_res)
def _run(self,shared): p=self.prep(shared); e=self._exec(p); return self.post(shared,p,e)

见 pocketflow/__init__.py:9-13。_run 就是三段流水线的胶水:prep 的结果喂给 _exec,_exec 的结果和前两者一起喂给 post,post 的返回值(动作名)向上传。

注意这里的分层:公开的三段(prep/exec/post)给你覆写;带下划线的 _exec/_run 是内部钩子,子类(Node、BatchNode、AsyncNode)靠覆写 _exec 来插入重试、批处理、异步——而不动你写的 exec。这个「公开方法 vs 内部包装」的分层是整个框架扩展性的来源,后面几章反复用到。

关键细节​

  • BaseNode.run(pocketflow/__init__.py:14-16)是给单节点测试用的便捷入口;如果这个节点已经连了后继却直接 run,会 warnings.warn("Node won't run successors. Use Flow.")——提醒你:单节点 run 不会跳转,要跳转得用 Flow。

3.2 共享字典(shared):节点间怎么传数据​

思路​

PocketFlow 不给数据流建模——没有「输入端口/输出端口」那套。节点之间传数据,靠的就是一个普通 Python 字典 shared,从头到尾同一个对象、被所有节点共享读写。

怎么用(示意,非源码)​

# 一个节点往 shared 写,后一个节点从 shared 读——约定好键名即可
class Search(Node):
    def post(self, shared, prep_res, results):
        shared["context"] = results      # 写
        return "decide"

class Decide(Node):
    def prep(self, shared):
        return shared.get("context", "")  # 读上一步写的

真实印证见 cookbook/pocketflow-agent/nodes.py:SearchWeb.post 把搜索结果拼进 shared["context"],DecideAction.prep 再把 shared["context"] 读出来喂给模型。

关键细节 / 坑​

  • shared 是你自己传进 run(shared) 的那个 dict,框架从不替你初始化里面的键。第一次用某个键前要自己 setdefault 或 if "x" not in shared(chat 示例就是这么做的)。
  • 后面会看到:shared 全程是同一个对象、从不复制;被复制的只有「节点」。这条区别是理解嵌套 Flow 的钥匙(见第 02 章)。

3.3 建图:用 >> 和 - 把节点连起来​

它要解决的小问题​

怎么让「连线」读起来像画流程图,而不是一堆 a.set_next(b)?PocketFlow 用 Python 运算符重载,让你写 a >> b(a 之后默认走 b)和 a - "search" >> b(a 报 "search" 时走 b)。

每个节点有一张「后继表」​

每个节点存一个 successors 字典:动作名 → 下一个节点。连线就是往这张表里塞一条。

def __init__(self): self.params,self.successors={},{}
def next(self,node,action="default"):
    if action in self.successors: warnings.warn(f"Overwriting successor for action '{action}'")
    self.successors[action]=node; return node

见 pocketflow/__init__.py:4-8。next 是底层 API:「当我报出 action 时,下一个是 node」。默认动作名是 "default"。重复给同一动作连线会告警。

两个运算符怎么映射到 next​

怎么读这张图:>> 直接连默认边;- 先造一个「半成品转移」记住动作名,再由它的 >> 补上目标。

  a >> b                         a - "search" >> b
  │                              │
  └▶ a.__rshift__(b)             ├▶ a.__sub__("search")
     └▶ a.next(b, "default")     │     └▶ 返回 _ConditionalTransition(a,"search")
        a.successors["default"]=b│  然后对它 >> b:
                                 └▶ _ConditionalTransition.__rshift__(b)
                                       └▶ a.next(b, "search")
                                          a.successors["search"]=b

对应源码:

def __rshift__(self,other): return self.next(other)          # a >> b
def __sub__(self,action):                                    # a - "x"
    if isinstance(action,str): return _ConditionalTransition(self,action)
    raise TypeError("Action must be a string")

class _ConditionalTransition:
    def __init__(self,src,action): self.src,self.action=src,action
    def __rshift__(self,tgt): return self.src.next(tgt,self.action)  # (a-"x") >> b

见 pocketflow/__init__.py:17-24。- "x" 单独一步不会连线,它只造一个 _ConditionalTransition 记住「源节点 + 动作名」;必须再跟一个 >> 目标 才真正落到 next。所以 a - "search" >> b 读作「a 报 search 时去 b」。

一个完整的图长什么样​

来自 cookbook/pocketflow-agent/flow.py 的 create_agent_flow——这就是一个 Agent 循环的全部连线:

decide - "search" >> search    # 决定搜索 → 去搜索
decide - "answer" >> answer    # 决定回答 → 去回答
search - "decide" >> decide    # 搜完 → 回到决定(形成循环)
return Flow(start=decide)

三条边就画出了「决策 ↔ 搜索」的循环 + 「决策 → 回答」的出口。动作名从哪来? 从 DecideAction.post 的 return exec_res["action"](模型吐出的 "search"/"answer")。图的走向由运行时的模型输出驱动——这就是「Agent」的本质。

3.4 小结:三块基石怎么合体​

  Node                     图(successors 表)              shared(dict)
  ┌──────────────┐         ┌───────────────────┐         ┌──────────────┐
  │ prep  读 ────┼────────▶│  节点A --"x"--▶ B  │         │ question: .. │
  │ exec  算     │         │  节点A --"y"--▶ C  │◀────────┤ context:  .. │
  │ post  写+动作┼────────▶│  (post 的返回值    │  读/写   │ messages: .. │
  └──────────────┘         │   决定走哪条边)    │         └──────────────┘
                           └───────────────────┘
  • 节点负责「做一件事 + 报一个动作名」,不知道下家是谁。
  • 图(每个节点的 successors)把动作名翻译成「下一个节点」。
  • shared 是横穿全场的数据总线。

下一章看 Flow 怎么拿着这三块,真正把一次运行跑起来。

代码地图(导航索引)​

主题文件符号
三段生命周期 + 胶水pocketflow/__init__.pyBaseNode.prep、exec、post、_run
单节点入口与告警pocketflow/__init__.pyBaseNode.run
后继表与连线pocketflow/__init__.pyBaseNode.next、successors
运算符建图pocketflow/__init__.py__rshift__、__sub__、_ConditionalTransition
真实建图示例cookbook/pocketflow-agent/flow.pycreate_agent_flow
节点读写 shared 示例cookbook/pocketflow-agent/nodes.pyDecideAction、SearchWeb