@cogitator-ai/ 工作流程

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

安装

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

网站文件 : 工作流量, (中文(简体) ), 构建器, (中文(简体) ), 节点, (中文(简体) ), 执行, (中文(简体) ), 模式, (中文(简体) ), 萨加斯, (中文(简体) ), 时间安排。 。 。 .

特征

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

快速启动

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);

核心概念

工作流量

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 } 以动态方式运行.

工作流程执行器

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 节点超时.

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

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,
});

实时流线

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。 。 。 .

预建的节点

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和工具呼叫.

分行、循环和加入

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'] 是一个候选分支.

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 是工作流程的切入点 :

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。 。 。 .

检查站

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 池边:

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 年长的检查站的索引条目:

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

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

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

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)如此多 TimerManagers可以共享一个商店而无需发射两次计时器,一个坠毁的工人的计时器一旦索赔期满就会回来.

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

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) 并允许处理上述索赔.

计时器

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 而不是放弃它.

萨加图案

重试

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 });

断路器

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 当运行的后期节点失败时,执行器会将其卷回:

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数字 :

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观察每走一步.

死信队列和同上

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,作为第三个论点:

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()数字 :

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,
});

子工作流程

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 涵盖其他常见模式.

人类圈

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: '[email protected]',
        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: '[email protected]',
  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 请求.

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

try {
  await approvalStore.submitResponse({
    requestId,
    decision: false,
    respondedBy: '[email protected]',
    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

地图

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 覆盖其他形状.

触发器

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一样运行,结果会作为常规的结果或失败到达收件箱.

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。 。 。 .

观察性

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 没有出口.

工作流程管理

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)中运行的查询、计数和数据.

许可证

麻省理工学院