# @cogitator-ai/workflows 中文文档

规范地址：https://npmcn.com/package/%40cogitator-ai/workflows/

版本：0\.15\.6
npm 元数据更新时间：2026-10-29T00:51:08.297Z
中文译文更新时间：2026-11-10T09:54:56.413Z
软件包许可：MIT

本文由上游 README 翻译，可能存在偏差，请结合原文核对 API、命令与配置。版权与许可归原作者所有。

原始 npm 页面：https://www.npmjs.com/package/%40cogitator-ai/workflows
原文对照：https://npmcn.com/package/%40cogitator-ai/workflows/?lang=original
源码仓库：https://github.com/cogitator-ai/Cogitator-AI

## 关键信息（来自 npm 元数据）

- 包体积（解压后）：1.2 MB，223 个文件
- 模块格式：ES Module
- TypeScript 类型：包内自带
- Node.js 要求：\>=22\.12\.0
- 运行依赖：4 个
- 首次发布：2025-12-31T09:04:48.215Z
- 最新版本发布：2026-10-29T00:51:08.055Z
- 近 30 天下载量：12014（2026-10-11 至 2026-11-09，npm 官方下载统计）
- 已发布版本数：109（近 12 个月 109 个）

## 中文简介

基于 Cogitator 代理的 DAG 工作流程引擎

## 安装

```sh
npm install @cogitator-ai/workflows
```

也可以使用：`pnpm add @cogitator-ai/workflows`、`yarn add @cogitator-ai/workflows`、`bun add @cogitator-ai/workflows`

## README

# @cogitator-ai/ 工作流程

[![npm version](https://img.shields.io/npm/v/@cogitator-ai/workflows.svg)](https://www.npmjs.com/package/@cogitator-ai/workflows) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

基于DAG的工作流程引擎用于Cogitator代理. 构建多步工作流程,以分行为主,循环为主,联通为主,检查站为主,人行为主,计时为主,分行为主,分行为主,分行为主,分行为主,分行为主,分行为主,分行为主,分行为主,分行为主,分行为主,分行为主.

## 安装

```bash
pnpm add @cogitator-ai/workflows @cogitator-ai/core

# Optional, for the durable stores
pnpm add pg                                 # PostgreSQL stores
pnpm add @cogitator-ai/redis ioredis        # Redis stores
```

网站文件 : [工作流量](https://cogitator.app/docs/workflows), (中文(简体) ), [构建器](https://cogitator.app/docs/workflows/builder), (中文(简体) ), [节点](https://cogitator.app/docs/workflows/nodes), (中文(简体) ), [执行](https://cogitator.app/docs/workflows/execution), (中文(简体) ), [模式](https://cogitator.app/docs/workflows/patterns), (中文(简体) ), [萨加斯](https://cogitator.app/docs/workflows/sagas), (中文(简体) ), [时间安排](https://cogitator.app/docs/workflows/scheduling)。 。 。 .

## 特征

-   **DAG 构建器** ——节点,条件,回路和平行扇出并有验证和入点检测
-   **正确的连接** ——有多个上游分行的节点运行一次,经过全部分行完成之后
-   **节点政策** 每个节点 `timeout`, (中文(简体) ), `retries` 财务报告和审定财务报表 `retryDelay`
-   **实时流线** - 执行事件的同步生成器
-   **检查站** ——保存并恢复工作流程状态
-   **持久存储** - Redis或PostgreSQL的检查站、运行、核准和定时器,由若干程序共享
-   **预建的节点** ——代理,工具,功能和自定义节点,外加用于计时器的适配器,人类批准,地图减量和子工作流程
-   **萨加图案** ——复线,断路器,补偿,死字排队,一意孤行
-   **触发器** ——克龙,网钩和事件触发器
-   **观察性** 追踪(主机、OTLP、Zipkin)和度量衡(Prometheus格式)
-   **工作流程管理** ——运行存储,调度,取消,重试,重放

## 快速启动

```typescript
import { Cogitator, Agent } from '@cogitator-ai/core';
import { WorkflowBuilder, WorkflowExecutor, agentNode } from '@cogitator-ai/workflows';

type ReportState = {
  topic: string;
  analysis?: string;
};

const cogitator = new Cogitator({ llm: { defaultModel: 'ollama/llama3.2' } });
const analyst = new Agent({
  name: 'analyst',
  model: 'ollama/llama3.2',
  instructions: 'Analyze topics concisely.',
});

const workflow = new WorkflowBuilder<ReportState>('report')
  .initialState({ topic: '' })
  .addNode(
    'analyze',
    agentNode<ReportState>(analyst, {
      inputMapper: (state) => `Analyze: ${state.topic}`,
      stateMapper: (result) => ({ analysis: result.output }),
    })
  )
  .addNode('report', async (ctx) => ({ output: `Report: ${ctx.state.analysis}` }), {
    after: ['analyze'],
  })
  .build();

const result = await new WorkflowExecutor(cogitator).execute(workflow, { topic: 'Edge AI' });
console.log(result.state.analysis, result.error);
```

* * *

## 核心概念

### 工作流量

```typescript
const workflow = new WorkflowBuilder<{ count: number }>('counter')
  .initialState({ count: 0 })
  .addNode('increment', async (ctx) => ({ state: { count: ctx.state.count + 1 } }))
  .addNode('print', async (ctx) => ({ output: `Count: ${ctx.state.count}` }), {
    after: ['increment'],
    config: { timeout: 5_000, retries: 2, retryDelay: 500 },
  })
  .build();
```

-   国家类型必须满足 `WorkflowState` (单位:千美元)`Record<string, unknown>`: 以 `type`,或作为 `interface MyState extends WorkflowState { ... }` (单位:千美元)`import type { WorkflowState } from '@cogitator-ai/types'`) (中文(简体) ). 一个平地 `interface` 不符合限制.
-   `addNode(name, fnOrNode, { after, config })` 接受由工厂创建的节点功能或节点.
-   切入点为单个节点(或有条件/平行/落地构造),无 `after`. 拒绝独立根-增加一个平行的扇子出击或呼叫 `.entryPoint(name)`。 。 。 .
-   接收节点函数 `ctx.state` (副本), `ctx.input` (上游节点的产出), `ctx.nodeId`, (中文(简体) ), `ctx.workflowId`, (中文(简体) ), `ctx.step`, (中文(简体) ), `ctx.idempotencyKey`, (中文(简体) ), `ctx.reportProgress()` 和内置节点, `cogitator`, (中文(简体) ), `signal` (运行取消)和 `depth` (子工作流入巢).
-   `ctx.idempotencyKey` (单位:千美元)`workflow:<workflowId>:node:<name>:visit:<n>`) 命名节点的这次访问:当节点被重试或恢复运行时相同,当工作流程回到节点时是新的. 传递给外部的API,该API会取出"一能"密钥,因此再次运行的节点不收卡或两次发送消息.
-   返回时 `{ state }` 合并到工作流程状态, `{ output }` 用于下游节点和 `{ next }` 以动态方式运行.

### 工作流程执行器

```typescript
const executor = new WorkflowExecutor(cogitator);

const result = await executor.execute(
  workflow,
  { count: 5 },
  {
    maxConcurrency: 4, // parallel nodes per step
    maxIterations: 100, // guards loops
    signal: abortController.signal,
    onNodeStart: (node) => console.log('start', node),
    onNodeComplete: (node, output, duration) => console.log('done', node, duration),
    onNodeError: (node, error) => console.error(node, error.message),
  }
);

console.log(result.state, result.nodeResults, result.duration, result.error);
```

`execute()` 永不丢弃节点失败: 失败返回到 `result.error` (单位:千美元)`NodeTimeoutError` 节点超时.

运行级政策适用于每个节点:

```typescript
await executor.execute(workflow, input, {
  defaultRetry: { maxRetries: 3, backoff: 'exponential', initialDelay: 500 }, // nodes without config.retries
  defaultCircuitBreaker: breakerConfig, // per-node breaker, shared across runs of this executor
  deadLetterQueue: createInMemoryDLQ(), // entry for every node that finally failed
  idempotencyStore, // reuse results of nodes that already completed for this workflowId and visit, across resumes
  approvalStore, // default store for humanWorkflowNode
  timerStore, // default store for persisted timers
  tracer,
  metricsCollector,
});
```

* * *

## 实时流线

```typescript
for await (const event of executor.stream(workflow, { count: 0 })) {
  switch (event.type) {
    case 'workflow_started':
    case 'node_started':
    case 'node_progress':
    case 'node_completed':
    case 'node_error':
      console.log(event.type);
      break;
    case 'workflow_completed':
      console.log('final state', event.result.state);
      break;
  }
}
```

节点报告进度 `ctx.reportProgress(0..100)`· ; `execute()` 暴露出来 `onNodeProgress`。 。 。 .

* * *

## 预建的节点

```typescript
import { agentNode, toolNode, functionNode, customNode } from '@cogitator-ai/workflows';

builder
  .addNode(
    'research',
    agentNode<MyState>(researcher, {
      inputMapper: (state) => state.question,
      stateMapper: (result) => ({ findings: result.output }),
      runOptions: { timeout: 60_000 },
    })
  )
  .addNode(
    'calculate',
    toolNode<MyState, { expression: string }>(calculator, {
      argsMapper: (state) => ({ expression: state.formula }),
      stateMapper: (result) => ({ value: result }),
    }),
    { after: ['research'] }
  )
  .addNode(
    'normalize',
    functionNode<MyState, string>('normalize', async (state) => state.findings!.trim(), {
      stateMapper: (output) => ({ findings: output as string }),
    }),
    { after: ['calculate'] }
  );
```

当 `agentNode` 代理调用一个需要批准的工具, 它的暂停运行从来没有通过一个答案。 与一个 `approvalStore` (执行选项,或 `approvals.approvalStore`)每一次等待电话成为了人行商店的核准/拒绝请求,一旦得到答复,运行恢复, `approvals: { assignee, timeout, timeoutAction, ... }` 形状请求。 没有商店,或与 `approvals: false`时,节点以 `AgentRunPausedError` 携带检查站,不经过复试。 对一个问题暂停运行的运行 `context.interrupt()`) 节点失败的方式相同,批准商店与否,问题在 `error.interrupts`:只有一个人回答,所以节点从不单独恢复. 回答他们 `cogitator.resume(agent, error.checkpoint, { responses })`。 。 。 .

A 类 `toolNode` 运行其工具通过 `cogitator.invokeTool`,一个代理运行的方式称它: 验证的论点反对工具的策略, 护栏,沙盒和 `tool.timeout` 应用程序,工具为 `requiresApproval` 在经营前询问同一家批准店 一个被拒绝的电话,或者一个没有人要问,不运行工具就使节点失效.

代理和工具节点使用运行的中止信号,因此取消工作流程取消飞行中LLM和工具呼叫.

* * *

## 分行、循环和加入

```typescript
const workflow = new WorkflowBuilder<{ approved: boolean }>('review-flow')
  .initialState({ approved: false })
  .addNode('review', reviewFn)
  .addConditional('check', (state) => (state.approved ? 'publish' : 'revise'), {
    after: ['review'],
  })
  .addNode('publish', publishFn, { after: ['check'] })
  .addNode('revise', reviseFn, { after: ['check'] })
  .addNode('notify', notifyFn, { after: ['publish', 'revise'] })
  .build();
```

有条件返回要取的分支的名称; 每个节点( 或构造) 与 `after: ['check']` 是一个候选分支.

```typescript
const workflow = new WorkflowBuilder<{ attempts: number; done: boolean }>('retry-flow')
  .initialState({ attempts: 0, done: false })
  .addNode('attempt', async (ctx) => ({
    state: { attempts: ctx.state.attempts + 1, done: await tryIt() },
  }))
  .addLoop('again', {
    condition: (state) => !state.done && state.attempts < 3, // typed with the builder's state
    back: 'attempt',
    exit: 'finish',
    after: ['attempt'],
  })
  .addNode('finish', finishFn)
  .build();
```

只有循环 `back` 财务报告和审定财务报表 `exit` 节点可跟随它; 任何其它有循环的节点 `after` 制造 `build()` 掷出. 一个根 `addParallel` 是工作流程的切入点 :

```typescript
const workflow = new WorkflowBuilder('fan-out')
  .addParallel('fan', ['fetch-a', 'fetch-b'])
  .addNode('fetch-a', fetchA)
  .addNode('fetch-b', fetchB)
  .addNode('normalize-b', normalizeB, { after: ['fetch-b'] })
  .addNode('merge', async (ctx) => ({ output: ctx.input }), { after: ['fetch-a', 'normalize-b'] })
  .build();
```

`merge` 运行一次,在两个分支完成之后,并收到两个产出作为 `ctx.input`。 。 。 .

* * *

## 检查站

```typescript
import { FileCheckpointStore } from '@cogitator-ai/workflows';

const executor = new WorkflowExecutor(cogitator, new FileCheckpointStore('./checkpoints'));

const first = await executor.execute(workflow, input, {
  checkpoint: true,
  checkpointStrategy: 'per-node', // or 'per-iteration' (default)
});

if (first.error && first.checkpointId) {
  const resumed = await executor.resume(workflow, first.checkpointId);
}
```

进程之间共享的检查点位于 Redis 或 Postgres; 商店占用您现有的客户端( a) `@cogitator-ai/redis` 客户端或iuredis, a `pg` 池边:

```typescript
import { RedisCheckpointStore, PostgresCheckpointStore } from '@cogitator-ai/workflows';

new RedisCheckpointStore({ client: redis, keyPrefix: 'myapp:checkpoints' });
new PostgresCheckpointStore({ client: pool, table: 'workflow_checkpoints' }); // created on first use
```

`ttl` (seconds)使 Redis 放弃了一个检查点, 及其工作流程的索引, 在上次保存到它之后很久, 因为没有人恢复或删除运行 。 这需要 `expire` 财务报告和审定财务报表 `zrangebyscore` 在客户端上,两个客户都有,并且每个保存都会降低比 ttl 年长的检查站的索引条目:

```typescript
new RedisCheckpointStore({ client: redis, keyPrefix: 'myapp:checkpoints', ttl: 7 * 24 * 3600 });
```

恢复运行的只是没有完成的节点,之后的节点会得到完成的节点输出作为输入.

运行、批准和定时器有同样的耐用用品:

```typescript
import {
  PostgresRunStore,
  RedisApprovalStore,
  PostgresTimerStore,
  createWorkflowManager,
} from '@cogitator-ai/workflows';

createWorkflowManager({ cogitator, runStore: new PostgresRunStore({ client: pool }) });
new RedisApprovalStore({ client: redis, pollInterval: 1000 }); // answers from other processes arrive by polling
new PostgresTimerStore({ client: pool, claimTtl: 60_000 }); // each overdue timer is claimed by one TimerManager
```

Postgres商店在第一使用时创建他们的表格(当两个进程同时启动时安全); Redis将索引存储在排序的集中运行和请求,而不是扫描密钥. 在另一个过程中得到的核准到达了内部等待的人类节点 `pollInterval` (违约1 s); 对逾期的计时器提出索赔 `claimTtl` (默认60 s)如此多 `TimerManager`s可以共享一个商店而无需发射两次计时器,一个坠毁的工人的计时器一旦索赔期满就会回来.

A 类 `TimerManager` 处理者开始前重新提出权利要求 `claimTtl / 3` 当它运行时, 如此慢的处理者 保持他们的定时器。 如果索赔被损失(另一工人拿走),处理人被跳过或 `onClaimLost` 报告; 没有处理器的计时器, 或超过民意测验的 `batchSize`,然后释放,以便另一名经理立即带走他们。 一个失败的处理者将索赔保留到租赁结束为止,从而排除了重复:

```typescript
import { createTimerManager, RedisTimerStore } from '@cogitator-ai/workflows';

const timers = createTimerManager(new RedisTimerStore({ client: redis, claimTtl: 30_000 }), {
  pollInterval: 1_000,
  onClaimLost: (entry) => console.warn('another worker took timer', entry.id),
});
timers.setDefaultHandler(async (entry) => console.log('fired', entry.id));
await timers.start();
```

自定义存储已执行 `TimerStore`· ; `claimTtl`, (中文(简体) ), `renew(id)` 财务报告和审定财务报表 `release(id)` 并允许处理上述索赔.

* * *

## 计时器

```typescript
import {
  delayNode,
  dynamicDelayNode,
  cronWaitNode,
  untilNode,
  timerWorkflowNode,
  parseDuration,
  formatDuration,
} from '@cogitator-ai/workflows';

builder
  .addNode('cool-down', timerWorkflowNode(delayNode('cool-down', parseDuration('5m'))))
  .addNode(
    'backoff',
    timerWorkflowNode(dynamicDelayNode<MyState>('backoff', (state) => state.retries * 1000)),
    { after: ['cool-down'] }
  )
  .addNode('business-hours', timerWorkflowNode(cronWaitNode('business-hours', '0 9 * * 1-5')), {
    after: ['backoff'],
  })
  .addNode('deadline', timerWorkflowNode(untilNode<MyState>('deadline', (state) => state.dueAt)), {
    after: ['business-hours'],
  });

formatDuration(90_000); // '1.5m'
```

中止运行( N)`signal`经理的 `pause()` 或者说 `cancel()`)停止等待,以一个 `AbortError`,因此节点没有检查完成后再次恢复运行。 过 `{ timerStore }` 改为 `timerWorkflowNode` 合并 `persist: true` 配置(或使用) `createTimerNodeHelpers(store)`持续计时器:中断的恒定计时器在其储存处被取消,在恢复节点时只等待到原始 `firesAt`因此,暂停永远不会缩短或延长等待。 持续计时器在节点等待时通过它的商店被取消(`TimerManager.cancel`未标记已发射:节点报告 `cancelled: true` 财务报告和审定财务报表 `onCancelled` 当等待结束. `createTimerManager(store)` 财务报告和审定财务报表 `createRecurringScheduler(manager)` 程序一直存在,而且经常出现定时器。 Cron 帮助 : `validateCronExpression`, (中文(简体) ), `getNextCronOccurrence(s)`, (中文(简体) ), `describeCronExpression`, (中文(简体) ), `CRON_PRESETS`。 。 。 .

工作流程的计时器和克龙运行在核心的调度器上:克龙帮助器通过核心的克龙引擎来计算 `CRON_PRESETS` 顶上( e)`workflowsCron`, (中文(简体) ), `TimerManager` 将计时器通过核心的计票人和索赔 `InMemoryTimerStore` 财务报告和审定财务报表 `FileTimerStore` 核心类(a) `FileTimerStore` (参见其他进程所写的). `JobScheduler` 拒绝无效的 cron 表达式或时区 `VALIDATION_ERROR` 并报告无法安排的运行 `onError` 而不是放弃它.

* * *

## 萨加图案

### 重试

```typescript
import { executeWithRetry, withRetry } from '@cogitator-ai/workflows';

const outcome = await executeWithRetry((attempt) => callService(attempt), {
  maxRetries: 4,
  backoff: 'exponential', // 'constant' | 'linear' | 'exponential'
  initialDelay: 500,
  maxDelay: 10_000,
  jitter: 0.1,
  isRetryable: (error) => !error.message.includes('invalid'),
});
if (!outcome.success) console.error(outcome.error);

const fetchWithRetry = withRetry((url: string) => fetch(url), { maxRetries: 3 });
```

### 断路器

```typescript
import { createCircuitBreaker, CircuitBreakerOpenError } from '@cogitator-ai/workflows';

const breaker = createCircuitBreaker({ threshold: 5, resetTimeout: 30_000, successThreshold: 2 });

try {
  await breaker.execute('payments', () => payments.charge(order));
} catch (error) {
  if (error instanceof CircuitBreakerOpenError) {
    // use a fallback
  }
}

console.log(breaker.getStats('payments'));
```

### 补偿

给节点 `config.compensation` 当运行的后期节点失败时,执行器会将其卷回:

```typescript
builder
  .addNode('reserve', reserveFn, {
    config: { compensation: { compensate: async (state) => inventory.release(state.orderId) } },
  })
  .addNode('charge', chargeFn, {
    after: ['reserve'],
    config: {
      compensation: {
        compensate: async (state) => payments.refund(state.paymentId!),
        compensateTimeout: 10_000,
      },
    },
  })
  .addNode('ship', shipFn, { after: ['charge'] });

const result = await executor.execute(workflow, input, {
  onCompensationStart: (node) => console.log(`rolling back ${node}`),
  onCompensationComplete: (node) => console.log(`rolled back ${node}`),
});
```

已完成的节点为 a `compensate` 按倒数完成顺序向后滚动(`compensateOrder: 'parallel' | 'forward'` 改变, `compensateCondition` 跳过一步; 中止、 暂停和取消的运行不会得到补偿 。 对于执行器外的回滚,请使用 `CompensationManager`数字 :

```typescript
import { compensationBuilder } from '@cogitator-ai/workflows';

const compensation = compensationBuilder<OrderState>()
  .addStep('reserve', async (state) => inventory.release(state.orderId))
  .addStep('charge', async (state) => payments.refund(state.paymentId!))
  .build();

compensation.markCompleted('reserve', reservation);
compensation.markCompleted('charge', payment);

const report = await compensation.compensate(state, 'ship', new Error('carrier down'), {
  onStepStart: (nodeId) => console.log(`compensating ${nodeId}`),
});
```

对已完成步骤的补偿按倒数完成顺序进行;可选 `CompensationHooks` 参数( e)`onStepStart`, (中文(简体) ), `onStepComplete`观察每走一步.

### 死信队列和同上

```typescript
import {
  createFileDLQ,
  createDLQEntry,
  createInMemoryIdempotencyStore,
  idempotent,
} from '@cogitator-ai/workflows';

const dlq = createFileDLQ('./dlq');
await dlq.add(createDLQEntry('charge', workflowId, 'checkout', state, error, { attempts: 3 }));
const failed = await dlq.list({ workflowName: 'checkout' });

const store = createInMemoryIdempotencyStore();
const receipt = await idempotent(store, `charge:${orderId}`, () => payments.charge(orderId));
```

#### 重试已死信件

`manager.retryDeadLetter(dlq, id)` 又运行一个已死字母的节点。 它从失败的节点重放节点运行(在它保留其检查结果而不会运行两次之前的节点),或者在运行在第一个检查点之前失败时从其输入中再次运行工作流程. 尝试先被记录在条目上,在重试成功后该条目被去掉,因此再次失败的条目会再被保留一次尝试. 工作流程必须向管理人员登记,管理人员需要一名 `checkpointStore`。通过运行所需的执行选项,例如 `approvalStore` 或者说 `timerStore`,作为第三个论点:

```typescript
const manager = createWorkflowManager({ cogitator, checkpointStore, runStore });
await manager.execute(checkout, input, { deadLetterQueue: dlq });

for (const entry of await dlq.list({ workflowName: 'checkout-workflow' })) {
  const result = await manager.retryDeadLetter(dlq, entry.id, { approvalStore });
  if (result.error) console.log(`${entry.nodeId} failed again: ${result.error.message}`);
}
```

#### 邮袋 DLQ

`PostgresDLQ` 保留 Postgres 中的队列, 因此失败的节点在重新启动后存活, 每个进程都看到相同的队列 。 它需要一个 `pg` Pool 或客户端, 在首次使用时创建表格, 在 SQL 中过滤并删除过期的条目 `cleanupExpired()`数字 :

```typescript
import pg from 'pg';
import { PostgresDLQ } from '@cogitator-ai/workflows';

const dlq = new PostgresDLQ({
  client: new pg.Pool({ connectionString: process.env.DATABASE_URL }),
  table: 'checkout_dead_letters', // default cogitator_workflow_dead_letters
  defaultTTL: 14 * 24 * 60 * 60 * 1000,
});
```

* * *

## 子工作流程

```typescript
import {
  subworkflowNode,
  subworkflowWorkflowNode,
  fanOutFanIn,
  parallelSubworkflowsNode,
} from '@cogitator-ai/workflows';

builder
  .addNode(
    'enrich',
    subworkflowWorkflowNode(
      subworkflowNode<ParentState, ChildState>('enrich', {
        workflow: enrichmentWorkflow,
        inputMapper: (state) => ({ record: state.record }),
        outputMapper: (result, state) => ({ ...state, enriched: result.state.record }),
        timeout: 60_000,
        onError: 'retry', // 'propagate' | 'ignore' | 'catch' | 'retry'
        maxDepth: 5,
      })
    )
  )
  .addNode(
    'per-region',
    parallelSubworkflowsNode(
      fanOutFanIn<ParentState, RegionState>('per-region', {
        workflow: regionWorkflow,
        getInputs: (state) => state.regions.map((region) => ({ id: region, input: { region } })),
        aggregator: (results, state) => ({ ...state, regionCount: results.size }),
        concurrency: 3,
      })
    ),
    { after: ['enrich'] }
  );
```

向父母宣传儿童失职(或随 `onError`); 超时取消孩子的跑步. `scatterGather`, (中文(简体) ), `raceSubworkflows` 财务报告和审定财务报表 `fallbackSubworkflows` 涵盖其他常见模式.

* * *

## 人类圈

```typescript
import {
  approvalNode,
  humanWorkflowNode,
  InMemoryApprovalStore,
  WebhookNotifier,
} from '@cogitator-ai/workflows';

const approvalStore = new InMemoryApprovalStore();
const approvalNotifier = new WebhookNotifier({ url: 'https://hooks.example.com/approvals' });

builder
  .addNode(
    'approve-expense',
    humanWorkflowNode(
      approvalNode<ExpenseState>('approve-expense', {
        title: 'Approve expense',
        description: (state) => `Amount: $${state.amount}`,
        assignee: 'manager@company.com',
        timeout: 24 * 60 * 60 * 1000,
        timeoutAction: 'reject',
      }),
      { approvalStore, approvalNotifier, stateMapper: (result) => ({ approved: result.approved }) }
    )
  )
  .addConditional('route', (state) => (state.approved ? 'pay' : 'decline'), {
    after: ['approve-expense'],
  });

// Elsewhere (API handler, UI, Slack action):
await approvalStore.submitResponse({
  requestId,
  decision: true,
  respondedBy: 'manager@company.com',
  respondedAt: Date.now(),
});
```

其他配置 : `choiceNode`, (中文(简体) ), `inputNode`, (中文(简体) ), `ratingNode`, (中文(简体) ), `chainNode`, (中文(简体) ), `managementChain`他们都归来一个化身 `name` 已经设定,所以 `config.name` 是一个 `string`。通知者: `ConsoleNotifier`, (中文(简体) ), `WebhookNotifier`, (中文(简体) ), `slackNotifier`, (中文(简体) ), `CompositeNotifier`, (中文(简体) ), `filteredNotifier`, (中文(简体) ), `priorityRouter`。 。 。 . `FileApprovalStore` 持续的请求; `RedisApprovalStore` 财务报告和审定财务报表 `PostgresApprovalStore` 在进程之间分享.

等待在重启中幸存下来。 请求的ID来自运行和问题,所以重启后再次运行的人类节点会发现自己的请求而不是两次询问,在进程倒下时会得到一个给出的答案,并保持原来的最后期限. 升级也一样 循环中再次访问的节点询问了改变的状态,从而打开了新的请求. 启动时, `await manager.recoverRuns({ approvalStore })` 恢复运行所停止的运行进程,从他们最后一个检查点运行或等待(从运行这些工作流程的一个过程调用).

在每个商店(从本质上讲跨越Redis和Postgres的流程):第一个 `submitResponse` 胜负相投 `ApprovalAlreadyAnsweredError` 背着这个答案 一个人类节点,在有人回答后, 点出时间, 保持答案。 删除(或到期)一个没人回答的请求会撤回:任何过程中的服务员都会得到一个退出,节点会用 `withdrawn: true` 而不是永远等待。 暂停或退出从不作为批准,对于 `multi-choice` 请求.

```typescript
import { ApprovalAlreadyAnsweredError } from '@cogitator-ai/workflows';

try {
  await approvalStore.submitResponse({
    requestId,
    decision: false,
    respondedBy: 'cfo@company.com',
    respondedAt: Date.now(),
  });
} catch (error) {
  if (error instanceof ApprovalAlreadyAnsweredError) {
    console.log('Already decided:', error.existing?.decision);
  }
}

await approvalStore.deleteRequest(staleRequestId); // a pending request is withdrawn
```

* * *

## 地图

```typescript
import { mapReduceNode, mapReduceWorkflowNode } from '@cogitator-ai/workflows';

builder.addNode(
  'score-documents',
  mapReduceWorkflowNode(
    mapReduceNode<DocsState, number, number>('score-documents', {
      map: {
        items: (state) => state.documents,
        mapper: async (doc) => scoreDocument(doc as string),
        concurrency: 5,
        continueOnError: true,
      },
      reduce: {
        initial: 0,
        reducer: (sum, item) => sum + item.result,
        finalize: (sum, state) => sum / state.documents.length,
      },
    }),
    { stateMapper: (result) => ({ averageScore: result.reduced }) }
  )
);
```

地图员的第四个论点是 `{ signal, attempt }`: 当尝试时间结束时, 信号中止, 其它项目失败地图( 没有 `continueOnError`)或运行被取消或暂停,所以传递给映射器开始的工作. 第一次失败停止了地图: 没有其他项目开始, 错误是 `partialResults` 保持每件东西的真正作用.

`mapNode` + 键 `mapWorkflowNode`, (中文(简体) ), `parallelMap`, (中文(简体) ), `batchedMap` 和减少器的预设 `collect`, (中文(简体) ), `sum`, (中文(简体) ), `count`, (中文(简体) ), `groupBy`, (中文(简体) ), `partition`, (中文(简体) ), `flatMap` 财务报告和审定财务报表 `stats` 覆盖其他形状.

* * *

## 触发器

```typescript
import { createTriggerManager, cronTrigger, webhookTrigger } from '@cogitator-ai/workflows';

const triggers = createTriggerManager({
  onTriggerFire: async (trigger, context) => {
    const run = await manager.schedule(workflows[trigger.workflowName], {
      input: { payload: context.payload },
    });
    return run;
  },
});
await triggers.start(); // brings back the triggers of its store, disabled ones stay disabled

// registering the same trigger again (on every boot) takes over the one the store brought back,
// one whose filter or transform is another function stays a trigger of its own

await triggers.register({
  workflowName: 'daily-report',
  type: 'cron',
  config: cronTrigger('0 9 * * *', { timezone: 'Europe/Berlin' }),
  enabled: true,
});

await triggers.register({
  workflowName: 'github-sync',
  type: 'webhook',
  config: webhookTrigger('/hooks/github', 'POST', {
    auth: { type: 'hmac', secret: process.env.GH_SECRET! },
  }),
  enabled: true,
});

// In your HTTP handler:
const response = await triggers.handleWebhook({
  path: req.path,
  method: req.method,
  headers,
  body,
});
```

* * *

### 常规作为工作流程(实验性)

核心触发管理器的常规( A)`experimental_triggers`)可以运行一个工作流程而不是代理. `experimental_routineWorkflow` 运行在射击的运行代号下 与常规的提示 和事件在它的状态: 其人类节点等待批准 进入收件箱 并得到管理者的回答 `respond()` (或来自 Studio),复试后最终失败的节点会进入收件箱和死信队列,补偿会像任何saga一样运行,结果会作为常规的结果或失败到达收件箱.

```typescript
import { experimental_routineWorkflow } from '@cogitator-ai/workflows';

const cog = new Cogitator({
  experimental_triggers: {
    workflows: {
      booking: experimental_routineWorkflow({ workflow: bookingWorkflow, runtime: engine }),
    },
    routines: [
      {
        id: 'booking',
        prompt: 'Book the trip the request asks for.',
        target: { workflow: 'booking' },
        triggers: [{ kind: 'api' }],
      },
    ],
  },
});

const [approval] = await cog.experimental_triggers!.inbox.list({ kind: 'approval' });
await cog.experimental_triggers!.respond(approval.id, { decision: true });
```

每次射击都运行工作流程从 `{ input, trigger }`数字 : `input` 是例行公事的提示, 将事件包裹为不可信数据, `trigger` 管理者担保什么(事件ID,来源,触发,时间,发送者如何检查). 事件的有效载荷和文本只到达内部状态 `input`,或者通过 `input(context)` 自己写的 `context.event`。 。 。 .

`experimental_inboxNotifier(inbox)` (单位:千美元) `ApprovalNotifier`和 `experimental_inboxDeadLetterQueue(inbox, dlq)` 将任何工作流程的核准和失败放入收件箱。 通知人每个请求保留一个项目(`experimental_approvalItemId(requestId)`),请求的受让人项目(常规用户为 `metadata.routineUserId`:超时或升级将其关闭 `failure`,一个升级或授权请求作为其自身 `approval` 因此,无法从两个项目中回答任何请求。 回答同时被答复或超时的请求时 `INBOX_ITEM_ANSWERED`。 。 。 .

* * *

## 观察性

```typescript
import { createTracer, createMetricsCollector } from '@cogitator-ai/workflows';

const tracer = createTracer({
  enabled: true,
  serviceName: 'billing-workflows',
  exporter: 'otlp', // 'console' | 'otlp' | 'jaeger' | 'zipkin' | 'noop'
  exporterEndpoint: 'http://localhost:4318/v1/traces',
});
const metricsCollector = createMetricsCollector({ prefix: 'cogitator_workflow' });

await executor.execute(workflow, input, { tracer, metricsCollector });
await tracer.flush();

console.log(metricsCollector.getWorkflowMetrics('report'));
console.log(metricsCollector.toPrometheusFormat());
```

执行器在每个节点执行时创建一个工作流程跨度和一个孩子跨度(平行节点保持正确的家长)并记录工作流程/节点计数,延迟和重复. `sampleRate` (0-1) 样本痕迹; 有 0, `tracer.isSampled()` 实值 `false` 没有出口.

* * *

## 工作流程管理

```typescript
import { createWorkflowManager, createFileRunStore } from '@cogitator-ai/workflows';

const manager = createWorkflowManager({
  cogitator,
  runStore: createFileRunStore({ directory: './runs' }),
  maxConcurrency: 10,
  defaultTimeout: 300_000, // runs exceeding it are cancelled and marked failed
  tracer,
  metrics: metricsCollector,
});
manager.start();

const result = await manager.execute(workflow, input, { tags: ['nightly'] });

const runId = await manager.schedule(workflow, { at: Date.now() + 60_000, input, priority: 1 });
await manager.cancel(runId);

const runs = await manager.listRuns({ status: 'failed', workflowName: 'report', limit: 20 });
const stats = await manager.getStats('report');
await manager.retry(runs[0].id);

const jobId = manager.registerCronJob(workflow, '0 2 * * *', {
  jobOptions: { input, maxRetries: 2 },
});
```

`registerCronJob` 在管理器启动时排队运行每起事件( e)`schedule({ cron })` 仅排入下一个; `unregisterCronJob`, (中文(简体) ), `setCronJobEnabled` 财务报告和审定财务报表 `getCronJobs` 管理工作。 预定的运行得到经理的检查站商店,追踪器,度量器和他们的 `timeout` (次数) `defaultTimeout`),以及 `maxRetries` 自动重试一个失败的计划运行 。 与一个 `checkpointStore`, (中文(简体) ), `pause(runId)` 中止运行并保留运行 `paused`,以及 `resume(runId, options?)` 在最后一个检查站继续检查.

每次运行——执行,安排或由触发器启动——记录 `currentNodes`, (中文(简体) ), `completedNodes` 财务报告和审定财务报表 `failedNodes`,在运行完成时全部写入。 运行仓库返回副本,所以更改返回的运行不会改变存储的运行; `listRuns()` 被排序为最新的第一. `getStats()` 被取消的数既不是成功率,也不是失败率。 使用 `PostgresRunStore` 编号 : `RedisRunStore` 以在进程之间共享运行:在数据库(Postgres)或排序集索引(Redis)中运行的查询、计数和数据.

* * *

## 许可证

麻省理工学院

## 常见问题

### @cogitator-ai/workflows 是什么？

基于 Cogitator 代理的 DAG 工作流程引擎

### 如何安装 @cogitator-ai/workflows？

在项目目录运行 npm install @cogitator-ai/workflows，或使用 pnpm add / yarn add / bun add。

### @cogitator-ai/workflows 支持 TypeScript 吗？

支持。包内自带 TypeScript 类型声明（package\.json 声明了 types 入口），安装后即可获得类型提示，无需再安装 @types 包。

### @cogitator-ai/workflows 支持 ESM 还是 CommonJS？

以 ES Module 形式发布，推荐使用 import 引入。在 CommonJS 项目中，Node\.js 20\.19\+ / 22\.12\+ 可以直接 require\(\) 不含顶层 await 的 ES Module，更早的版本需要改用动态 import\(\)。

### @cogitator-ai/workflows 需要什么版本的 Node\.js？

package\.json 的 engines 字段要求 Node\.js \>=22\.12\.0。版本不符时 npm 默认只给出警告，开启 engine-strict 后会拒绝安装。

### @cogitator-ai/workflows 可以商用吗？

@cogitator-ai/workflows 使用 MIT 许可。宽松许可：允许商业使用、修改和再分发，通常只需在分发时保留原版权与许可声明。以上为概括，不构成法律意见，请以许可原文为准。

### @cogitator-ai/workflows 的下载量有多少？

根据 npm 官方下载统计，@cogitator-ai/workflows 在 2026/10/11 – 2026/11/09 这 30 天内被下载 12,014 次，日均约 400 次。下载量会把 CI 构建、镜像同步和作为其他包依赖的安装一并计入，适合横向比较热度，不能直接等同于使用人数。

### @cogitator-ai/workflows 还在维护吗？

最新版本 0\.15\.6 发布于 2026/10/29（12 天前）；至今共发布 109 个版本，其中过去 12 个月发布了 109 个；首次发布于 2025/12/31。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。