Agent 框架常见问题过度抽象把简单功能埋在复杂性之下实现混乱让开发者疲于应对依赖冲突和版本问题每个 AI 产品都是一张有向图节点干活边做连接一个共享字典传数据PocketFlowNode 负责做事Flow 负责连线与调度Nodeprep读数据exec执行逻辑post写回结果 决定下一步 actionclassBaseNode: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):pself.prep(shared)eself.exec(p)returnself.post(shared,p,e)设计模式Flow根据每个节点post返回的action通过successors找到下一个节点把所有节点串起来依次执行classFlow(BaseNode):def__init__(self,start):super().__init__()self.startstartdeforch(self,shared,paramsNone):# 编排逻辑currcopy.copy(self.start)whilecurr:actioncurr.run(shared)currcopy.copy(curr.successors.get(actionordefault))Flow根据条件动态调度Shared Store一个内存中的字典load_data_nodeLoadDataNode()summarize_nodeSummarizeNode()load_data_nodesummarize_node# 定义流程flowFlow(startload_data_node)shared{file_name:data.txt}flow.run(shared)扩展思想BatchNode一次处理一批输入AsyncParallelBatchNode异步并行执行提高吞吐容错max_retries失败重试exec_fallback兜底逻辑Flow 也是 Node可以嵌套在更大的 Flow 中支持构建复杂的任务图GraphAI 设计模式基于这套抽象可以实现主流的模式Agent具备自主决策能力Workflow将多个任务串联成流水线RAG数据检索与内容生成无缝集成MapReduce大规模数据处理的经典范式Structured output确保输出格式一致性Multi agent协调多个智能体共同工作