【PocketFlow】

先上核心代码！

import asyncio, warnings, copy, time

class BaseNode:
    def __init__(self): self.params,self.successors={},{}
    def set_params(self,params): self.params=params
    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
    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)
    def run(self,shared): 
        if self.successors: warnings.warn("Node won't run successors. Use Flow.")  
        return self._run(shared)
    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")

class _ConditionalTransition:
    def __init__(self,src,action): self.src,self.action=src,action
    def __rshift__(self,tgt): return self.src.next(tgt,self.action)

class Node(BaseNode):
    def __init__(self,max_retries=1,wait=0): super().__init__(); self.max_retries,self.wait=max_retries,wait
    def exec_fallback(self,prep_res,exc): raise exc
    def _exec(self,prep_res):
        for self.cur_retry in range(self.max_retries):
            try: return self.exec(prep_res)
            except Exception as e:
                if self.cur_retry==self.max_retries-1: return self.exec_fallback(prep_res,e)
                if self.wait>0: time.sleep(self.wait)

class BatchNode(Node):
    def _exec(self,items): return [super(BatchNode,self)._exec(i) for i in (items or [])]

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

class BatchFlow(Flow):
    def _run(self,shared):
        pr=self.prep(shared) or []
        for bp in pr: self._orch(shared,{**self.params,**bp})
        return self.post(shared,pr,None)

class AsyncNode(Node):
    async def prep_async(self,shared): pass
    async def exec_async(self,prep_res): pass
    async def exec_fallback_async(self,prep_res,exc): raise exc
    async def post_async(self,shared,prep_res,exec_res): pass
    async def _exec(self,prep_res): 
        for self.cur_retry in range(self.max_retries):
            try: return await self.exec_async(prep_res)
            except Exception as e:
                if self.cur_retry==self.max_retries-1: return await self.exec_fallback_async(prep_res,e)
                if self.wait>0: await asyncio.sleep(self.wait)
    async def run_async(self,shared): 
        if self.successors: warnings.warn("Node won't run successors. Use AsyncFlow.")  
        return await self._run_async(shared)
    async def _run_async(self,shared): p=await self.prep_async(shared); e=await self._exec(p); return await self.post_async(shared,p,e)
    def _run(self,shared): raise RuntimeError("Use run_async.")

class AsyncBatchNode(AsyncNode,BatchNode):
    async def _exec(self,items): return [await super(AsyncBatchNode,self)._exec(i) for i in items]

class AsyncParallelBatchNode(AsyncNode,BatchNode):
    async def _exec(self,items): return await asyncio.gather(*(super(AsyncParallelBatchNode,self)._exec(i) for i in items))

class AsyncFlow(Flow,AsyncNode):
    async def _orch_async(self,shared,params=None):
        curr,p,last_action =copy.copy(self.start_node),(params or {**self.params}),None
        while curr: curr.set_params(p); last_action=await curr._run_async(shared) if isinstance(curr,AsyncNode) else curr._run(shared); curr=copy.copy(self.get_next_node(curr,last_action))
        return last_action
    async def _run_async(self,shared): p=await self.prep_async(shared); o=await self._orch_async(shared); return await self.post_async(shared,p,o)
    async def post_async(self,shared,prep_res,exec_res): return exec_res

class AsyncBatchFlow(AsyncFlow,BatchFlow):
    async def _run_async(self,shared):
        pr=await self.prep_async(shared) or []
        for bp in pr: await self._orch_async(shared,{**self.params,**bp})
        return await self.post_async(shared,pr,None)

class AsyncParallelBatchFlow(AsyncFlow,BatchFlow):
    async def _run_async(self,shared): 
        pr=await self.prep_async(shared) or []
        await asyncio.gather(*(self._orch_async(shared,{**self.params,**bp}) for bp in pr))
        return await self.post_async(shared,pr,None)

核心代码一句话总概括

这段代码 = 一个造工作流的轮子
你可以把它理解成：流水线搭建工具

• 你写一个个小任务（节点）
• 用 >> 把它们串起来
• 框架自动帮你按顺序执行、重试、批量处理、异步并行

它就是一个流程自动化引擎。

---

看懂 3 个核心概念（看懂就全懂了）

1. 节点 Node = 一个任务

比如：

• 查数据库
• 发请求
• 算个数
• 读文件

一个节点 = 干一件事。

2. 流程 Flow = 一串节点

节点1执行完 → 自动执行节点2 → 再执行节点3

3. 执行生命周期（所有节点都一样）

所有节点都固定分 3 步跑：

1. prep：准备（比如拿参数）
2. exec：真正干活（核心代码）
3. post：收尾（比如存结果）

---

逐行超通俗解释

1. BaseNode 基类（所有节点的祖宗）

class BaseNode:
    def __init__(self):
        self.params = {}    # 节点参数
        self.successors = {} # 下一个要执行谁

    def next(self, node):
        把当前节点 → 指向下一个节点
        比如 A.next(B) 就是 A 执行完跑 B

    def prep(self): 准备
    def exec(self): 干活
    def post(self): 收尾

    def run(self):
        执行 prep → exec → post

作用：定义所有节点必须有的结构。

---

2. Node 普通同步节点（最常用）

class Node(BaseNode):
    def __init__(self, max_retries=1, wait=0):
        最大重试次数
        重试等待时间

自带失败重试！
失败了会自动重试 N 次。

---

3. BatchNode 批量节点

给它一个列表，它会把列表里每一条都执行一遍。

---

4. Flow 流程（最关键！）

class Flow(BaseNode):
    它的作用：自动跑一串节点
    从起点开始 → 跑完一个自动跑下一个 → 直到结束

比如：

A >> B >> C

Flow 会自动跑：
A → B → C

---

5. 条件跳转（超级好用）

A - "success" >> B
A - "fail" >> C

意思：

• A 返回 “success” → 跑 B
• A 返回 “fail” → 跑 C

这就是流程分支判断！

---

异步部分（超简单理解）

所有带 Async 的类 = 异步版本

• 同步：一件事做完再做下一件
• 异步：多件事同时跑（不等待）

类名规律：

• AsyncNode：异步节点
• AsyncFlow：异步流程
• AsyncBatchNode：异步批量
• AsyncParallelBatchNode：异步并行（同时跑一堆任务）

---

最关键的执行逻辑（Flow 怎么跑？）

1. 从 start 节点开始
2. 执行节点
3. 看节点返回什么
4. 跳转到对应的下一个节点
5. 直到没有下一个节点，结束

完全自动化，不用你写循环和判断！

---

用一个超级小例子

# 定义节点1
class A(Node):
    def exec(self, x):
        print("我是A")
        return "next"

# 定义节点2
class B(Node):
    def exec(self, x):
        print("我是B")

# 搭建流程
flow = Flow()
a = A()
b = B()
a - "next" >> b   # A 返回 next 就执行 B
flow.start(a)

# 运行
flow.run({})

输出：

我是A
我是B

框架自动执行！

---

这个框架到底能干嘛？

你可以用它做：

• 自动化流程
• 数据处理管道
• 接口自动化
• 批量任务
• 异步并发任务
• 带重试、带分支的工作流

---

最简总结

这段代码就是：

一个轻量级流程编排引擎

• 节点 = 任务
• 流程 = 任务连线
• >> = 执行顺序
• 支持：重试、批量、异步、分支判断

你只需要写任务，框架帮你调度执行。

---

基于核心代码可根据需求拓展多智能体、agent-skill等等多方能力。