AI 工作流编排实战：从单步调用到多 Agent 协同的工程化方案
一、单步调用的天花板：为什么需要工作流编排
当 AI 能力从"单次问答"走向"多步骤任务"时，单步 LLM 调用的局限性暴露无遗。一个典型的内容生产场景：用户输入一个主题，系统需要依次完成"调研搜索 -> 大纲生成 -> 内容撰写 -> 质量审核 -> 格式排版"五个步骤。每个步骤的输入依赖上一步的输出，且不同步骤可能需要不同的模型配置和 Prompt 策略。
如果将这些步骤硬编码在一个函数中，代码会迅速膨胀为难以维护的"面条逻辑"。更关键的是，中间步骤失败时缺乏重试机制，步骤间的状态传递缺乏类型安全，整个流程缺乏可观测性。AI 工作流编排的核心目标，就是将多步骤 AI 任务从"过程式代码"转化为"声明式流程"，实现可复用、可观测、可恢复的任务执行。
二、工作流编排架构：DAG 驱动的任务调度引擎
工作流编排的本质是将任务抽象为有向无环图（DAG），节点代表执行步骤，边代表数据依赖关系。调度引擎按照拓扑排序执行节点，支持并行执行无依赖关系的步骤。
flowchart TD
A[用户输入主题] –> B[调研搜索节点]
B –> B1[Web搜索]
B –> B2[知识库检索]
B1 –> C[信息整合节点]
B2 –> C
C –> D[大纲生成节点]
D –> E{大纲审核}
E –>|通过| F[内容撰写节点]
E –>|不通过| D
F –> G[质量审核节点]
G –> H{质量评分}
H –>|>=80分| I[格式排版节点]
H –>|<80分| F
I –> J[最终输出]
style E fill:#f9f,stroke:#333
style H fill:#f9f,stroke:#333
上图展示了内容生产工作流的完整 DAG 结构。其中，调研搜索节点的两个子任务（Web 搜索和知识库检索）可以并行执行，审核节点包含条件分支逻辑。以下是核心调度引擎的实现：
// 工作流节点定义
interface WorkflowNode<TInput, TOutput> {
id: string;
execute: (input: TInput, context: WorkflowContext) => Promise<TOutput>;
retryConfig?: { maxRetries: number; backoffMs: number };
timeout?: number;
}
// 工作流上下文：在节点间传递状态
interface WorkflowContext {
getOutput: (nodeId: string) => unknown;
setOutput: (nodeId: string, output: unknown) => void;
logger: WorkflowLogger;
}
// 工作流编排器
class WorkflowOrchestrator {
private nodes: Map<string, WorkflowNode<any, any>> = new Map();
private edges: Map<string, string[]> = new Map(); // nodeId -> 依赖的节点ID列表
addNode<TInput, TOutput>(node: WorkflowNode<TInput, TOutput>): this {
this.nodes.set(node.id, node);
return this;
}
addEdge(from: string, to: string): this {
// to 依赖 from 的输出
const deps = this.edges.get(to) || [];
deps.push(from);
this.edges.set(to, deps);
return this;
}
async run(initialInput: unknown): Promise<Map<string, unknown>> {
const context = this.createContext();
const executionOrder = this.topologicalSort();
for (const nodeId of executionOrder) {
const node = this.nodes.get(nodeId)!;
const deps = this.edges.get(nodeId) || [];
// 收集依赖节点的输出作为当前节点的输入
const input = deps.length === 0
? initialInput
: deps.length === 1
? context.getOutput(deps[0])
: Object.fromEntries(
deps.map(depId => [depId, context.getOutput(depId)])
);
// 带重试和超时的执行
const output = await this.executeWithRetry(node, input, context);
context.setOutput(nodeId, output);
}
return context.getAllOutputs();
}
private async executeWithRetry(
node: WorkflowNode<any, any>,
input: any,
context: WorkflowContext
): Promise<any> {
const { maxRetries = 2, backoffMs = 1000 } = node.retryConfig || {};
let lastError: Error | null = null;
for (let attempt = 0; attempt <= maxRetries; attempt++) {
try {
// 超时控制：防止LLM调用无限挂起
const result = await Promise.race([
node.execute(input, context),
this.createTimeout(node.timeout || 60000),
]);
context.logger.log(node.id, 'success', { attempt });
return result;
} catch (error) {
lastError = error as Error;
context.logger.log(node.id, 'retry', { attempt, error: lastError.message });
if (attempt < maxRetries) {
await this.sleep(backoffMs * Math.pow(2, attempt));
}
}
}
throw new Error(`节点 ${node.id} 执行失败，已重试 ${maxRetries} 次: ${lastError!.message}`);
}
// 拓扑排序：确定节点的执行顺序
private topologicalSort(): string[] {
const visited = new Set<string>();
const result: string[] = [];
const visit = (nodeId: string) => {
if (visited.has(nodeId)) return;
visited.add(nodeId);
const deps = this.edges.get(nodeId) || [];
deps.forEach(visit);
result.push(nodeId);
};
this.nodes.forEach((_, id) => visit(id));
return result;
}
}
三、多 Agent 协同模式：从流水线到协作网络
当工作流中的节点本身也需要"思考"和"决策"时，就需要引入 Agent 模式。与固定流程的工作流不同，Agent 可以根据中间结果动态选择下一步行动。
// Agent定义：具备工具调用能力的智能节点
interface Agent {
role: string;
systemPrompt: string;
tools: Tool[];
maxIterations: number;
}
interface Tool {
name: string;
description: string;
execute: (params: Record<string, unknown>) => Promise<unknown>;
}
// 多Agent协同调度器
class MultiAgentCoordinator {
private agents: Map<string, Agent> = new Map();
// 编排模式1：顺序传递（Pipeline）
async runPipeline(
input: string,
agentIds: string[]
): Promise<string> {
let currentInput = input;
for (const agentId of agentIds) {
const agent = this.agents.get(agentId)!;
currentInput = await this.runAgent(agent, currentInput);
}
return currentInput;
}
// 编排模式2：评审循环（Review Loop）
async runWithReview(
input: string,
writerAgent: Agent,
reviewerAgent: Agent,
maxRounds: number = 3
): Promise<string> {
let content = await this.runAgent(writerAgent, input);
for (let round = 0; round < maxRounds; round++) {
// 评审Agent给出修改意见
const review = await this.runAgent(
reviewerAgent,
`请评审以下内容并给出具体修改建议：\\n${content}`
);
// 解析评审结果，判断是否需要修改
const reviewResult = JSON.parse(review);
if (reviewResult.score >= 80) {
return content; // 质量达标，结束循环
}
// 写作Agent根据修改意见重写
content = await this.runAgent(
writerAgent,
`请根据以下修改建议重写内容：\\n原始内容：${content}\\n修改建议：${reviewResult.suggestions}`
);
}
return content;
}
private async runAgent(agent: Agent, input: string): Promise<string> {
const messages: ChatMessage[] = [
{ role: 'system', content: agent.systemPrompt },
{ role: 'user', content: input },
];
for (let i = 0; i < agent.maxIterations; i++) {
const response = await callLLM(messages, {
tools: agent.tools.map(t => ({
name: t.name,
description: t.description,
})),
});
// 如果模型选择调用工具，执行工具并将结果返回模型
if (response.toolCalls) {
for (const toolCall of response.toolCalls) {
const tool = agent.tools.find(t => t.name === toolCall.name)!;
const result = await tool.execute(toolCall.arguments);
messages.push({
role: 'tool',
content: JSON.stringify(result),
});
}
continue; // 让模型继续思考
}
// 模型返回最终文本结果
return response.content;
}
throw new Error(`Agent ${agent.role} 超过最大迭代次数`);
}
}
四、工作流编排的现实约束与适用边界
工作流编排并非适用于所有 AI 场景，过度设计反而增加复杂度。
延迟叠加问题：多步骤工作流的总延迟是各步骤延迟之和。一个5步工作流，每步平均2秒，总延迟至少10秒。对于实时交互场景，这种延迟不可接受。优化策略包括：并行执行无依赖步骤、使用流式传递中间结果、对非关键步骤异步化处理。
错误传播风险：工作流中任何一步失败都会导致后续步骤无法执行。虽然重试机制可以缓解瞬时故障，但如果某步骤的输出质量本身有问题（如 LLM 幻觉），后续步骤会在错误基础上继续推理，产生"错误放大"效应。解决方案是在关键节点增加校验逻辑，对 LLM 输出做结构化校验。
成本倍增效应：每增加一个工作流节点，就意味着一次 LLM 调用。一个5步工作流的 Token 消耗是单次调用的5倍以上。对于高频调用的生产场景，需要评估工作流的 ROI，避免"为了编排而编排"。
适用场景建议：工作流编排最适合"步骤固定、依赖明确、质量要求高"的任务，如内容生产、数据处理、报告生成。对于"步骤动态、需要实时响应"的场景（如对话式 AI），应优先使用单次调用 + 工具使用模式，而非复杂的工作流编排。
五、总结
AI 工作流编排将多步骤任务从过程式代码转化为声明式 DAG，实现了可复用、可观测、可恢复的任务执行。DAG 驱动的调度引擎支持拓扑排序执行、并行调度和条件分支，多 Agent 协同模式则在此基础上引入了动态决策能力。但工作流编排存在延迟叠加、错误传播和成本倍增的现实约束，适用于步骤固定且质量要求高的场景，而非所有 AI 调用都需要编排化。工程化的核心是"按需设计"，用最简单的方案解决最核心的问题。
