Pocket Flow

宋涛收录于 Agent

2026-06-01 约 1119 字预计阅读 3 分钟

今天我们来介绍第一个 Agent Framework Pocket Flow

1. Workflow 的抽象: Graph

个人认为，大多数 AI Framework 最大的价值就是提供了对 Workflow 的抽象，这个抽象基本上就是 Grap。因为 Grap 表达能力最强，能够支持很多流行的 Agent 设计模式。

无论是今天要讲的 Pocket Flow，还是后面我们会介绍的 Langgraph。我们要弄清楚这些框架，无非就是搞清楚 Grap 所要实现的以下内容:

如何表示 Graph 中节点以及节点的触发关系
如何在节点传递共享数据
Grap 被驱动执行的过程

那为什么要先介绍 Pocket Flow 呢？因为 Pocket Flow 足够简单。在 Pocket Flow 上面 Grap 对应为以下三个概念:

Node + Action
Shared Store
Flow

2. Node

Node 包含三个步骤: prep -> exec -> post

prep: 从 shared store 中读取数据，传递给 exec
exec: 执行节点的计算逻辑
post: 更新 shared store，并返回 action 决定下一步触发的节点。

节点依赖关系

节点依赖关系通过 src_node -> action -> tgt_node 来表示。代码里定义了 __sub__ 和 __rshift__ 方法，可以像下面这样，直接表达节点之间的触发关系:

1
2
3


review - "approved" >> payment        # If approved, process payment
review - "needs_revision" >> revise   # If needs changes, go to revision
review - "rejected" >> finish         # If rejected, finish the process

这种映射关系保存在 src_node 的 successors 属性中:

1
2
3
4
5


class BaseNode:
    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

3. Shared Store

Share Store 是一个自定义的全局对象，完全由用户定义。在 Flow.run 中传入，由 Flow 负责在所有节点之间共享传递。

4. Flow

Flow 驱动 Graph 执行。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


# 定义节点
load_data = LoadData()
summarize = Summarize()

# 描述节点依赖关系
load_data >> summarize

# 定义启动节点
flow = Flow(start=load_data)

# 运行流
shared = {}
flow.run(shared)

Flow 驱动 Graph 执行的核心是 run 方法:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24


class BaseNode:
    def _run(self,shared): p=self.prep(shared); e=self._exec(p); return self.post(shared,p,e)
    def run(self,shared): 
        if self.successors: warnings.warn("Node won't run successors. Use Flow.")  
        return self._run(shared)

class Flow(BaseNode):
    def __init__(self,start=None): super().__init__(); self.start_node=start
    def start(self,start): self.start_node=start; return start
    def get_next_node(self,curr,action):
        nxt=curr.successors.get(action or "default")
        if not nxt and curr.successors: warnings.warn(f"Flow ends: '{action}' not found in {list(curr.successors)}")
        return nxt
    def _orch(self,shared,params=None):
        curr,p,last_action =copy.copy(self.start_node),(params or {**self.params}),None
        # start_node 是启动节点，shared 会在所有节点之间共享传递
        # curr._run 执行 BaseNode._run 方法，
            # 内部调用 Node 定义的钩子函数 prep, exec, post
            # post 返回 action 决定下一步触发的节点
        # get_next_node 从 src_node 获取下一个执行的 node 
        while curr: curr.set_params(p); last_action=curr._run(shared); curr=copy.copy(self.get_next_node(curr,last_action))
        return last_action
    def _run(self,shared): p=self.prep(shared); o=self._orch(shared); return self.post(shared,p,o)
    def post(self,shared,prep_res,exec_res): return exec_res

Flow 继承自 BaseNode，所以 Flow 也可以作为节点。作为节点执行 Flow 的执行顺序:

Flow.prep(): 继承自 BaseNode 的空函数
Flow._run(): 执行子 Flow，返回最终的 last_action
Flow.post(): exec_res 接收的是 _run 返回的 last_action，所以直接 return exec_res，作为触发下一个节点的 action。

5. 使用示例

官网提供了很多基于 Pocket Flow 实现的 Agent 设计模式:

6. Pocket Flow 的设计理念

总结一下，Pocket Flow 本身设计非常简介，基于 Pocket Flow 构建的项目也非常清晰:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


my_project/
├── main.py
├── nodes.py
├── flow.py
├── utils/
│   ├── __init__.py
│   ├── call_llm.py
│   └── search_web.py
├── requirements.txt
└── docs/
    └── design.md

这种代码组织方式，职责清晰，和原始需求及设计完全对齐。通过把复杂的具体的业务逻辑以工具函数的方式抽象到具体的位置上，让节点的实现变得只跟数据交互相关。

虽然当时项目创建的时候， Harness 相关概念还未被提出，但是项目就已经体现了 Harness 的设计思想。

目录