Pocketflow 框架

仅需100行代码，这就是整个框架。

大多数大语言模型框架都包含数千行的抽象代码——类、适配器、供应商封装、插件系统等——但其核心思想始终如一：将若干个LLM调用串联起来，在它们之间传递状态，并处理可能出现的失败。

Pocketflow 则去除了所有这些冗余。它的核心引擎是一个单独的 TypeScript 文件，将 LLM 工作流建模为一个 嵌套有向图：节点执行具体任务，动作负责在节点间路由，流程编排整个图结构，而共享状态则将这一切连接在一起。

这是 Pocketflow 平台 背后的原始核心抽象，在该平台上，非开发者可以使用自然语言创建自定义的 AI 工作流。

---

目录

• 为何存在
• 架构
• 工作原理
• 快速入门
• 使用指南
  • 节点
  • 流程
  • 分支与循环
  • RetryNode
  • 嵌套流程
  • BatchFlow
• 设计模式
• CLI
• 项目结构
• 白皮书
• 贡献
• 许可证

---

为何存在

每个 LLM 框架最终都会归结为相同的模式：

1. 将任务分解为多个步骤
2. 允许每一步读取和写入共享上下文
3. 根据结果决定下一步操作
4. 优雅地处理失败情况

区别在于围绕这一模式有多少额外的复杂性。Pocketflow 的答案是：几乎没有。整个运行时代码约 100 行。没有供应商锁定，无需强制使用适配器，也没有配置文件。你可以自由选择自己的 LLM 客户端、数据库或其他任何工具，框架只需负责编排整个图结构。

关键特性：

• 供应商无关 — 可以调用 OpenAI、Anthropic、本地模型或 REST API，框架对此毫不在意。
• 极小的接口面积 — 仅有一个文件和四个类。你可以在10分钟内阅读并理解整个代码库。
• 嵌套式组合 — 流程可以包含其他流程。通过简单且经过测试的组件构建复杂的系统。
• 并行执行 — BatchFlow 可以同时运行多个流程实例，并自动进行扇出。
• 内置重试机制 — RetryNode 可以将任意节点包装起来，提供可配置的重试和退避策略。

---

架构

该框架将每个工作流建模为一个 嵌套有向图：

概念	作用
节点	工作的基本单元。从共享状态读取数据，执行逻辑，写回结果，并返回一个动作字符串。
动作	连接两个节点的带标签边。post() 方法返回的字符串决定了路由方向。
流程	一个编排者，从起始节点开始遍历图结构，按照动作依次执行，直到找不到后续节点为止。
共享状态	一个普通的对象，会传递给每一个节点。节点之间的通信正是通过它来实现的——无需消息传递或事件机制。

---

工作原理

每个节点都遵循相同的三步生命周期：

步骤	方法	目的
1. 准备	prep(sharedState)	从共享状态中提取数据，验证输入，并返回准备结果供下一步使用。
2. 执行	execCore(prepResult)	执行实际的工作——调用 LLM、查询数据库、进行计算等。纯逻辑操作，不对共享状态产生副作用。
3. 后处理	post(prep, exec, sharedState)	将结果写回共享状态，并返回一个动作字符串，用于选择下一个节点。

execWrapper 处于准备和执行之间，它是 RetryNode 添加重试逻辑的地方，你也可以在这里添加自己的中间件（如限流、缓存、日志记录等）。

---

快速入门

git clone https://github.com/Osly-AI/Pocket-Flow-Framework.git
cd Pocket-Flow-Framework
npm install

运行测试套件：

npm test

构建：

npm run build

---

使用指南

节点

每个节点都继承自 BaseNode，并实现三个方法：

import { BaseNode, DEFAULT_ACTION } from "pocketflow";

class SummarizeNode extends BaseNode {
  async prep(sharedState: any) {
    // 从共享状态中读取数据
    return { text: sharedState.document };
  }

  async execCore(prepResult: any) {
    // 调用你的 LLM（你需要自己提供客户端）
    const response = await openai.chat.completions.create({
      model: "gpt-4",
      messages: [{ role: "user", content: `总结：${prepResult.text}` }]
    });
    return { summary: response.choices[0].message.content };
  }

  async post(prepResult: any, execResult: any, sharedState: any) {
    // 将结果写回共享状态
    sharedState.summary = execResult.summary;
    return DEFAULT_ACTION;
  }

  _clone() { return new SummarizeNode(); }
}

流程

使用 addSuccessor 将节点连接起来，然后将其包裹在一个 Flow 中：

import { Flow, DEFAULT_ACTION } from "pocketflow";

const summarize = new SummarizeNode();
const review = new ReviewNode();
const publish = new PublishNode();

// 线性链：summarize → review → publish
summarize.addSuccessor(review, DEFAULT_ACTION);
review.addSuccessor(publish, DEFAULT_ACTION);

const pipeline = new Flow(summarize);
await pipeline.run({ document: "..." });

分支与循环

节点可以根据不同的条件返回不同的动作字符串来控制执行流程：

class QualityCheckNode extends BaseNode {
  async post(prepResult: any, execResult: any, sharedState: any) {
    if (execResult.score >= 0.8) return "approved";
    if (sharedState.retryCount < 3) return "retry";
    return "reject";
  }
  // ...
}

const check = new QualityCheckNode();
const publish = new PublishNode();
const revise = new ReviseNode();
const reject = new RejectNode();

check.addSuccessor(publish, "approved");
check.addSuccessor(revise, "retry");
check.addSuccessor(reject, "reject");

// 创建一个循环：revise 会反馈回 check
revise.addSuccessor(check, DEFAULT_ACTION);

重试节点

使用自动重试和可配置退避机制包装任何节点：

import { RetryNode } from "pocketflow";

class ResilientLLMNode extends RetryNode {
  constructor() {
    super(3, 2000); // 3次尝试，每次重试间隔2秒
  }

  async prep(sharedState: any) { /* ... */ }
  async execCore(prepResult: any) {
    // 如果这里抛出异常，RetryNode会捕获并重试
    return await callUnreliableAPI(prepResult);
  }
  async post(prep: any, exec: any, sharedState: any) { /* ... */ }
  _clone() { return new ResilientLLMNode(); }
}

嵌套流程

流程本身就是节点——因此你可以任意嵌套它们：

// 内部流程：提取 → 总结
const extract = new ExtractNode();
const summarize = new SummarizeNode();
extract.addSuccessor(summarize, DEFAULT_ACTION);
const extractAndSummarize = new Flow(extract);

// 外部流程：extractAndSummarize → 发布
const publish = new PublishNode();
extractAndSummarize.addSuccessor(publish, DEFAULT_ACTION);
const fullPipeline = new Flow(extractAndSummarize);

await fullPipeline.run(sharedState);

这是关键的设计洞察：由于 Flow 继承自 BaseNode，你可以用小型、可单独测试的流程来组合出任意复杂的系统。

批量处理流

并发处理多个项目。通过重写 prep() 方法返回一个数组——每个元素都会启动一次并行流程执行：

import { BatchFlow } from "pocketflow";

class ProcessDocuments extends BatchFlow {
  async prep(sharedState: any) {
    // 返回每个文档对应的配置——每个配置都会并行运行一次流程
    return sharedState.documents.map((doc: string) => ({ document: doc }));
  }
}

// 内部流程会针对每个文档并发执行一次
const summarize = new SummarizeNode();
const batch = new ProcessDocuments(summarize);

await batch.run({
  documents: ["paper1.pdf", "paper2.pdf", "paper3.pdf", /* …还有100份 */]
});

---

设计模式

所有常见的 LLM 模式不过是不同的图拓扑结构：

模式	图形形状	示例
链式调用	线性路径	总结 → 草拟回复
聊天	带有历史记录的自循环	聊天节点与存储的上下文循环交互
RAG	通过向量存储进行写入和读取	上传文档 → 回答问题
思维链	带有推理存储的自循环	思考 → 评估 → 再次思考
Map-Reduce	扇出 + 合并	分割文本 → 总结各块 → 归约
智能体	带分支的循环	总结 → 审核 ↔ 草拟回复
多智能体	多个带有发布/订阅机制的循环	智能体通过共享状态进行通信
监督者	带审批循环的嵌套流程	工作流程 ↔ 监督者的审批

你不需要为这些模式分别提供不同的 API。它们都只是相同的 Node + Flow + addSuccessor 原语，以不同的方式连接起来而已。

---

核心抽象

抽象	描述
节点	单步处理单元
流程	多步编排（通过节点的有向路径）
通信	用于节点间通信的共享状态存储
批量处理	对多个输入重复执行同一流程
异步	重叠 I/O 操作以实现并行执行
动作	条件路由（分支和循环）

---

命令行工具

快速搭建新组件：

npx pocket new node MyProcessor    # 创建 src/nodes/MyProcessor.node.ts
npx pocket new flow MyPipeline     # 创建 src/flows/MyPipeline.flow.ts

生成的文件包括类型化接口、生命周期存根以及内联文档注释。

---

项目结构

├── src/
│   └── pocket.ts          # 整个框架（约100行代码）
├── tests/
│   ├── pocket.test.ts     # 14个测试，覆盖率约99%
│   └── testNodes.ts       # 测试辅助节点
├── cli/                   # 代码生成命令行工具
│   ├── src/               # CLI 逻辑
│   └── templates/         # 节点和流程模板
├── examples/
│   └── mcp-addition-ts/   # MCP 服务器示例
├── docs/                  # 完整文档网站（MkDocs）
├── WHITEPAPER.md           # 技术白皮书
└── CONTRIBUTING.md         # 贡献指南

---

白皮书

如需深入了解设计哲学、架构决策以及与其他框架的比较，请参阅 WHITEPAPER.md。

---

贡献

我们欢迎贡献！请参阅 CONTRIBUTING.md 获取设置说明和贡献指南。

---

许可证

MIT

Pocket-Flow-Framework 快速上手指南

Pocket-Flow-Framework 是一个极简的 LLM 工作流编排框架，核心代码仅约 100 行。它通过“嵌套有向图”模型，将复杂的 AI 任务拆解为节点（Node）、动作（Action）和流程（Flow），无厂商锁定，支持任意 LLM 客户端。

环境准备

在开始之前，请确保您的开发环境满足以下要求：

• 操作系统：Linux, macOS 或 Windows
• 运行时：Node.js (建议 v18 或更高版本)
• 包管理器：npm 或 yarn
• 前置依赖：无需特殊系统库，仅需标准的 Node.js 环境。

提示：国内开发者若遇到 npm 安装缓慢，可临时切换至淘宝镜像源： npm config set registry https://registry.npmmirror.com

安装步骤

通过 Git 克隆项目并安装依赖：

git clone https://github.com/Osly-AI/Pocket-Flow-Framework.git
cd Pocket-Flow-Framework
npm install

验证安装是否成功（运行测试套件）：

npm test

构建项目（如需生成生产代码）：

npm run build

基本使用

Pocket-Flow 的核心逻辑是：节点处理数据 -> 返回动作字符串 -> 流程根据动作路由到下一个节点。所有节点通过共享状态对象（sharedState）通信。

1. 定义一个节点 (Node)

创建一个继承自 BaseNode 的类，实现 prep（准备数据）、execCore（执行核心逻辑）和 post（写入结果并路由）三个方法。

import { BaseNode, DEFAULT_ACTION } from "pocketflow";

class SummarizeNode extends BaseNode {
  // 1. Prep: 从共享状态读取数据
  async prep(sharedState: any) {
    return { text: sharedState.document };
  }

  // 2. ExecCore: 执行实际任务（如调用 LLM）
  async execCore(prepResult: any) {
    // 此处可替换为您自己的 LLM 客户端 (OpenAI, Anthropic, 本地模型等)
    const response = await openai.chat.completions.create({
      model: "gpt-4",
      messages: [{ role: "user", content: `Summarize: ${prepResult.text}` }]
    });
    return { summary: response.choices[0].message.content };
  }

  // 3. Post: 将结果写回共享状态，并决定下一步动作
  async post(prepResult: any, execResult: any, sharedState: any) {
    sharedState.summary = execResult.summary;
    return DEFAULT_ACTION; // 默认流向下一个连接的节点
  }

  // 必须实现 _clone 方法以支持流程复制
  _clone() { return new SummarizeNode(); }
}

2. 编排流程 (Flow)

实例化节点，使用 addSuccessor 连接它们，最后放入 Flow 中运行。

import { Flow, DEFAULT_ACTION } from "pocketflow";

// 实例化节点
const summarize = new SummarizeNode();
const review = new ReviewNode(); // 假设已定义
const publish = new PublishNode(); // 假设已定义

// 线性连接：summarize -> review -> publish
summarize.addSuccessor(review, DEFAULT_ACTION);
review.addSuccessor(publish, DEFAULT_ACTION);

// 创建流程并指定起始节点
const pipeline = new Flow(summarize);

// 运行流程，传入初始共享状态
await pipeline.run({ document: "这是需要处理的长文档内容..." });

// 运行结束后，sharedState 中将包含 summary 等结果

进阶特性简述

• 分支与循环：在 post 方法中返回不同的字符串（如 "retry", "approved"），并通过 addSuccessor(node, "actionName") 实现条件路由。
• 自动重试：继承 RetryNode 而非 BaseNode，即可为任何节点添加 configurable 的重试机制。
• 嵌套流程：Flow 本身也是一个 Node，可以将子流程嵌入主流程，构建复杂系统。
• 并行处理：使用 BatchFlow，在 prep 中返回数组，即可并发处理多个任务。