每个 AI 产品都是一张有向图

Agent 框架常见问题：

1. 过度抽象把简单功能埋在复杂性之下
2. 实现混乱让开发者疲于应对依赖冲突和版本问题

每个 AI 产品都是一张有向图

节点干活，边做连接，一个共享字典传数据

PocketFlow

Node 负责做事，Flow 负责连线与调度

Node

• prep：读数据
• exec：执行逻辑
• post：写回结果 & 决定下一步 action

classBaseNode:def__init__(self):self.params,self.successors={},{}defprep(self,shared):pass# 准备工作defexec(self,prep_res):pass# 执行任务defpost(self,shared,prep_res,exec_res):pass# 后处理defrun(self,shared):p=self.prep(shared)e=self.exec(p)returnself.post(shared,p,e)

设计模式：

Flow

• 根据每个节点post返回的action
• 通过successors找到下一个节点
• 把所有节点串起来依次执行

classFlow(BaseNode):def__init__(self,start):super().__init__()self.start=startdeforch(self,shared,params=None):# 编排逻辑curr=copy.copy(self.start)whilecurr:action=curr.run(shared)curr=copy.copy(curr.successors.get(actionor"default"))

Flow根据条件动态调度：

Shared Store

一个内存中的字典：

load_data_node=LoadDataNode()summarize_node=SummarizeNode()load_data_node>>summarize_node# 定义流程flow=Flow(start=load_data_node)shared={"file_name":"data.txt"}flow.run(shared)

扩展思想

• BatchNode

  • 一次处理一批输入

• AsyncParallelBatchNode

  • 异步并行执行，提高吞吐

• 容错

  • max_retries：失败重试
  • exec_fallback：兜底逻辑

• Flow 也是 Node

  • 可以嵌套在更大的 Flow 中
  • 支持构建复杂的任务图（Graph）

AI 设计模式

基于这套抽象，可以实现主流的模式

• Agent：具备自主决策能力
• Workflow：将多个任务串联成流水线
• RAG：数据检索与内容生成无缝集成
• MapReduce：大规模数据处理的经典范式
• Structured output：确保输出格式一致性
• Multi agent：协调多个智能体共同工作