@db-lyon/flowkit (英语)
YAML 为Node.js配置了任务和流管乐引擎.
定义可重复使用 任务 然后把它们编成 流动 使用声明式YAML. Flowkit)处理配置分层,任务解析,相继执行,被巢流,生命周期钩等.
安装
npm install @db-lyon/flowkit
需要节点.js >= 20.
快开始
1. 定义您的配置 (单位:千美元)pipeline.yml:
tasks:
build:
class_path: tasks.Build
description: Compile the project
options:
target: production
test:
class_path: tasks.Test
description: Run the test suite
deploy:
class_path: tasks.Deploy
description: Deploy artifacts
flows:
ci:
description: Build, test, deploy
steps:
1:
task: build
2:
task: test
3:
task: deploy
options:
environment: staging
2. 创建任务 (单位:千美元)tasks/Build.ts:
import { BaseTask, type TaskResult } from '@db-lyon/flowkit';
interface BuildOptions {
target: string;
}
export default class Build extends BaseTask<BuildOptions> {
get taskName() { return 'build'; }
protected validate() {
if (!this.options.target) throw new Error('target is required');
}
async execute(): Promise<TaskResult> {
this.logger.info(`Building for ${this.options.target}`);
// ... do work ...
return { success: true, data: { target: this.options.target } };
}
}
3. 运行它数字 :
import {
loadConfig,
EngineConfigSchema,
TaskRegistry,
FlowRunner,
} from '@db-lyon/flowkit';
const { config } = loadConfig({
filename: 'pipeline.yml',
schema: EngineConfigSchema,
configDir: './config',
});
const registry = new TaskRegistry();
// Tasks with class_path like "tasks.Build" are resolved dynamically
// from the filesystem (tasks/Build.ts), or register them explicitly:
// registry.register('build', Build);
const runner = new FlowRunner({
tasks: config.tasks,
flows: config.flows,
registry,
context: { logger: console },
});
const result = await runner.run({ flowName: 'ci' });
console.log(result.success); // true
特征
YAML 驱动配置
在 YAML 中定义任务和流量 。 每项任务参考a class_path (已解入磁盘上的文件或已注册的构造器),并可以携带默认 options。流程是参考任务或其他流程的顺序步骤.
配置层
配置加载器按顺序合并多个 YAML 文件 :
defaults (code) → pipeline.yml → pipeline.staging.yml → pipeline.local.yml
const { config } = loadConfig({
filename: 'pipeline.yml',
schema: EngineConfigSchema,
env: 'staging', // loads pipeline.staging.yml overlay
configDir: './config',
});
环境重叠和 .local.yml 文件允许您自定义每个环境或每个开发者而不触及基础配置。 见 文件/配置。 。 。 .
自定义任务
扩展 BaseTask 创建自己的任务。 生命周期为: validate() → execute() ——结果有时间. 发现例外并返回 { success: false } 自动.
class MyTask extends BaseTask<MyOptions> {
get taskName() { return 'my_task'; }
async execute(): Promise<TaskResult> {
return { success: true };
}
}
见 文件/海关任务。 。 。 .
内建的贝壳任务
不写入自定义任务类而运行 shell 命令 :
tasks:
lint:
class_path: shell
options:
command: npm run lint
cwd: /path/to/project
timeout: 60000
在您的登记簿上登记:
import { ShellTask } from '@db-lyon/flowkit';
registry.register('shell', ShellTask as any);
现有的YAML消费者不需要改变. 取消一项方案引用,
通过一个具体援引 AbortSignal数字 :
const controller = new AbortController();
const result = await new ShellTask({}, {
command: 'npm run build',
signal: controller.signal,
}).run();
AI探员
运行 LLM 调用为步骤 。 福克基特已经 无 SDK 依赖性 你电线
模型不可知论 LLMProvider 上下文为 ctx.llm,还有两个任务
消费:
agent_prompt(单位:千美元)AgentPromptTask)——一发即发,可选 JSON-Schema结构化输出(已验证,并有修复再即期).agent(单位:千美元)AgentTask)——一种工具调用回路. 工具参考流程 kit 任务, 或由许可列表加盖的门牌 每工具论证. 多个工具依次调用(包括 平行子代理)在封顶下同时运行.
可重复使用的剂生活在 agents: 根键并运行为流程步骤或
具有强制性预算的其他代理工具maxIterations, (中文(简体) ), tokenBudget, (中文(简体) ),
maxAgentDepth) (中文(简体) ). 迭代和相通都住在代理运行时间,所以流动
保持相继的脊椎没有 loop: 或平行步骤原始.
主机可以选择通过 nestedAgentTaskFactory 改为 FlowRunner 用于控制
任务对象仅用于通过引用的已配置子代理 agent: 工具.
省略时, Flowkit 表现为构造 new AgentTask(ctx, options)。 。 。 .
代理直接运行,因为流程步骤仍然使用正常的注册路径.
tasks:
extract:
class_path: agent_prompt
options:
prompt: "Pull the ticket fields from:\n${steps.1.text}"
schema:
type: object
required: [title, priority]
properties:
title: { type: string }
priority: { type: string, enum: [low, medium, high] }
每一次呼叫都因共享核心而更坚固: timeout (与供应商)
中止, retries 带有指数后退、结构化输出验证+
修理,和输出大小的盖。 见 文件/助理人员。 。 。 .
网状流动
一个步骤可以参考另一种流程而不是任务:
flows:
ci:
description: CI pipeline
steps:
1: { task: build }
2: { task: test }
release:
description: Full release
steps:
1: { flow: ci }
2: { task: deploy }
一个流步 options 覆盖任务选项 内部 筑巢者
,按任务名键。 优先度, 低 + 高: 任务默认 + 附加流量
覆盖 + 步骤本身的内含选项 + 运行时间 params。 。 。 .
release:
steps:
1:
flow: ci
options:
test: { coverage: 90 } # overrides the `test` task's options inside `ci`
2: { task: deploy }
分步扫描运行时间选项
运行时间 params 默认到达每个步骤。 打开 options_scope: step
(在流量,跑者,或跑者)按任务名称或
步骤路径 :
await runner.run({
flowName: 'release',
optionsScope: 'step',
params: { deploy: { environment: 'prod' }, '2/1': { coverage: 90 } },
});
匹配任何步骤的选取器在开始前失败运行 。 见 文件/配置。 。 。 .
有条件的步骤when(中文(简体) )
一个步骤只有在 when 真实的。 它接受布尔或弦
在运行时评价表达式。 没有 conditionEvaluator 配置, a
字符串解析为 ${...} 参考和真实性测试;提供a
conditionEvaluator 插入真实表达式语言。 病态结果跳过
步骤skipReason: 'when')不失入流.
steps:
1: { task: build }
2: { task: deploy, when: '${steps.1.shouldDeploy}' }
3: { task: notify, when: false }
new FlowRunner({
// ...
conditionEvaluator: (expr, ctx) => evalMyDsl(expr, ctx), // ctx: { steps, params, context, error }
});
飞行前检查
流动和步骤可以宣布 checks 由同一条件评价员评估.
发火检查错误(在步骤开始前)、跳过或警告:
steps:
1:
task: import
checks:
- { when: "editor.has_modal_dialog", action: error, message: A modal dialog is open. }
runner.preflight(flowName, params) 不运行评估每张检查
将每个步骤报告为 run, (中文(简体) ), skip, (中文(简体) ), error 或者说 unknown。见
文件/配置。 。 。 .
失败时继续( N)ignore_failure(中文(简体) )
默认情况下, 任何失败的步骤都会中止流量 。 标出一步 ignore_failure: true
记录失败,但继续前进(步骤结果有: ignoredFailure: true) (中文(简体) ).
steps:
1: { task: deploy }
2: { task: publish_release_notes, ignore_failure: true }
3: { task: announce }
跳过步骤
按任务名称或步骤编号跳过 :
await runner.run({ flowName: 'release', skip: ['deploy'] });
await runner.run({ flowName: 'release', skip: ['2'] });
或标记为 YAML 永久跳过的步骤 :
steps:
3:
task: None # or `flow: None` for a step that ran a flow
计划模式
在不运行任何操作的情况下预览实施计划 :
const result = await runner.run({ flowName: 'ci', plan: true });
result.steps.forEach(s =>
console.log(`${s.stepNumber}: [${s.type}] ${s.name}${s.skipped ? ' (skip)' : ''}`)
);
过 expandNestedFlows: true 以递回地将嵌入式流入
他们的孩子的阶梯,每道标有等级 path (例如:. 2/1:
await runner.run({ flowName: 'release', plan: true, expandNestedFlows: true });
综合任务
儿童依赖其投入的任务贯穿每个儿童 this.step()。 。 。 .
孩子们通过跑道 就像流道(记录在父母的下面),
重新测试、验证、回滚),在流下或通过 runTask。 。 。 .
可选静态 expand() 列出不运行的计划中的儿童
他们。 见 文件/海关任务。 。 。 .
async execute() {
for (const file of this.options.files) {
const r = await this.step('asset.import', { file });
if (!r.success) return r;
}
return { success: true };
}
生命周期钩
附加用于观察或反应流执行的钩子 :
const runner = new FlowRunner({
// ...
hooks: {
beforeRun: async (flowName, plan) => { /* ... */ },
beforeStep: async (step) => { /* ... */ },
afterStep: async (step, result) => { /* ... */ },
onStepError: async (step, error, completed) => { /* ... */ },
afterRun: async (result) => { /* ... */ },
},
});
beforeRun编号 :afterRun 一次火 为顶级流. beforeStep编号 :afterStep 每一步都放出火来 包括那些在巢中流出的人.
DAG 公用事业
具有周期和缺失依赖性检测的地形分类:
import { topologicalSort } from '@db-lyon/flowkit';
const sorted = topologicalSort([
{ id: 'a', dependencies: [], data: null },
{ id: 'b', dependencies: ['a'], data: null },
{ id: 'c', dependencies: ['a', 'b'], data: null },
]);
// sorted: [a, b, c]
掷出 CircularDependencyError 或者说 MissingDependencyError 在无效的图表上.
锁定接口
Flowkit 接受执行该程序的任何日志 Logger 界面( 与皮诺、 winton等兼容):
interface Logger {
debug(...args: unknown[]): void;
info(...args: unknown[]): void;
warn(...args: unknown[]): void;
error(...args: unknown[]): void;
child(bindings: Record<string, unknown>): Logger;
}
通过任务上下文或流跑者配置传递 。 A 类 noopLogger 默认使用.
子路径导出
import { loadConfig } from '@db-lyon/flowkit/config';
import { BaseTask, TaskRegistry } from '@db-lyon/flowkit/task';
import { FlowRunner } from '@db-lyon/flowkit/flow';
import { topologicalSort } from '@db-lyon/flowkit/dag';
文档
- 开始 ——分步设置指导
- 自定义任务 ——写作和登记任务
- AI探员 — LLM 提示,结构化输出,工具调用代理
- 配置 ——YAML schema,分层,深层合并
- 警卫 围绕东道方业务的管道之前/之后
- API 参考 ——完整类型和功能参考
许可证
麻省理工学院——见 自由主义。 。 。 .