工作流Agent架构
要点
- 工作流 Agent 按预定义步骤执行任务，步骤转移在运行前确定；ReAct 循环里 LLM 每一步都重新决策
- 工作流适合步骤固定、可审计、需要人工介入的场景；自主 Agent 适合探索性强、步骤事先无法确定的场景
- 四种基础模式——顺序执行、并行分叉/汇聚、条件分支、人工审批——可以用状态机统一建模
- 用 TypeScript 实现工作流引擎，核心是状态机驱动的步骤调度器，每步执行完持久化状态
- 步骤越多，单步失败的概率累积越高，错误处理和重试策略在工作流中不可少
1. 什么是工作流Agent
前面几篇的 Agent 一直是 ReAct 循环：LLM 看当前状态，决定下一步，执行工具，拿结果，再决定下一步。每一步都交给 LLM 现场推理。
这种方式灵活性高，但有两个问题：
- 不可预测 — 同一请求跑两次，LLM 可能走不同的路径。财务审批、内容审核这类业务很难接受这种不确定性
- 难以审计 — 出了问题只能看 LLM 推理日志，搞清楚它「为什么」走了某条路径，但决策理由未必可靠
工作流 Agent 换了思路：把执行流程从 LLM 推理中抽出来，变成预定义步骤。 LLM 仍参与某些步骤（生成内容、判断条件），但步骤之间的转移顺序是确定的。
自主 Agent（ReAct）：
LLM -> 工具 -> LLM -> 工具 -> ... -> 回复
（每一步由 LLM 决定）
工作流 Agent：
步骤A -> 步骤B -> [条件判断] -> 步骤C 或 D -> 步骤E
（步骤顺序预定义，LLM 只在特定步骤参与）工作流 Agent 像一条流水线——每个工位做什么、做完传给谁，开工前就排好了。LLM 是某些工位上负责语言理解和生成的工人，其他工位是普通代码逻辑。
2. 工作流 vs 自主Agent的取舍
两种架构不是替代关系，适用场景不同。
工作流更适合的场景：
- 步骤可以穷举 — 流程在开发阶段就能画出来
- 需要审计和合规 — 每一步必须可追溯
- 涉及人工介入 — 金额超阈值要主管审批，流程必须能暂停
- 对延迟和成本有预算 — LLM 调用次数上限在定义阶段就能确定
自主 Agent 更适合的场景：
- 步骤无法确定 — 查哪些数据源、查几次取决于中间结果
- 任务边界模糊 — Agent 需要自己判断范围
- 需要动态探索 — 根据错误信息逐步缩小范围
实际项目中两者经常混用。工作流的某个步骤内部可能跑着一个小型 ReAct 循环；反过来，自主 Agent 的某些步骤也可能切换到预定义流程。
3. 常见工作流模式
复杂工作流通常是几种基础模式的组合。
顺序执行
步骤 A 做完传给 B，B 做完传给 C，依次执行。每一步的输出是下一步的输入。
async function runSequential(steps: Step[], input: unknown): Promise<unknown> {
let result = input
for (const step of steps) {
result = await step.execute(result)
}
return result
}并行分叉与汇聚
没有依赖关系的步骤可以同时执行。分叉点拆分并行分支，汇聚点等全部完成后再继续。
┌-> [摘要生成] ─┐
[获取文章全文] ──┼-> [关键词提取] ─┼-> [合并结果] -> [发送邮件]
└-> [情感分析] ─┘并行分支之间不能共享可变状态。如果两个分支都需要写同一个对象，得在汇聚点统一处理。
条件分支
根据上一步结果决定走哪条路径。条件可以是代码逻辑，也可以交给 LLM 判断。交给 LLM 时，Prompt 中要列出所有分支和判断标准，要求返回固定枚举值（比如 "refund" 或 "regular"），而不是自由文本，代码才能可靠路由。
人工审批节点
工作流暂停，等待人工确认后再继续。涉及资金操作、内容发布时几乎不可少。实现方式是持久化工作流状态到数据库，审批节点设为 waiting。外部系统通过 API 提交审批结果，工作流从暂停处恢复。
// 审批节点返回 waiting，引擎保存状态后中断执行
async function humanReview(input: unknown): Promise<StepResult> {
return { status: 'waiting', output: input }
}4. 工作流引擎实现
把前面的模式统一起来，需要一个引擎：接收工作流定义，按转移规则执行步骤，管理重试和超时。
类型定义
// src/workflow/types.ts
type StepResult = {
status: 'success' | 'failed' | 'waiting'
output: unknown
error?: string
}
type Step = {
id: string
name: string
execute: (input: unknown, ctx: WorkflowContext) => Promise<StepResult>
retry?: { maxAttempts: number; delayMs: number }
timeoutMs?: number
}
type Transition = {
from: string
to: string
condition?: (output: unknown) => Promise<boolean>
}
type WorkflowDefinition = {
id: string
steps: Step[]
transitions: Transition[]
startStepId: string
endStepIds: string[]
}
type WorkflowContext = {
workflowId: string
stepOutputs: Map<string, unknown>
metadata: Record<string, unknown>
}执行引擎
引擎核心是一个循环：从当前步骤找匹配的转移规则，确定下一步，执行，直到到达结束步骤。
// src/workflow/engine.ts
export class WorkflowEngine {
private definition: WorkflowDefinition
constructor(definition: WorkflowDefinition) {
this.definition = definition
}
async run(initialInput: unknown) {
const context: WorkflowContext = {
workflowId: crypto.randomUUID(),
stepOutputs: new Map(),
metadata: {},
}
let currentStepId = this.definition.startStepId
let output: unknown = initialInput
while (true) {
const step = this.definition.steps.find((s) => s.id === currentStepId)
if (!step) throw new Error(`步骤未找到：${currentStepId}`)
const result = await this.executeStep(step, output, context)
context.stepOutputs.set(step.id, result.output)
if (result.status === 'failed') return { status: 'failed', context }
if (result.status === 'waiting') return { status: 'waiting', context }
if (this.definition.endStepIds.includes(currentStepId)) {
return { status: 'completed', context }
}
const nextStepId = await this.resolveNextStep(currentStepId, result.output)
if (!nextStepId) return { status: 'failed', context }
currentStepId = nextStepId
output = result.output
}
}
private async resolveNextStep(stepId: string, output: unknown): Promise<string | null> {
for (const t of this.definition.transitions.filter((t) => t.from === stepId)) {
if (!t.condition || await t.condition(output)) return t.to
}
return null
}
private async executeStep(step: Step, input: unknown, ctx: WorkflowContext): Promise<StepResult> {
const maxAttempts = step.retry?.maxAttempts ?? 1
for (let attempt = 1; attempt <= maxAttempts; attempt++) {
try {
const result = step.timeoutMs
? await Promise.race([
step.execute(input, ctx),
new Promise<StepResult>((_, rej) =>
setTimeout(() => rej(new Error(`步骤 ${step.name} 超时`)), step.timeoutMs)),
])
: await step.execute(input, ctx)
if (result.status !== 'failed' || attempt >= maxAttempts) return result
} catch (error) {
if (attempt >= maxAttempts) {
return { status: 'failed', output: null, error: String(error) }
}
}
await sleep(step.retry?.delayMs ?? 1000)
}
return { status: 'failed', output: null, error: '不可达' }
}
}引擎做了四件事：按转移规则确定执行顺序、带重试的步骤执行、带超时的步骤执行、工作流暂停与恢复。
在 Hono 中接入
// src/routes/workflow.ts
import { Hono } from 'hono'
import { WorkflowEngine } from '../workflow/engine'
import { contentPublishWorkflow } from '../workflows/content-publish'
import type { AppEnv } from '../types'
const workflowApp = new Hono<AppEnv>()
workflowApp.post('/start', async (c) => {
const { topic } = await c.req.json<{ topic: string }>()
const engine = new WorkflowEngine(contentPublishWorkflow)
const state = await engine.run({ topic })
if (state.status === 'waiting') {
// 持久化状态到数据库，返回 workflowId 供后续恢复
await saveWorkflowState(c.env.DB, state)
return c.json({ workflowId: state.context.workflowId, status: 'waiting' })
}
return c.json({ workflowId: state.context.workflowId, status: state.status })
})
// 审批通过后恢复工作流
workflowApp.post('/resume', async (c) => {
const { workflowId } = await c.req.json<{ workflowId: string }>()
const saved = await loadWorkflowState(c.env.DB, workflowId)
if (!saved) return c.json({ error: '工作流不存在' }, 404)
// 从暂停步骤继续执行...
return c.json({ workflowId, status: 'resumed' })
})5. 状态持久化
前面的引擎在内存里跑完整个流程。如果 Worker 重启或崩溃，状态就丢了。人工审核可能要等几小时，这个问题必须处理。
解决办法是把每一步状态持久化到数据库。引擎每次只执行一个步骤，执行完写入数据库，下一步由另一个请求触发。
// src/workflow/store.ts
type PersistedState = {
id: string
definitionId: string
currentStepId: string
status: 'running' | 'completed' | 'failed' | 'waiting'
context: Record<string, unknown>
updatedAt: string
}
export async function saveWorkflowState(db: D1Database, state: PersistedState) {
await db.prepare(`
INSERT INTO workflow_states (id, definition_id, current_step_id, status, context, updated_at)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
current_step_id = excluded.current_step_id,
status = excluded.status,
context = excluded.context,
updated_at = excluded.updated_at
`).bind(
state.id, state.definitionId, state.currentStepId,
state.status, JSON.stringify(state.context), state.updatedAt,
).run()
}每一步执行完都有快照。Worker 崩溃后，下一个请求从数据库读状态继续跑，最多重复执行一个步骤。对有副作用的步骤（发邮件、写数据库），需要注意幂等性——同一个步骤执行两次，结果应该和执行一次一样。
人工审核的过期工作流也需要清理：
async function cleanupExpiredWorkflows(db: D1Database, maxWaitMs: number) {
const cutoff = new Date(Date.now() - maxWaitMs).toISOString()
await db.prepare(`
UPDATE workflow_states SET status = 'failed'
WHERE status = 'waiting' AND updated_at < ?
`).bind(cutoff).run()
}6. 错误处理与重试策略
工作流步骤越多，单步失败的概率累积越高。10 个步骤、每步 99% 成功率，整个工作流成功率只有 0.99^10 ≈ 90.4%。
重试策略
不同步骤适合不同的方式：
- 固定间隔 — 外部 API 限流，每次等固定时间
- 指数退避 — 网络抖动，间隔逐次翻倍（1s, 2s, 4s...）
- 不重试 — 幂等性无法保证的操作（发邮件重复发送会造成困扰）
引擎的 executeStep 已经读取步骤的 retry 配置做重试。没配置重试的步骤失败后直接标记工作流为 failed。
失败后的处理
步骤失败后有三种策略：
- 终止 — 标记
failed，通知相关人员 - 跳过 — 标记
skipped，继续下一步。适合非关键步骤（通知发送失败不影响主流程） - 回退 — 回到指定步骤重新执行
可以在 Transition 上加一个 onFailure 字段配置失败后的转移：
type Transition = {
from: string
to: string
condition?: (output: unknown) => Promise<boolean>
onFailure?: 'terminate' | 'skip' | string // 回退到指定步骤 id
}7. 实际案例：退款处理Pipeline
电商平台处理退款申请，流程如下：
[拉取退款申请] -> [数据清洗] -> [风控检查] -> [金额超阈值？]
├── 是 -> [人工审批] -> [执行退款] -> [通知]
└── 否 -> [自动退款] ──────────────────┘const refundPipeline: WorkflowDefinition = {
id: 'refund-pipeline',
steps: [
{
id: 'fetch', name: '拉取退款申请',
async execute(): Promise<StepResult> {
return { status: 'success', output: await fetchPendingRefunds() }
},
},
{
id: 'clean', name: '数据清洗',
async execute(input): Promise<StepResult> {
return { status: 'success', output: (input as Refund[]).map(normalize) }
},
},
{
id: 'risk_check', name: '风控检查',
retry: { maxAttempts: 2, delayMs: 3000 },
timeoutMs: 10_000,
async execute(input): Promise<StepResult> {
const results = await checkRisk(input as Refund[])
return { status: 'success', output: results }
},
},
{
id: 'manual_review', name: '人工审批',
async execute(input): Promise<StepResult> {
return { status: 'waiting', output: input }
},
},
{
id: 'process', name: '执行退款',
async execute(input): Promise<StepResult> {
await processRefunds(input as Refund[])
return { status: 'success', output: input }
},
},
{
id: 'auto_refund', name: '自动退款',
async execute(input): Promise<StepResult> {
await processRefunds(input as Refund[])
return { status: 'success', output: input }
},
},
{
id: 'notify', name: '发送通知',
async execute(input): Promise<StepResult> {
await sendNotifications(input as Refund[])
return { status: 'success', output: { done: true } }
},
},
],
startStepId: 'fetch',
endStepIds: ['notify'],
transitions: [
{ from: 'fetch', to: 'clean' },
{ from: 'clean', to: 'risk_check' },
{ from: 'risk_check', to: 'manual_review',
condition: async (o) => (o as { highRisk: boolean }).highRisk },
{ from: 'risk_check', to: 'auto_refund',
condition: async (o) => !(o as { highRisk: boolean }).highRisk },
{ from: 'manual_review', to: 'process' },
{ from: 'process', to: 'notify' },
{ from: 'auto_refund', to: 'notify' },
],
}关键设计点：风控检查设了重试（外部 API 可能临时不可用）和超时（10 秒，防止拖慢流程）；高风险退款走人工审批，工作流在这里暂停；两条路径最终汇聚到通知步骤。
总结
回顾一下这篇的要点：
- 工作流 Agent 按预定义步骤执行，和 ReAct 循环的区别在于步骤转移是确定的，不是 LLM 现场推理
- 工作流适合步骤固定、需要审计、涉及人工介入的场景；自主 Agent 适合探索性强的场景；两者经常混用
- 四种基础模式——顺序执行、并行分叉/汇聚、条件分支、人工审批——可以组合出复杂工作流
- 工作流引擎核心是状态机：按转移规则找下一步，执行步骤，处理重试和超时
- 生产环境需要持久化每步状态到数据库，Worker 崩溃后能从快照恢复；有副作用的步骤要注意幂等性
- 错误处理按步骤特性选重试策略，关键步骤设超时，人工审批节点需要过期清理机制
- 退款处理是典型案例，综合了条件分支、人工审批、重试和超时
下一篇讨论工作流引擎的可视化，把代码定义的工作流渲染成流程图，方便调试和沟通。