第6章 工作流编排

6.1 问题场景

当业务流程包含多步校验、外部调用、条件分支、人工确认和结果回写时，单轮对话很难稳定完成任务。 如果没有编排机制，系统很容易出现：

• 步骤漏执行。
• 状态丢失。
• 外部依赖失败后无法恢复。
• 用户只看到一个聊天结果，却不知道流程到底做到哪一步。

很多前端工程师第一次做 Agent 工作流时，常见误区是：

• 把所有逻辑都写在一个大函数里。
• 让模型自己记住当前进度。
• 失败时从头重跑，而不是只恢复出错节点。
• 前端只展示最终结果，不展示过程状态。

一旦流程从“查一个数据”升级到“读一份文档 -> 提取任务 -> 等待负责人确认 -> 写回系统”，单 Agent 的简单调用就不够了。 你需要把系统升级成一个真正可编排、可追踪、可恢复的工作流。

前端迁移提示

这一章其实和前端工程高度相关，尤其适合前端背景来做：

• 组件状态流转 -> 工作流节点状态机。
• 页面步骤条 / 向导页 -> 任务面板 / 节点执行轨迹。
• 异步请求管理 -> 节点重试、超时、取消与补偿。
• 权限与交互控制 -> 人工确认节点与高风险操作控制。

最小项目目标

本章建议先完成一个“多步骤任务最小闭环”项目，例如：

• 会议纪要整理 -> 待办拆解 -> 人工确认 -> 导出任务。
• 线索初筛 -> 摘要生成 -> 分级建议 -> 等待审批。
• 需求受理 -> 信息补全 -> 分类 -> 分派建议。

参考入口：

• 示例3 工作流状态机
• 章节代码片段索引

6.2 核心原理

工作流状态图

工作流编排的关键，不是把步骤串起来而已，而是把“节点职责、状态流转、异常处理、人工介入和结果回放”做成可控系统。 如果只把它理解成“多次调用模型的串联”，系统会很快失控。

一、状态驱动：每一步都基于明确状态

工作流和普通聊天最大的区别，是它有明确状态。 一个最小节点状态通常至少包含：

• pending：等待执行。
• running：正在执行。
• succeeded：执行成功。
• failed：执行失败。
• waiting_human：等待人工确认。
• cancelled：已取消。

你可以直接参考：

• 示例3 工作流状态机
• docs/examples/assets/samples/example-03-workflow-state.json

二、节点分工：把复杂任务拆成独立责任边界

工作流里最常见的错误，是把解析、检索、规划、执行、回写、通知全塞进一个节点。 更稳妥的方式是按职责拆分，例如：

• 读取输入材料。
• 提取结构化信息。
• 生成任务建议。
• 等待负责人确认。
• 导出或写回外部系统。

节点边界清晰后，你才能：

• 看懂当前流程做到哪一步。
• 单独重试某个节点。
• 为不同节点设计不同超时和权限策略。

三、失败治理：不是只关注成功路径

真实工作流系统一定会失败，区别只是失败后能不能控。 因此你必须提前定义：

• 节点失败后是重试、跳过、回退还是转人工。
• 高风险动作失败后是否需要补偿。
• 节点失败时前端展示什么。
• 用户是否能继续、取消或手动接管。

四、人工确认：工作流不是全自动才高级

很多流程里，人工确认不是“系统不够强的补丁”，而是产品设计的一部分。 尤其在这些场景里必须考虑 waiting_human：

• 写入真实业务系统。
• 发送通知、消息、邮件。
• 对内容做最终定稿或审批。
• 风险高、置信度低、结果影响面大。

五、前端工作台：把执行过程做成可理解界面

工作流 Agent 不适合只有一个聊天框。 更合适的交互结构通常是：

• 对话输入区。
• 节点状态区。
• 结果与证据区。
• 控制区（继续、确认、取消、重试、转人工）。

这正是前端工程师最能体现优势的地方。

6.3 实操步骤

推荐做法是先做一个 4 到 6 节点的最小工作流，而不是一开始就上复杂框架或多 Agent。 先把节点、状态、确认和失败恢复做实，后续再增加更复杂的分支与治理能力。

步骤 1：先画出最小流程图

建议至少画清楚：

• 入口是什么。
• 中间有哪些处理节点。
• 哪一步需要人工确认。
• 哪一步会写回系统。
• 失败后怎么处理。

如果你连流程图都画不清，代码通常也会越写越乱。

步骤 2：为每个节点定义输入输出契约

至少要写清楚：

• 节点名称。
• 节点目标。
• 输入字段。
• 输出字段。
• 失败类型。
• 是否可重试。
• 是否需要人工确认。

步骤 3：先实现状态机，再实现花哨逻辑

推荐优先做：

1. 节点状态定义。
2. 节点之间的流转规则。
3. 当前流程是否可恢复。
4. 前端如何映射状态。

而不是一开始先写很多提示词（prompt）或复杂业务规则。

步骤 4：实现最小执行器

最小执行器建议具备：

• 顺序执行节点。
• 更新节点状态。
• 记录失败原因。
• 在 waiting_human 停下并等待用户操作。

参考入口：

• 示例3 工作流状态机

步骤 5：补人工确认节点

对确认节点，建议至少明确：

• 等待确认时前端显示什么。
• 用户可以点什么按钮。
• 拒绝确认后流程是否取消或回退。
• 确认后是否继续执行后续节点。

步骤 6：补失败恢复和证据材料

建议至少处理这些异常：

• 某个节点超时。
• 节点返回空结果。
• 工具执行失败。
• 人工确认长时间未完成。

同时建议保留这些证据材料：

• 状态样本：docs/examples/assets/samples/example-03-workflow-state.json
• 截图清单：docs/examples/assets/screenshots/README.md
• 片段索引

步骤 7：把工作流能力映射回项目线

如果你已经做通最小工作流，可以进一步映射：

• 第 6 章：工作流编排的最小闭环。
• 第 8 章：给工作流加评测和回归。
• 第 9 章：给工作流加上线、告警和回滚。
• 项目线 C：把它升级成多步骤业务工作流 Agent。

代码实战：可运行的工作流引擎

前面讲了状态驱动、节点分工、失败治理和人工确认的设计原则，这一节用可运行的代码把它们串起来。

最小工作流引擎（TypeScript）

下面这个引擎实现了"会议纪要 → 待办拆解 → 人工确认 → 导出"四节点工作流，包含状态流转、失败重试和人工确认三个核心能力。

// workflow-engine.ts — 最小工作流引擎
import OpenAI from 'openai'

const client = new OpenAI()

// 节点状态
type NodeStatus = 'pending' | 'running' | 'succeeded' | 'failed' | 'waiting_human' | 'cancelled'

// 节点定义
interface WorkflowNode {
  id: string
  name: string
  type: 'ai' | 'tool' | 'human_confirm'
  status: NodeStatus
  input?: Record<string, unknown>
  output?: Record<string, unknown>
  error?: string
  retryCount: number
  maxRetries: number
}

// 工作流定义
interface Workflow {
  id: string
  name: string
  nodes: WorkflowNode[]
  currentNodeIndex: number
  status: 'running' | 'completed' | 'failed' | 'waiting_human'
}

// 创建示例工作流：会议纪要 → 待办拆解 → 人工确认 → 导出
function createMeetingWorkflow(meetingNotes: string): Workflow {
  return {
    id: `wf-${Date.now()}`,
    name: '会议纪要处理工作流',
    currentNodeIndex: 0,
    status: 'running',
    nodes: [
      {
        id: 'extract',
        name: '提取关键信息',
        type: 'ai',
        status: 'pending',
        input: { meetingNotes },
        retryCount: 0,
        maxRetries: 2,
      },
      {
        id: 'split_tasks',
        name: '拆解待办事项',
        type: 'ai',
        status: 'pending',
        retryCount: 0,
        maxRetries: 2,
      },
      {
        id: 'human_review',
        name: '负责人确认',
        type: 'human_confirm',
        status: 'pending',
        retryCount: 0,
        maxRetries: 0,
      },
      {
        id: 'export',
        name: '导出任务清单',
        type: 'tool',
        status: 'pending',
        retryCount: 0,
        maxRetries: 1,
      },
    ],
  }
}

// AI 节点执行器
async function executeAINode(
  node: WorkflowNode,
  context: Record<string, unknown>,
): Promise<Record<string, unknown>> {
  const prompts: Record<string, string> = {
    extract: `从以下会议纪要中提取：参会人、讨论要点、决策事项、待办事项。输出 JSON 格式。\n\n会议纪要：\n${context.meetingNotes}`,
    split_tasks: `把以下会议要点拆解为具体的待办任务，每个任务包含：标题、负责人、截止日期建议、优先级。输出 JSON 数组。\n\n会议要点：\n${JSON.stringify(context.extractResult)}`,
  }

  const response = await client.chat.completions.create({
    model: 'gpt-4o-mini',
    messages: [
      { role: 'system', content: '你是一个专业的项目助手，输出结构化 JSON。' },
      { role: 'user', content: prompts[node.id] || '处理任务' },
    ],
    response_format: { type: 'json_object' },
    temperature: 0,
  })

  return JSON.parse(response.choices[0].message.content!)
}

// 工具节点执行器
async function executeToolNode(
  node: WorkflowNode,
  context: Record<string, unknown>,
): Promise<Record<string, unknown>> {
  console.log('[导出] 任务清单已写入系统')
  return {
    exported: true,
    taskCount: (context.tasks as unknown[])?.length ?? 0,
    timestamp: new Date().toISOString(),
  }
}

// 工作流引擎主循环
async function runWorkflow(workflow: Workflow) {
  console.log(`\n🚀 启动工作流: ${workflow.name}\n`)
  const context: Record<string, unknown> = {}

  if (workflow.nodes[0]?.input) {
    Object.assign(context, workflow.nodes[0].input)
  }

  while (workflow.currentNodeIndex < workflow.nodes.length) {
    const node = workflow.nodes[workflow.currentNodeIndex]
    node.status = 'running'
    console.log(`▶ [${node.id}] ${node.name} — 执行中...`)

    try {
      if (node.type === 'human_confirm') {
        node.status = 'waiting_human'
        workflow.status = 'waiting_human'
        console.log(`⏸ [${node.id}] 等待人工确认...`)
        console.log('  当前待办列表:', JSON.stringify(context.tasks, null, 2))
        // 实际项目中这里会暂停，等待前端回调
        console.log('  ✅ 人工已确认，继续执行')
        node.status = 'succeeded'
        workflow.status = 'running'
      } else if (node.type === 'ai') {
        const result = await executeAINode(node, context)
        node.output = result
        node.status = 'succeeded'
        if (node.id === 'extract') context.extractResult = result
        if (node.id === 'split_tasks') context.tasks = result
      } else if (node.type === 'tool') {
        const result = await executeToolNode(node, context)
        node.output = result
        node.status = 'succeeded'
      }

      console.log(`✅ [${node.id}] ${node.name} — 完成\n`)
      workflow.currentNodeIndex++
    } catch (err) {
      node.retryCount++
      const message = (err as Error).message
      console.log(`❌ [${node.id}] 失败: ${message}（重试 ${node.retryCount}/${node.maxRetries}）`)

      if (node.retryCount <= node.maxRetries) {
        console.log(`🔄 [${node.id}] 重试中...\n`)
        continue
      }

      node.status = 'failed'
      node.error = message
      workflow.status = 'failed'
      console.log(`\n💥 工作流失败，停在节点: ${node.id}`)
      return
    }
  }

  workflow.status = 'completed'
  console.log(`\n🎉 工作流完成！共处理 ${workflow.nodes.length} 个节点`)
  workflow.nodes.forEach(n => {
    console.log(`  ${n.status === 'succeeded' ? '✅' : '❌'} ${n.name}`)
  })
}

// 运行示例
const workflow = createMeetingWorkflow(`
  参会人：张三、李四、王五
  讨论内容：
  1. 新版本 RAG 系统上线时间确定为下周三
  2. 李四负责完成向量数据库迁移
  3. 王五负责前端对话界面改版
  4. 需要在周五前完成性能压测
`)

runWorkflow(workflow)

运行方式：

# 安装依赖
npm install openai tsx

# 设置 API Key（使用你自己的 key 或兼容服务）
export OPENAI_API_KEY="sk-..."

# 运行
npx tsx workflow-engine.ts

预期输出：

🚀 启动工作流: 会议纪要处理工作流

▶ [extract] 提取关键信息 — 执行中...
✅ [extract] 提取关键信息 — 完成

▶ [split_tasks] 拆解待办事项 — 执行中...
✅ [split_tasks] 拆解待办事项 — 完成

▶ [human_review] 负责人确认 — 执行中...
⏸ [human_review] 等待人工确认...
  当前待办列表: [ ... ]
  ✅ 人工已确认，继续执行
✅ [human_review] 负责人确认 — 完成

▶ [export] 导出任务清单 — 执行中...
[导出] 任务清单已写入系统
✅ [export] 导出任务清单 — 完成

🎉 工作流完成！共处理 4 个节点
  ✅ 提取关键信息
  ✅ 拆解待办事项
  ✅ 负责人确认
  ✅ 导出任务清单

如果某个 AI 节点调用失败（网络超时、模型返回格式异常等），引擎会自动重试最多 maxRetries 次。超过重试次数后工作流停在失败节点，可以排查后手动恢复。

前端工作流面板（React）

有了后端引擎，前端需要一个工作台把节点状态可视化。下面这个组件用步骤条展示每个节点的执行状态，并在 waiting_human 节点提供确认和拒绝按钮。

// WorkflowPanel.tsx — 工作流状态面板
import React from 'react'

type NodeStatus = 'pending' | 'running' | 'succeeded' | 'failed' | 'waiting_human' | 'cancelled'

interface WorkflowNode {
  id: string
  name: string
  status: NodeStatus
  error?: string
}

interface WorkflowPanelProps {
  workflowName: string
  nodes: WorkflowNode[]
  onConfirm?: (nodeId: string) => void
  onReject?: (nodeId: string) => void
  onRetry?: (nodeId: string) => void
}

const STATUS_CONFIG: Record<NodeStatus, { icon: string; color: string; label: string }> = {
  pending:       { icon: '⏳', color: '#9ca3af', label: '等待中' },
  running:       { icon: '⚙️', color: '#3b82f6', label: '执行中' },
  succeeded:     { icon: '✅', color: '#22c55e', label: '已完成' },
  failed:        { icon: '❌', color: '#ef4444', label: '失败' },
  waiting_human: { icon: '👤', color: '#f59e0b', label: '待确认' },
  cancelled:     { icon: '🚫', color: '#6b7280', label: '已取消' },
}

export function WorkflowPanel({
  workflowName,
  nodes,
  onConfirm,
  onReject,
  onRetry,
}: WorkflowPanelProps) {
  return (
    <div style={{ maxWidth: 480, margin: '0 auto', fontFamily: 'system-ui, sans-serif' }}>
      <h3 style={{ marginBottom: 16 }}>{workflowName}</h3>

      {nodes.map((node, index) => {
        const config = STATUS_CONFIG[node.status]
        const isLast = index === nodes.length - 1

        return (
          <div key={node.id} style={{ display: 'flex', gap: 12 }}>
            {/* 左侧时间线 */}
            <div style={{ display: 'flex', flexDirection: 'column', alignItems: 'center', width: 32 }}>
              <div
                style={{
                  width: 32,
                  height: 32,
                  borderRadius: '50%',
                  backgroundColor: config.color,
                  display: 'flex',
                  alignItems: 'center',
                  justifyContent: 'center',
                  fontSize: 16,
                  flexShrink: 0,
                }}
              >
                {config.icon}
              </div>
              {!isLast && (
                <div style={{ width: 2, flex: 1, backgroundColor: '#e5e7eb', minHeight: 24 }} />
              )}
            </div>

            {/* 右侧内容 */}
            <div style={{ flex: 1, paddingBottom: isLast ? 0 : 16 }}>
              <div style={{ display: 'flex', alignItems: 'center', gap: 8 }}>
                <span style={{ fontWeight: 600 }}>{node.name}</span>
                <span
                  style={{
                    fontSize: 12,
                    padding: '2px 8px',
                    borderRadius: 4,
                    backgroundColor: `${config.color}20`,
                    color: config.color,
                  }}
                >
                  {config.label}
                </span>
              </div>

              {node.error && (
                <div style={{ color: '#ef4444', fontSize: 13, marginTop: 4 }}>
                  错误：{node.error}
                </div>
              )}

              {/* 人工确认按钮 */}
              {node.status === 'waiting_human' && (
                <div style={{ display: 'flex', gap: 8, marginTop: 8 }}>
                  <button
                    onClick={() => onConfirm?.(node.id)}
                    style={{
                      padding: '6px 16px',
                      backgroundColor: '#22c55e',
                      color: '#fff',
                      border: 'none',
                      borderRadius: 4,
                      cursor: 'pointer',
                    }}
                  >
                    确认通过
                  </button>
                  <button
                    onClick={() => onReject?.(node.id)}
                    style={{
                      padding: '6px 16px',
                      backgroundColor: '#ef4444',
                      color: '#fff',
                      border: 'none',
                      borderRadius: 4,
                      cursor: 'pointer',
                    }}
                  >
                    拒绝
                  </button>
                </div>
              )}

              {/* 失败重试按钮 */}
              {node.status === 'failed' && (
                <button
                  onClick={() => onRetry?.(node.id)}
                  style={{
                    marginTop: 8,
                    padding: '6px 16px',
                    backgroundColor: '#3b82f6',
                    color: '#fff',
                    border: 'none',
                    borderRadius: 4,
                    cursor: 'pointer',
                  }}
                >
                  重试
                </button>
              )}
            </div>
          </div>
        )
      })}
    </div>
  )
}

使用方式：

<WorkflowPanel
  workflowName="会议纪要处理工作流"
  nodes={[
    { id: 'extract',      name: '提取关键信息', status: 'succeeded' },
    { id: 'split_tasks',  name: '拆解待办事项', status: 'succeeded' },
    { id: 'human_review', name: '负责人确认',   status: 'waiting_human' },
    { id: 'export',       name: '导出任务清单', status: 'pending' },
  ]}
  onConfirm={(id) => console.log('确认节点:', id)}
  onReject={(id) => console.log('拒绝节点:', id)}
  onRetry={(id) => console.log('重试节点:', id)}
/>

面板效果：每个节点在左侧有圆形状态图标，右侧显示节点名称和状态标签。waiting_human 节点会展示"确认通过 / 拒绝"按钮，failed 节点会展示"重试"按钮。

前端知识桥梁

如果你有前端背景，下面这张表帮你把工作流概念映射到已经熟悉的前端模式：

工作流概念	前端对应概念	说明
节点状态机	React useReducer / 有限状态机	节点的 pending → running → succeeded/failed 流转，和组件状态管理本质一样
节点顺序执行	Promise 链 / async-await 序列	引擎主循环就是一个 while + await，和前端处理多步异步请求相同
失败重试	axios retry / React Query retry	retryCount 和 maxRetries 的逻辑，React Query 用户应该很熟悉
人工确认	confirm 弹窗 / 审批流组件	waiting_human 状态等价于前端弹窗等待用户点击后才继续
工作台面板	步骤条（Stepper）/ 向导页 / Dashboard	上面的 WorkflowPanel 本质就是一个带状态的步骤条组件

做 Agent 工作流不需要学全新的东西，把你在前端已经掌握的异步控制、状态管理和交互设计搬过来就行。区别只是节点里跑的不是 API 请求，而是模型调用。

6.4 常见坑

坑 1：业务逻辑塞进单个节点

这样后续一旦要插入人工确认、失败回退或新增分支，改动范围会非常大。

坑 2：只关注成功路径

工作流最脆弱的地方恰恰在失败路径。 如果没有失败分类与恢复策略，线上一旦异常，系统会非常难救。

坑 3：没有统一状态机

如果前端和后端对“当前到底完成没完成”理解不一致，用户体验会很差。

坑 4：把人工确认当成后补功能

正确做法是从方案设计阶段就把人工确认节点定义进去，而不是最后再加一个弹窗。

坑 5：前端仍然只展示聊天结果

这样会把一个多步骤系统退化成黑盒，用户根本不知道系统做到哪一步。

坑 6：失败后只能从头开始

很多节点是可以局部重试的。如果每次失败都从头来，不仅浪费成本，也很伤体验。

6.5 验证方式

工作流编排做完后，不应只看“流程能不能跑通”，还要看它能不能在真实变化和失败中保持可控。

一、核心流程成功率

建议先验证：

• 高频场景下，整体流程是否能稳定执行完成。
• 每个节点的成功率是否合理。
• 关键节点是否存在明显长尾失败。

二、异常场景恢复能力

至少要覆盖：

• 节点失败后是否能进入预期回退或人工介入。
• 某节点失败时，前面成功节点的结果是否还能保留。
• 用户是否可以重试失败节点而不是整条链路。

三、流程可观测性

从日志与状态角度检查：

• 是否能看到每个节点状态变化。
• 是否能追踪哪一步耗时最长。
• 是否能快速定位失败原因。

四、前端工作台是否可理解

从用户体验角度检查：

• 当前步骤是否清晰可见。
• 等待确认时是否有明确入口。
• 失败时是否有下一步建议。
• 完成后是否有清晰交付结果。

五、证据材料

• 状态样本：docs/examples/assets/samples/example-03-workflow-state.json
• 截图清单：docs/examples/assets/screenshots/README.md
• 图解配图：docs/examples/assets/diagrams/chapter-06-workflow-state.svg
• 章节片段索引
• 示例骨架

6.6 面试表达

我把复杂业务流程拆成状态驱动的节点编排系统，并把失败治理、人工确认和前端工作台一起设计进去。 这样系统不只是能跑通主链路，而是能在异常情况下恢复、在关键节点等待确认、在界面上可视化整个任务过程。

实战案例：工单流转自动编排

• 背景：客服工单需要跨系统流转，人工处理耗时长且容易漏步骤。
• 动作：设计解析、分派、执行、回写四节点工作流，引入统一节点状态和失败回退逻辑，并在前端工作台展示执行轨迹。
• 结果：流程更稳定，失败可回放、可局部恢复，业务方也更容易理解系统当前状态。

推荐答题框架

如果面试官问“工作流和普通 Agent 的差别是什么”，可以按下面顺序回答：

1. 先讲为什么多步骤任务不能只靠一次回答解决。
2. 再讲节点拆分、状态机和失败治理。
3. 再讲人工确认和前端任务工作台。
4. 最后讲验证方式：流程成功率、恢复能力和可观测性。

6.7 练习任务

1. 选一个 4 到 6 步的业务流程，画出节点流转图并标明人工确认点。
2. 为每个节点定义输入输出结构、失败类型和是否可重试。
3. 设计一组最小状态枚举，至少覆盖 pending / running / succeeded / failed / waiting_human / cancelled。
4. 参考 示例3 工作流状态机，写出一个最小执行器骨架。
5. 设计一个前端任务面板，说明每个节点状态分别如何展示。
6. 运行本章"代码实战"中的 workflow-engine.ts，观察四个节点的执行顺序和状态变化。尝试故意让某个节点失败（比如传入空的会议纪要），验证重试和失败停止逻辑是否符合预期。
7. 基于 WorkflowPanel 组件，在自己的项目中渲染一个工作流面板，点击确认/拒绝按钮后更新节点状态。

6.8 验收清单

• 任务完成率 >= 85%
• 关键指标达标率 >= 90%
• 异常场景通过率 >= 90%
• 至少完成 1 个 4 节点以上的最小工作流闭环。
• 至少保留 1 份状态样本、1 组截图占位清单和 1 份章节代码片段索引。
• 至少能用 2 分钟讲清楚“为什么这个系统不是多次调用模型，而是一个状态驱动的工作流系统”。