Workflow 编排：构建可靠的 AI 工作流

Agent 适合探索，Workflow 适合生产。当你需要可预测、可调试的 AI 系统时，Workflow 是更好的选择。

Workflow vs Agent

特性	Agent	Workflow
控制流	LLM 动态决定	代码预定义
可预测性	低	高
适用场景	探索性任务	重复性任务

何时使用 Workflow: 代码审查、文档生成、数据处理、测试生成等固定流程任务。

---

Workflow 基础模式

1. Prompt Chain (提示链)

多个 LLM 调用串联：

async function generateAPIEndpoint(spec) {
  // Step 1: 生成类型定义
  const types = await llm.chat({
    messages: [{ role: "user", content: `根据 API 规格生成 TypeScript 类型:\n${spec}` }]
  });

  // Step 2: 生成验证 Schema
  const validation = await llm.chat({
    messages: [{ role: "user", content: `根据类型生成 Zod Schema:\n${types}` }]
  });

  // Step 3: 生成 API Handler
  const handler = await llm.chat({
    messages: [{
      role: "user",
      content: `生成 Next.js API Route:\n类型: ${types}\n验证: ${validation}\n规格: ${spec}`
    }]
  });

  return { types, validation, handler };
}

2. 并行执行

独立任务并行处理：

async function reviewCode(files) {
  // 并行审查所有文件
  const reviews = await Promise.all(
    files.map(file => reviewSingleFile(file))
  );

  // 合并结果
  const summary = await llm.chat({
    messages: [{
      role: "user",
      content: `综合以下代码审查结果:\n${JSON.stringify(reviews)}`
    }]
  });

  return { reviews, summary };
}

3. 路由 (Routing)

根据输入选择不同路径：

async function processUserRequest(request) {
  // 分类请求
  const classification = await llm.chat({
    messages: [{
      role: "user",
      content: `分类请求类型:\n${request}\n\n返回: bug_report | feature_request | question`
    }]
  });

  // 路由到不同处理器
  switch (classification.trim()) {
    case 'bug_report':
      return await handleBugReport(request);
    case 'feature_request':
      return await handleFeatureRequest(request);
    case 'question':
      return await handleQuestion(request);
  }
}

4. 编排 (Orchestration)

复杂的多步骤流程：

async function deployFeature(featureSpec) {
  // 1. 生成代码
  const code = await generateCode(featureSpec);

  // 2. 生成测试
  const tests = await generateTests(code);

  // 3. 运行测试
  const testResults = await runTests(tests);

  if (!testResults.passed) {
    // 4a. 修复失败的测试
    const fixedCode = await fixCode(code, testResults.failures);
    return await deployFeature({ ...featureSpec, code: fixedCode });
  }

  // 4b. 生成文档
  const docs = await generateDocs(code);

  // 5. 部署
  await deploy(code, tests, docs);

  return { code, tests, docs };
}

---

实战案例

案例 1: 代码审查 Workflow

async function codeReviewWorkflow(pullRequest) {
  // 1. 获取变更文件
  const files = await getChangedFiles(pullRequest);

  // 2. 并行审查
  const reviews = await Promise.all(
    files.map(async (file) => {
      const content = await readFile(file.path);

      return await llm.chat({
        messages: [{
          role: "system",
          content: "你是代码审查专家。检查安全、性能、可维护性问题。"
        }, {
          role: "user",
          content: `审查代码:\n\`\`\`${file.language}\n${content}\n\`\`\``
        }]
      });
    })
  );

  // 3. 生成总结
  const summary = await llm.chat({
    messages: [{
      role: "user",
      content: `综合审查结果:\n${reviews.join('\n\n')}`
    }]
  });

  // 4. 发布评论
  await postComment(pullRequest, summary);

  return { reviews, summary };
}

案例 2: 文档生成 Workflow

async function generateDocumentation(codebase) {
  // 1. 提取 API
  const apis = await extractAPIs(codebase);

  // 2. 并行生成文档
  const docs = await Promise.all(
    apis.map(async (api) => {
      return await llm.chat({
        messages: [{
          role: "user",
          content: `为以下 API 生成文档:\n${api.signature}\n\n包括: 描述、参数、返回值、示例`
        }]
      });
    })
  );

  // 3. 生成目录
  const toc = await llm.chat({
    messages: [{
      role: "user",
      content: `生成文档目录:\n${docs.map(d => d.title).join('\n')}`
    }]
  });

  // 4. 组装文档
  return { toc, docs };
}

案例 3: 测试生成 Workflow

async function generateTests(sourceCode) {
  // 1. 分析代码
  const analysis = await llm.chat({
    messages: [{
      role: "user",
      content: `分析代码并列出需要测试的场景:\n${sourceCode}`
    }]
  });

  // 2. 生成测试用例
  const testCases = await llm.chat({
    messages: [{
      role: "user",
      content: `根据场景生成测试用例:\n${analysis}`
    }]
  });

  // 3. 生成测试代码
  const testCode = await llm.chat({
    messages: [{
      role: "user",
      content: `生成 Jest 测试代码:\n源码: ${sourceCode}\n用例: ${testCases}`
    }]
  });

  // 4. 验证测试
  const validation = await runTests(testCode);

  if (!validation.passed) {
    return await generateTests(sourceCode); // 重试
  }

  return testCode;
}

---

错误处理

1. 重试机制

async function withRetry(fn, maxRetries = 3) {
  for (let i = 0; i < maxRetries; i++) {
    try {
      return await fn();
    } catch (error) {
      if (i === maxRetries - 1) throw error;
      await sleep(1000 * Math.pow(2, i)); // 指数退避
    }
  }
}

// 使用
const result = await withRetry(() => llm.chat({ messages }));

2. Fallback 策略

async function generateWithFallback(prompt) {
  try {
    // 尝试主模型
    return await llm.chat({ model: 'gpt-4o', messages: [{ role: 'user', content: prompt }] });
  } catch (error) {
    // 降级到备用模型
    return await llm.chat({ model: 'gpt-4o-mini', messages: [{ role: 'user', content: prompt }] });
  }
}

3. 补偿事务

async function deployWithRollback(code) {
  const backup = await createBackup();

  try {
    await deploy(code);
    await runSmokeTests();
  } catch (error) {
    // 回滚
    await restore(backup);
    throw error;
  }
}

---

人机协作

1. 人工审批

async function deployWorkflow(code) {
  // 1. 生成部署计划
  const plan = await generateDeploymentPlan(code);

  // 2. 请求人工审批
  const approved = await requestApproval(plan);

  if (!approved) {
    throw new Error('Deployment rejected');
  }

  // 3. 执行部署
  await deploy(code);
}

2. 人工修正

async function generateWithReview(prompt) {
  // 1. 生成初稿
  const draft = await llm.chat({ messages: [{ role: 'user', content: prompt }] });

  // 2. 人工审查
  const feedback = await requestHumanReview(draft);

  if (feedback.needsRevision) {
    // 3. 根据反馈修改
    return await llm.chat({
      messages: [
        { role: 'user', content: prompt },
        { role: 'assistant', content: draft },
        { role: 'user', content: `修改建议: ${feedback.comments}` }
      ]
    });
  }

  return draft;
}

---

可观测性

1. 追踪执行

class WorkflowTracer {
  constructor() {
    this.steps = [];
  }

  async trace(stepName, fn) {
    const start = Date.now();

    try {
      const result = await fn();
      this.steps.push({
        name: stepName,
        status: 'success',
        duration: Date.now() - start,
        result
      });
      return result;
    } catch (error) {
      this.steps.push({
        name: stepName,
        status: 'error',
        duration: Date.now() - start,
        error: error.message
      });
      throw error;
    }
  }

  getReport() {
    return {
      totalSteps: this.steps.length,
      successRate: this.steps.filter(s => s.status === 'success').length / this.steps.length,
      totalDuration: this.steps.reduce((sum, s) => sum + s.duration, 0),
      steps: this.steps
    };
  }
}

// 使用
const tracer = new WorkflowTracer();
await tracer.trace('generate_code', () => generateCode(spec));
await tracer.trace('generate_tests', () => generateTests(code));
console.log(tracer.getReport());

2. 成本监控

class CostTracker {
  constructor() {
    this.costs = [];
  }

  track(model, inputTokens, outputTokens) {
    const cost = calculateCost(model, inputTokens, outputTokens);
    this.costs.push({ model, inputTokens, outputTokens, cost });
  }

  getTotalCost() {
    return this.costs.reduce((sum, c) => sum + c.cost, 0);
  }
}

---

关键要点

1. Workflow 比 Agent 更可控: 适合生产环境的重复性任务
2. 选择合适的模式: Chain、Parallel、Router、Orchestrator
3. 人机协作很重要: 关键步骤需要人工确认
4. 健壮的错误处理: 重试、Fallback、补偿
5. 可观测性: 追踪每步执行，监控成本
6. 增量优化: 从简单 Chain 开始，按需添加复杂性

---