17.12-工作流编排

要点

工作流是多个任务的有序组合——任务 A 完成后触发任务 B，B 完成后触发 C
简单工作流用链式调用——一个消费者完成后发消息到下一个队列
复杂工作流用 DAG（有向无环图）——支持并行分支和汇聚
工作流状态需要持久化——每个步骤的输入、输出、状态都要记录
失败处理要考虑回滚——前面的步骤成功了，后面的失败了怎么办

1. 什么是工作流

前面讲的异步任务是单个任务——上传文件、解析 PDF、生成 embedding。但真实业务经常需要多个任务串联：

用户上传文档
  ↓
[1] 解析文档 → 提取文本
  ↓
[2] 切 chunk → 分成段落
  ↓
[3] 生成 embedding → 向量化
  ↓
[4] 写入向量库 → 建立索引
  ↓
[5] 生成摘要 → 用 LLM 总结
  ↓
[6] 发送通知 → 告诉用户处理完成

这 6 个步骤构成一个工作流。每个步骤是独立的异步任务，步骤之间有依赖关系。

工作流的特点：

有序——步骤 2 必须等步骤 1 完成
有状态——每个步骤的输入是上一步的输出
可重试——某一步失败可以重试，不需要从头开始
可观测——需要知道当前执行到哪一步

2. 简单工作流：链式调用

最简单的实现：一个消费者完成后，发消息到下一个队列。

// 消费者 1：解析文档
export async function parseConsumer(batch: MessageBatch<ParseTask>, env: Env) {
  for (const msg of batch.messages) {
    const text = await parseDocument(msg.body.fileKey, env)
 
    // 完成后发消息到下一个队列
    await env.CHUNK_QUEUE.send({
      workflowId: msg.body.workflowId,
      text,
    })
 
    msg.ack()
  }
}
 
// 消费者 2：切 chunk
export async function chunkConsumer(batch: MessageBatch<ChunkTask>, env: Env) {
  for (const msg of batch.messages) {
    const chunks = chunkText(msg.body.text)
 
    await env.EMBED_QUEUE.send({
      workflowId: msg.body.workflowId,
      chunks,
    })
 
    msg.ack()
  }
}
 
// 消费者 3：生成 embedding
export async function embedConsumer(batch: MessageBatch<EmbedTask>, env: Env) {
  for (const msg of batch.messages) {
    const vectors = await generateEmbeddings(msg.body.chunks, env)
 
    await env.INDEX_QUEUE.send({
      workflowId: msg.body.workflowId,
      chunks: msg.body.chunks,
      vectors,
    })
 
    msg.ack()
  }
}

链式调用的优点：简单、解耦。缺点：步骤多了之后难以管理，失败时难以回滚。

3. 工作流状态

工作流状态需要持久化——记录每个步骤的执行情况。

CREATE TABLE workflows (
  id TEXT PRIMARY KEY,
  user_id TEXT NOT NULL,
  type TEXT NOT NULL,              -- 工作流类型
  status TEXT NOT NULL DEFAULT 'pending',
  current_step TEXT,               -- 当前步骤
  context TEXT NOT NULL,           -- 工作流上下文（JSON，步骤间传递数据）
  created_at TEXT NOT NULL,
  updated_at TEXT NOT NULL
);
 
CREATE TABLE workflow_steps (
  id TEXT PRIMARY KEY,
  workflow_id TEXT NOT NULL,
  step_name TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'pending',
  input TEXT,                      -- 步骤输入（JSON）
  output TEXT,                     -- 步骤输出（JSON）
  error TEXT,
  started_at TEXT,
  completed_at TEXT
);

interface WorkflowContext {
  workflowId: string
  userId: string
  fileKey?: string
  text?: string
  chunks?: Chunk[]
  vectors?: number[][]
  summary?: string
  [key: string]: unknown
}
 
async function updateWorkflowStep(
  env: Env,
  workflowId: string,
  stepName: string,
  status: 'pending' | 'processing' | 'completed' | 'failed',
  input?: unknown,
  output?: unknown,
  error?: string
): Promise<void> {
  const now = new Date().toISOString()
 
  await env.DB.prepare(`
    INSERT INTO workflow_steps (id, workflow_id, step_name, status, input, output, error, started_at, completed_at)
    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
    ON CONFLICT(id) DO UPDATE SET
      status = excluded.status,
      output = excluded.output,
      error = excluded.error,
      completed_at = excluded.completed_at
  `).bind(
    `${workflowId}:${stepName}`,
    workflowId,
    stepName,
    status,
    input ? JSON.stringify(input) : null,
    output ? JSON.stringify(output) : null,
    error ?? null,
    status === 'processing' ? now : null,
    status === 'completed' || status === 'failed' ? now : null,
  ).run()
 
  // 更新工作流当前步骤
  await env.DB.prepare(`
    UPDATE workflows SET current_step = ?, status = ?, updated_at = ? WHERE id = ?
  `).bind(stepName, status === 'failed' ? 'failed' : 'processing', now, workflowId).run()
}

4. 工作流引擎

一个通用的工作流引擎，按步骤执行：

interface WorkflowStep {
  name: string
  execute: (context: WorkflowContext, env: Env) => Promise<WorkflowContext>
}
 
class WorkflowEngine {
  private steps: WorkflowStep[] = []
 
  addStep(step: WorkflowStep): this {
    this.steps.push(step)
    return this
  }
 
  async run(workflowId: string, initialContext: WorkflowContext, env: Env): Promise<void> {
    let context = initialContext
 
    for (const step of this.steps) {
      // 检查是否已完成（断点续传）
      const stepRecord = await env.DB.prepare(`
        SELECT status, output FROM workflow_steps WHERE id = ?
      `).bind(`${workflowId}:${step.name}`).first()
 
      if (stepRecord?.status === 'completed') {
        // 已完成，读取输出继续
        context = { ...context, ...JSON.parse(stepRecord.output) }
        continue
      }
 
      try {
        await updateWorkflowStep(env, workflowId, step.name, 'processing', context)
 
        context = await step.execute(context, env)
 
        await updateWorkflowStep(env, workflowId, step.name, 'completed', context, context)
      } catch (err) {
        await updateWorkflowStep(env, workflowId, step.name, 'failed', context, undefined, String(err))
        throw err
      }
    }
 
    // 所有步骤完成
    await env.DB.prepare(`
      UPDATE workflows SET status = 'completed', updated_at = ? WHERE id = ?
    `).bind(new Date().toISOString(), workflowId).run()
  }
}

使用：

// 定义工作流
const documentWorkflow = new WorkflowEngine()
  .addStep({
    name: 'parse',
    execute: async (ctx, env) => {
      const text = await parseDocument(ctx.fileKey!, env)
      return { ...ctx, text }
    },
  })
  .addStep({
    name: 'chunk',
    execute: async (ctx, env) => {
      const chunks = chunkText(ctx.text!)
      return { ...ctx, chunks }
    },
  })
  .addStep({
    name: 'embed',
    execute: async (ctx, env) => {
      const vectors = await generateEmbeddings(ctx.chunks!, env)
      return { ...ctx, vectors }
    },
  })
  .addStep({
    name: 'index',
    execute: async (ctx, env) => {
      await indexVectors(ctx.chunks!, ctx.vectors!, env)
      return ctx
    },
  })
 
// 消费者调用
export async function workflowConsumer(
  batch: MessageBatch<{ workflowId: string }>,
  env: Env
): Promise<void> {
  for (const msg of batch.messages) {
    const workflow = await env.DB.prepare(`
      SELECT * FROM workflows WHERE id = ?
    `).bind(msg.body.workflowId).first()
 
    const context = JSON.parse(workflow.context) as WorkflowContext
 
    try {
      await documentWorkflow.run(msg.body.workflowId, context, env)
      msg.ack()
    } catch (err) {
      msg.retry({ delaySeconds: 30 })
    }
  }
}

5. DAG 工作流

复杂工作流有并行分支和汇聚：

步骤 A → 步骤 B ─┐
       → 步骤 C ─┼→ 步骤 D
       → 步骤 E ─┘

步骤 B、C、E 并行执行，全部完成后才能执行步骤 D。

interface DAGStep {
  name: string
  dependsOn: string[]          // 依赖的步骤
  execute: (context: WorkflowContext, env: Env) => Promise<WorkflowContext>
}
 
class DAGWorkflowEngine {
  private steps: Map<string, DAGStep> = new Map()
 
  addStep(step: DAGStep): this {
    this.steps.set(step.name, step)
    return this
  }
 
  async run(workflowId: string, initialContext: WorkflowContext, env: Env): Promise<void> {
    const completed = new Set<string>()
    let context = initialContext
 
    while (completed.size < this.steps.size) {
      // 找出所有依赖已完成的步骤
      const ready = [...this.steps.values()].filter(
        (step) =>
          !completed.has(step.name) &&
          step.dependsOn.every((dep) => completed.has(dep))
      )
 
      if (ready.length === 0) {
        throw new Error('No ready steps - possible circular dependency')
      }
 
      // 并行执行所有就绪步骤
      const results = await Promise.all(
        ready.map(async (step) => {
          await updateWorkflowStep(env, workflowId, step.name, 'processing')
          const output = await step.execute(context, env)
          await updateWorkflowStep(env, workflowId, step.name, 'completed', context, output)
          return { step: step.name, output }
        })
      )
 
      // 合并结果
      for (const { output } of results) {
        context = { ...context, ...output }
      }
 
      // 标记完成
      for (const step of ready) {
        completed.add(step.name)
      }
    }
 
    await env.DB.prepare(`
      UPDATE workflows SET status = 'completed', updated_at = ? WHERE id = ?
    `).bind(new Date().toISOString(), workflowId).run()
  }
}

使用：

const dagWorkflow = new DAGWorkflowEngine()
  .addStep({
    name: 'A',
    dependsOn: [],
    execute: async (ctx, env) => { /* ... */ return ctx },
  })
  .addStep({
    name: 'B',
    dependsOn: ['A'],
    execute: async (ctx, env) => { /* ... */ return ctx },
  })
  .addStep({
    name: 'C',
    dependsOn: ['A'],
    execute: async (ctx, env) => { /* ... */ return ctx },
  })
  .addStep({
    name: 'D',
    dependsOn: ['B', 'C'],
    execute: async (ctx, env) => { /* ... */ return ctx },
  })

6. 失败和回滚

步骤失败时，前面的步骤已经成功了。需要考虑回滚。

interface WorkflowStep {
  name: string
  execute: (context: WorkflowContext, env: Env) => Promise<WorkflowContext>
  rollback?: (context: WorkflowContext, env: Env) => Promise<void>  // 回滚操作
}
 
class WorkflowEngine {
  async run(workflowId: string, initialContext: WorkflowContext, env: Env): Promise<void> {
    const executedSteps: WorkflowStep[] = []
    let context = initialContext
 
    try {
      for (const step of this.steps) {
        await updateWorkflowStep(env, workflowId, step.name, 'processing')
 
        context = await step.execute(context, env)
        executedSteps.push(step)
 
        await updateWorkflowStep(env, workflowId, step.name, 'completed')
      }
    } catch (err) {
      // 回滚已执行的步骤（逆序）
      for (const step of executedSteps.reverse()) {
        if (step.rollback) {
          try {
            await step.rollback(context, env)
            await updateWorkflowStep(env, workflowId, `${step.name}:rollback`, 'completed')
          } catch (rollbackErr) {
            console.error(`Rollback failed for ${step.name}:`, rollbackErr)
          }
        }
      }
 
      throw err
    }
  }
}

不是所有步骤都能回滚。发邮件、调外部 API 通常无法撤回。对于这类步骤，用补偿操作替代回滚——比如发一封「取消」邮件。

7. 超时处理

步骤可能卡住——外部 API 不响应、死循环。需要超时控制。

async function withTimeout<T>(
  promise: Promise<T>,
  timeoutMs: number,
  stepName: string
): Promise<T> {
  return Promise.race([
    promise,
    new Promise<never>((_, reject) =>
      setTimeout(() => reject(new Error(`Step ${stepName} timeout after ${timeoutMs}ms`)), timeoutMs)
    ),
  ])
}
 
// 使用
async function executeWithTimeout(step: WorkflowStep, context: WorkflowContext, env: Env) {
  return withTimeout(step.execute(context, env), 60000, step.name)  // 60 秒超时
}

8. 工作流查询

// 查询工作流状态
app.get('/api/workflows/:id', async (c) => {
  const [workflow, steps] = await Promise.all([
    c.env.DB.prepare('SELECT * FROM workflows WHERE id = ?').bind(c.req.param('id')).first(),
    c.env.DB.prepare('SELECT * FROM workflow_steps WHERE workflow_id = ? ORDER BY started_at').bind(c.req.param('id')).all(),
  ])
 
  return c.json({ workflow, steps: steps.results })
})

总结

回顾这一节的要点：

工作流是多个任务的有序组合——步骤之间有依赖关系
简单工作流用链式调用——一个消费者完成后发消息到下一个队列
复杂工作流用 DAG——支持并行分支和汇聚
工作流状态需要持久化——每个步骤的输入、输出、状态都记录在 D1
断点续传：从已完成的步骤继续执行，不需要从头开始
失败处理：回滚已执行的步骤（逆序），或执行补偿操作
超时控制：给每个步骤设超时，避免卡住

下一篇讲 Hono 作为任务调度入口——整合 HTTP、Queue、Cron。