@ cascade- flow/ workflow
为 CascadeFlow 编写 SDK 的工作流程 - 为构建工作流程提供类型安全步骤定义帮助和公用设施.
安装
bun add @cascade-flow/workflow
API 参考
步骤定义
defineStep(spec)
不输入输入的工作流程默认步骤定义帮助器 .
import { defineStep } from "@cascade-flow/workflow";
export const step = defineStep({
name: "my-step",
dependencies: {},
exportOutput: true,
fn: async ({ ctx }) => {
return { result: "value" };
}
});
baseDefineStep<TInput>()
创建已键入的工厂函数 defineStep 函数用于输入的工作流程.
用法模式 :
- 创建特定的工作流程
defineStep.ts数字 :
// workflows/my-workflow/define-step.ts
import { baseDefineStep } from "@cascade-flow/workflow";
import type { WorkflowInput } from "./input-schema";
export const defineStep = baseDefineStep<WorkflowInput>();
- 用在你的台阶上:
// workflows/my-workflow/steps/my-step/step.ts
import { defineStep } from "../../define-step";
export const step = defineStep({
fn: async ({ ctx }) => {
// ctx.input is now typed as WorkflowInput!
const message = ctx.input.message; // Fully typed
return { processed: message };
}
});
跳过和可选依赖
Skip
故意跳过步骤错误分类 。 跳过步数 :
- 标记为“ 滑动 ” (没有失败)
- 连为依附步骤( 它们也会跳过)
- 不触发重试
import { Skip } from "@cascade-flow/workflow";
export const step = defineStep({
fn: async ({ ctx }) => {
if (!ctx.input.shouldProcess) {
throw new Skip("Processing not needed");
}
return { result: "processed" };
}
});
有元数据 :
throw new Skip("Feature flag disabled", {
flag: "new-processor",
environment: process.env.NODE_ENV
});
optional(step)
将依赖标记为可选。 可选依赖性 :
- 跳过时不会引起级联跳过
- 将会是
undefined在依赖对象中,如果跳过 - 如果成功完成, 将具有正常输出值
import { defineStep, optional } from "@cascade-flow/workflow";
import { step as validation } from "../validation/step.ts";
import { step as enrichment } from "../enrichment/step.ts";
export const step = defineStep({
dependencies: {
validation, // Required - will cascade skip
enrichment: optional(enrichment) // Optional - won't cascade skip
},
fn: async ({ dependencies }) => {
// dependencies.validation is guaranteed to exist
// dependencies.enrichment is StepOutput | undefined
if (dependencies.enrichment) {
// Use enrichment data
} else {
// Fallback when enrichment was skipped
}
return { result: "processed" };
}
});
isOptional(dep)
检查是否将依赖标记为可选 。 (内部工具)
类型
RunnerContext
每个步骤可用的运行时间背景 :
type RunnerContext = {
runId: string;
workflow: {
slug: string; // Workflow directory name
name: string; // Friendly name from workflow.json
};
input?: unknown; // Validated workflow input
log: (...args: any[]) => void;
};
InferStepOutput<T>
用于从函数签名中推断步骤输出类型的工具类型 .
JSONValue, (中文(简体) ), JSONArray, (中文(简体) ), JSONObject, (中文(简体) ), JSONPrimitive
代表 JSON 可序列值的类型 。 所有步骤的产出必须是可按顺序排列的.
StepOutput
别名 JSONValue - 步骤返回值的类型限制.
OptionalDependency<T>
代表可选依赖性包装的类型.
步骤配置
这个 defineStep 光谱接受 :
{
name?: string; // Display name (defaults to directory name)
dependencies?: Record<...>; // Other steps this step depends on
exportOutput?: boolean; // Include in final workflow output
maxRetries?: number; // Retry count on failure (default: 0)
retryDelayMs?: number; // Delay between retries (default: 0)
timeoutMs?: number; // Max execution time (default: 300000ms / 5 min)
fn: (args) => Promise<Output> | Output; // Step implementation
}
组内安排步骤
可通过嵌入式目录结构组织步骤,以更好地组织:
workflows/my-workflow/steps/
├── step-1/step.ts # Flat step
└── data-processing/ # Group directory
├── extract/
│ └── fetch-data/step.ts # Nested step
└── transform/
└── normalize/step.ts # Nested step
关键点 :
- 团体纯粹是组织性的(没有特殊行为)
- 平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平起平
- 步骤ID使用路径 :
"step-1"(平面)或"data-processing/extract/fetch-data"(已注销) - 依赖使用相对进口:
// In steps/notifications/send/step.ts
import { step as fetchData } from "../../data-processing/extract/fetch-data/step.ts";
export const step = defineStep({
dependencies: { fetchData },
fn: async ({ dependencies }) => {
// Use data from nested group
return { sent: true, data: dependencies.fetchData };
}
});
实例
简单步骤
import { defineStep } from "@cascade-flow/workflow";
export const step = defineStep({
exportOutput: true,
fn: async () => {
return { timestamp: Date.now() };
}
});
与依附关系同步
import { defineStep } from "@cascade-flow/workflow";
import { step as stepA } from "../step-a/step.ts";
import { step as stepB } from "../step-b/step.ts";
export const step = defineStep({
dependencies: { stepA, stepB },
fn: async ({ dependencies }) => {
// Both dependencies are available and fully typed
const combined = {
a: dependencies.stepA.value,
b: dependencies.stepB.result
};
return combined;
}
});
步入重试逻辑
import { defineStep } from "@cascade-flow/workflow";
export const step = defineStep({
maxRetries: 3,
retryDelayMs: 1000,
fn: async () => {
const response = await fetch("https://api.example.com/data");
if (!response.ok) throw new Error("API request failed");
return await response.json();
}
});
有条件跳过
import { defineStep, Skip } from "@cascade-flow/workflow";
export const step = defineStep({
fn: async ({ ctx }) => {
if (!ctx.input.featureEnabled) {
throw new Skip("Feature disabled");
}
// Process when enabled
return { status: "processed" };
}
});
检查站
检查点在一步内节省了中间进展,在重复中持续取得成果:
import { defineStep } from "@cascade-flow/workflow";
export const step = defineStep({
fn: async ({ ctx }) => {
for (let i = 0; i < items.length; i++) {
const result = await ctx.checkpoint?.("process-item", async () => {
return processItem(items[i]);
});
}
return { processed: items.length };
}
});
用于平行检查站 awaitCheckpoints() 在错误传播前确保全部完整:
import { defineStep, awaitCheckpoints } from "@cascade-flow/workflow";
export const step = defineStep({
fn: async ({ ctx }) => {
const [a, b] = await awaitCheckpoints([
ctx.checkpoint?.("task-a", async () => fetchData()),
ctx.checkpoint?.("task-b", async () => validate()),
]);
return { a, b };
}
});