PocketFlow 快速入门指南

什么是 PocketFlow？

PocketFlow 是一个仅有 100 行代码的极简 LLM 框架，专为构建 AI 应用而设计。它具有以下特点：

• 轻量级：仅 100 行代码，零依赖，零厂商锁定
• 表达力强：支持多智能体、工作流、RAG 等所有你喜欢的功能

安装

pip install pocketflow

或者直接复制 源代码（仅 100 行）。

核心概念

PocketFlow 的核心抽象是图（Graph），包含两个主要组件：

1. Node（节点）

节点是执行具体任务的基本单元，包含三个生命周期方法：

• prep(shared): 准备阶段，从共享状态中获取输入
• exec(prep_res): 执行阶段，处理核心逻辑
• post(shared, prep_res, exec_res): 后处理阶段，将结果存储到共享状态

2. Flow（流程）

流程管理节点的执行顺序和数据流转。

快速开始：Hello World 示例

让我们创建一个简单的问答应用：

步骤 1：创建 LLM 调用函数

from openai import OpenAIdef call_llm(messages):client = OpenAI(api_key="YOUR_API_KEY_HERE",base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"  # 可选：使用其他 API 端点)response = client.chat.completions.create(model="qwen-turbo",  # 或使用 "gpt-4o"messages=messages,temperature=0.7)return response.choices[0].message.content

步骤 2：定义节点

from pocketflow import Node, Flowclass AnswerNode(Node):def prep(self, shared):# 从共享状态中读取问题return shared["question"]def exec(self, question):# 调用 LLM 生成答案return call_llm([{"role": "user", "content": question}])def post(self, shared, prep_res, exec_res):# 将答案存储到共享状态shared["answer"] = exec_res

步骤 3：创建流程

# 创建节点实例
answer_node = AnswerNode()# 创建流程
qa_flow = Flow(start=answer_node)

步骤 4：运行应用

def main():# 初始化共享状态shared = {"question": "用一句话解释宇宙的终极答案是什么？","answer": None}# 运行流程qa_flow.run(shared)# 输出结果print("问题:", shared["question"])print("答案:", shared["answer"])if __name__ == "__main__":main()

完整示例代码

from pocketflow import Node, Flow
from openai import OpenAIdef call_llm(messages):client = OpenAI(api_key="YOUR_API_KEY_HERE",base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")response = client.chat.completions.create(model="qwen-turbo",messages=messages,temperature=0.7)return response.choices[0].message.contentclass AnswerNode(Node):def prep(self, shared):return shared["question"]def exec(self, question):return call_llm([{"role": "user", "content": question}])def post(self, shared, prep_res, exec_res):shared["answer"] = exec_resdef main():shared = {"question": "用一句话解释宇宙的终极答案是什么？","answer": None}answer_node = AnswerNode()qa_flow = Flow(start=answer_node)qa_flow.run(shared)print("问题:", shared["question"])print("答案:", shared["answer"])if __name__ == "__main__":main()

运行结果

下一步

• 查看 官方文档
• 探索 示例项目
• 加入 Discord 社区

通过这个快速入门，你已经掌握了 PocketFlow 的基本用法。现在可以开始构建更复杂的 AI 应用了！