原创 Ace人生 2025-04-13 20:15 浙江
Pocket Flow:LLM框架清流。
def prep(self, shared): pass # 准备:从共享存储获取所需信息
def exec(self, prep_res): pass # 执行:进行特定操作
def post(self, shared, prep_res, exec_res): pass # 后处理:保存结果并决定下一步
Flow(流程):管理节点之间的流转:class Flow(BaseNode):
def orch(self, shared, params=None): # 协调节点执行
# 找到下一个节点,根据当前节点的决定移动
共享存储:一个简单的字典,用于节点间通信:shared = {"question": "谁是世界上最高的人?"}
这种简单而优雅的设计使得系统既容易理解又高度灵活。开发者可以轻松添加新节点、修改流程,或扩展功能,而不会被框架本身所限制。Pocket Flow核心组件Pocket Flow:对AI编程友好Pocket Flow不仅简化了开发过程,还特别适合与AI配合进行开发,这种方式被称为Agentic Coding(代理编码)。在这种开发模式中:人类开发者专注于高层设计和关键决策(Type 1任务)AI助手负责处理实现细节和技术执行Pocket Flow为AI编程提供了独特优势:文档即代码:Pocket Flow不依赖硬编码的应用特定封装,而是通过清晰的文档教导如何组合基础构建块。这使AI助手可以快速理解整个框架。概念简单:100行核心代码意味着AI可以完全理解整个框架,无需猜测隐藏的复杂性。模式清晰:基于图的模式非常直观,易于AI理解和生成。自动生成API封装:需要连接到特定LLM服务?AI可以即时编写10行代码的封装,无需依赖过时的内置API。这种设计理念也意味着,随着更多开发者采用Pocket Flow模式,未来的LLM将自然而然地内化这些模式,使AI辅助开发变得更加无缝。Pocket Flow:实现常用的Agentic Workflow使用Pocket Flow实现代理工作流程非常直观。基本步骤如下:定义节点:创建各种功能节点,每个节点负责特定任务决策节点:使用LLM决定下一步行动操作节点:执行具体任务(搜索、计算、生成等)输出节点:格式化和呈现最终结果连接节点:使用简单的语法将节点连接成图:# 如果决策返回"搜索",则执行搜索节点
decide - "search" >> search
# 如果决策返回"回答",则执行回答节点
decide - "answer" >> answer
设置共享存储:初始化包含输入数据的共享存储执行流程:启动工作流并获取结果flow = Flow(start=decide)
flow.run(shared)
这种方法可以实现多种常见的工作流模式:RAG系统(检索增强生成)多代理系统(多个协作代理)Map-Reduce操作(并行处理和聚合)流式处理(实时响应)监督系统(监控和纠正代理行为)Pocket Flow based agentic workflows一个例子:构建简单搜索代理让我们通过构建一个简单的网络搜索代理来展示Pocket Flow的强大和简洁。这个代理可以:Diagram of our research agent分析用户问题决定是否需要搜索网络执行搜索并收集信息根据收集的信息回答问题步骤1:定义决策节点class DecideAction(Node):
def prep(self, shared):
# 获取问题和当前上下文
context = shared.get("context", "无先前搜索")
question = shared["question"]
return question, context
def exec(self, inputs):
question, context = inputs
# 让LLM决定是搜索还是直接回答
prompt = f"""
问题: {question}
已知信息: {context}
决定: 是搜索更多信息还是直接回答?
"""
response = call_llm(prompt) # 调用语言模型
# 解析决策
return {"action": "search"if"需要搜索"in response else"answer"}
def post(self, shared, prep_res, exec_res):
if exec_res["action"] == "search":
# 生成搜索查询
shared["search_query"] = f"关于{shared['question']}的信息"
return exec_res["action"] # 返回"search"或"answer"
步骤2:定义搜索节点class SearchWeb(Node):
def prep(self, shared):
return shared["search_query"]
def exec(self, query):
# 实际中会调用搜索API
results = f"搜索'{query}'的结果: 相关信息..."
return results
def post(self, shared, prep_res, exec_res):
# 保存搜索结果
shared["context"] = exec_res
return "decide" # 返回决策节点重新评估
步骤3:定义回答节点class AnswerQuestion(Node):
def prep(self, shared):
return shared["question"], shared.get("context", "")
def exec(self, inputs):
question, context = inputs
# 生成答案
prompt = f"基于以下信息回答问题:\n问题: {question}\n信息: {context}"
return call_llm(prompt)
def post(self, shared, prep_res, exec_res):
# 保存最终答案
shared["answer"] = exec_res
return"done"# 流程结束
步骤4:连接节点并执行# 创建节点实例
decide = DecideAction()
search = SearchWeb()
answer = AnswerQuestion()
# 连接节点
decide - "search" >> search
decide - "answer" >> answer
search - "decide" >> decide
# 创建流程
flow = Flow(start=decide)
# 执行!
shared = {"question": "谁是第一个登上月球的人?"}
flow.run(shared)
print(shared["answer"])
这个简单的例子展示了代理如何动态决策、执行操作并生成答案。整个系统清晰、透明,且易于扩展。The Whole Process Visualized小结Pocket Flow通过其极简设计展示了一个重要事实:复杂的AI代理系统可以基于简单的基础构建。与大型框架相比,它提供了几个关键优势:极低的学习曲线:100行代码可以在短时间内完全理解高度灵活性:轻松定制和扩展,无框架限制完全控制:系统的每个部分都清晰可见,没有黑盒高效开发:减少依赖冲突和版本问题最重要的是,Pocket Flow变革了我们思考和构建AI代理的方式。它将复杂性还原为基础组件,让我们看到最重要的本质:LLM Agents仅仅是图。这种方法不仅使系统更容易构建和维护,还为AI开发带来了一种新范式:代理编码(Agentic Coding)—— 人类专注于高层设计,而AI助手处理实现细节。对于那些希望构建LLM应用而不希望陷入框架复杂性的开发者来说,Pocket Flow提供了一条清晰的道路:回归基础,从第一原则出发,用最小的构建块创建强大的系统。正如Pocket Flow的理念所示,有时候,少即是多。💡欢迎加入「AI行动派」,"用AI做点什么"。我在公众号「无人之路」每周更新"AI启示录",输出"学AI,用AI"的最新实践与心得。不过这只是冰山一角。在知识星球「AI行动派」中,有更多更丰富"学AI,用AI"的各种资源、技术、心得,每天更新。 最近主要集中在用AI和Agent来自动编程,实现心中的想法💡。欢迎加入,一起行动!AI行动派
