pi-dynamic-workflows
来源：GitHub · ⭐ 573 stars
# pi-dynamic-workflows：用 TypeScript 重新定义动态工作流编排
## 项目简介
在现代后端系统中，工作流编排早已不是新鲜概念。从 Airflow 的 DAG 到 Temporal 的状态机，开发者们习惯了先定义流程、再执行代码的开发模式。但这种方式存在一个根本性矛盾：**业务需求的变动速度远超代码部署的频率**。当市场团队需要一个紧急的审批流程变更时，开发团队却要走完代码修改、测试、发布的完整周期，这种滞后性在快节奏的业务场景中愈发刺眼。
[pi-dynamic-workflows](https://github.com/Michaelliv/pi-dynamic-workflows) 正是瞄准这一痛点而生。这个基于 TypeScript 的轻量级框架（573 Stars）提出了一种"运行时动态编排"的范式——工作流的节点、边、条件分支乃至执行逻辑，都可以在运行时通过配置或 API 动态注入，无需重启服务即可生效。它的核心价值在于将"流程定义"从编译期解耦到运行期，让业务人员能够像搭积木一样实时调整工作流，而开发者只需专注于原子化任务节点的实现。
## 核心特性
- **完全动态的工作流拓扑**：节点与边的关系不在代码中硬编码，而是通过运行时配置构建，支持实时增删改查工作流结构，变更秒级生效。
- **类型安全的节点契约**：尽管强调动态性，框架仍借助 TypeScript 的类型系统约束节点输入输出接口，在编译期捕获类型错误，避免运行时因数据格式不匹配导致的流程崩溃。
- **内置状态持久化与恢复**：工作流执行状态自动持久化到存储后端（默认支持 Redis/PostgreSQL），节点故障后支持断点续传，满足长流程业务的可靠性要求。
- **嵌套子工作流与条件网关**：支持将复杂流程拆分为可复用的子工作流模块，同时提供基于表达式的条件网关（如 `$.amount > 1000`），实现真正的动态分支决策。
- **轻量级嵌入设计**：核心无外部强依赖，可作为库嵌入现有 Express/Fastify/NestJS 应用，而非强制引入一套独立的运行时平台。
## 技术实现
pi-dynamic-workflows 的架构设计体现了"动态性"与"工程严谨"的巧妙平衡。其底层采用**有向无环图（DAG）的惰性求值模型**：工作流在启动时仅做拓扑合法性校验（检测环路与孤立节点），真正的节点调度发生在运行时。这种设计避免了传统工作流引擎"预编译全图"的开销，也为动态修改留下空间。
框架的核心抽象是 `NodeExecutor` 接口与 `WorkflowEngine` 调度器。每个任务节点需实现统一的执行契约：
interface NodeExecutor<TInput, TOutput> {
execute(ctx: ExecutionContext<TInput>): Promise<TOutput>;
// 声明输入输出 Schema，用于运行时类型校验
schema: {
input: ZodSchema<TInput>;
output: ZodSchema<TOutput>;
};
}
值得注意的是，框架选择 **Zod** 而非 JSON Schema 作为运行时校验工具，这一决策极具 TypeScript 生态特色。Zod 的 Schema 定义本身就是类型安全的，编译期类型与运行期校验天然同构，避免了"类型定义写一套、校验规则再写一套"的重复劳动。
调度层采用**事件驱动的状态机模型**。`WorkflowEngine` 维护每个工作流实例的 `ExecutionState`，节点完成后触发下游节点的调度事件。状态持久化通过可插拔的 `StateStore` 接口实现，默认的 Redis 实现利用 Lua 脚本保证状态变更的原子性，防止分布式场景下的竞态条件。对于需要人工审批的暂停节点，框架实现了基于 **Promise 的异步唤醒机制**——工作流实例在内存中挂起等待外部信号，超时后自动回滚或告警，而非占用线程轮询。
动态修改的安全性通过**版本隔离**保障：工作流定义变更时生成新版本号，正在运行的实例仍按旧版本执行，新实例采用最新定义，避免"中途换引擎"导致的状态不一致。
## 快速上手
以下示例展示如何动态注册一个简单的工作流并触发执行：
import { WorkflowEngine, InMemoryStateStore } from 'pi-dynamic-workflows';
import { z } from 'zod';
// 1. 初始化引擎
const engine = new WorkflowEngine({
stateStore: new InMemoryStateStore(), // 生产环境换为 RedisStore
});
// 2. 定义原子任务节点
const validateOrder = {
id: 'validate',
execute: async (ctx) => {
const { userId, amount } = ctx.input;
if (amount <= 0) throw new Error('Invalid amount');
return { valid: true, riskScore: amount > 10000 ? 'high' : 'low' };
},
schema: {
input: z.object({ userId: z.string(), amount: z.number() }),
output: z.object({ valid: z.boolean(), riskScore: z.enum(['low', 'high']) }),
},
};
const processPayment = {
id: 'payment',
execute: async (ctx) => {
console.log(Processing payment for ${ctx.input.userId});
return { transactionId: crypto.randomUUID() };
},
schema: {
input: z.object({ userId: z.string(), riskScore: z.string() }),
output: z.object({ transactionId: z.string() }),
},
};
// 3. 动态注册工作流（可在运行时通过 API 调用）
await engine.registerWorkflow({
id: 'order-flow',
nodes: [validateOrder, processPayment],
edges: [
{ from: 'start', to: 'validate' },
{ from: 'validate', to: 'payment', condition: '$.riskScore !== "high"' },
{ from: 'validate', to: 'manual-review', condition: '$.riskScore === "high"' },
],
});
// 4. 触发执行
const result = await engine.start('order-flow', {
userId: 'user_123',
amount: 500,
});
## 应用场景
**电商平台的促销规则引擎**：大促期间，满减规则、叠加策略、库存预占逻辑可能每小时调整。运营团队通过后台直接修改工作流配置，无需发布代码即可上线新玩法，活动结束后再回滚至常规流程。
**SaaS 产品的客户定制化审批**：不同企业客户的审批层级、抄送规则、会签/或签策略各异。传统做法是为每个客户维护分支代码，而 pi-dynamic-workflows 允许将审批模板存储为租户级别的配置，同一套服务支撑千企千面。
**AI Agent 的链式调用编排**：在 LLM 应用中，ReAct、CoT 等推理模式的步骤组合经常需要实验调优。开发者可动态调整工具调用顺序、反思循环次数，甚至 A/B 测试不同的链式策略，快速验证效果。
## 总结
pi-dynamic-workflows 的价值不在于替代 Airflow、Temporal 等重型编排系统，而是填补了一个长期被忽视的细分市场：**需要高度动态性、又不愿牺牲类型安全与工程规范的中小型工作流场景**。它特别适合以下人群：正在构建多租户 SaaS 需要租户级流程定制的全栈开发者、探索 LLM Agent 编排模式的 AI 工程师、以及希望将业务规则配置化以减少发布频率的技术团队。573 Stars 的成绩说明其定位已获社区认可，若能在可视化设计器、执行历史追踪等方向持续迭代，有望成为 TypeScript 生态中动态工作流的首选方案。