[!info] related notes

• 所属 MOC: AI Agent Application MOC, Agent MOC
• 相关概念: Agentic Workflow Patterns, 状态机, 图工作流
• 易混淆: Agent Orchestration
• 框架: LangGraph, CrewAI

工作流编排

一句话定义

工作流编排是定义、调度和管理多步骤任务执行的工程实践。在 AI Agent 场景中，它决定了多个 LLM 调用、工具执行和人类判断点如何组织成一个完整的任务流程。

它解决什么问题

单次 LLM 调用只能完成一个原子任务。但真实场景需要：

• 串行依赖: 先搜索，再总结，再翻译
• 并行处理: 同时搜索多个数据源
• 条件分支: 如果检索结果够用就直接回答，不够就换个关键词再搜
• 循环迭代: 生成 → 评估 → 修改 → 再评估
• 错误恢复: 某个步骤失败时的降级策略
• 人类介入: 关键节点暂停等待审批

工作流编排把这些模式组织成可执行、可监控、可恢复的流程。

核心原理

编排模式分类

模式	描述	适用场景
顺序 (Sequential)	A → B → C	有依赖关系的步骤
并行 (Parallel)	A + B → C	独立的子任务
条件 (Conditional)	if X → A, else → B	路由和分支
循环 (Loop)	A → B → 检查 → A 或结束	迭代优化
子工作流 (Sub-workflow)	A → [子流程] → C	复用和模块化

Anthropic 的五种 Workflow 模式

Anthropic 在 “Building Effective Agents” 中定义了五种递进的 Workflow 模式：

1. Prompt Chaining:  LLM₁ → LLM₂ → LLM₃
   串行分解，每步输出是下一步输入

2. Routing:          LLM(Classify) → HandlerA / HandlerB / HandlerC
   先分类再分流到专门处理器

3. Parallelization:  LLM₁ + LLM₂ + LLM₃ → Aggregator
   并行执行再汇总

4. Orchestrator-Workers:  LLM(Orchestrator) → Worker₁ + Worker₂ → Synthesis
   动态拆解子任务再汇总

5. Evaluator-Optimizer:  LLM(Generator) → LLM(Evaluator) → Loop
   生成与评估的闭环

Workflow vs Agent

	Workflow	Agent
控制流	预定义的代码路径	模型动态决定
可预测性	高	低
灵活性	低	高
适用场景	流程明确的任务	开放式探索任务

大多数生产系统是 Workflow 骨架 + Agent 节点 的组合。

在 React + Go + Python AI Service 架构中的位置

工作流编排主要在 Python AI Service 层实现：

• LangGraph / CrewAI 等框架提供编排能力
• Go 后端负责调用编排入口和转发结果
• 前端负责展示流程进度和审批界面

典型工程实现

LangGraph 示例

from langgraph.graph import StateGraph, END

# 定义状态
class AgentState(TypedDict):
    query: str
    search_results: list
    analysis: str
    report: str

# 构建工作流
graph = StateGraph(AgentState)

# 添加节点
graph.add_node("search", search_node)
graph.add_node("analyze", analyze_node)
graph.add_node("report", report_node)

# 定义边
graph.set_entry_point("search")
graph.add_edge("search", "analyze")
graph.add_conditional_edges(
    "analyze",
    should_expand_search,  # 条件函数
    {
        "expand": "search",  # 回到搜索
        "sufficient": "report",  # 进入报告
    }
)
graph.add_edge("report", END)

# 编译
app = graph.compile()

条件路由函数

def should_expand_search(state: AgentState) -> str:
    """判断搜索结果是否足够"""
    if len(state["search_results"]) < 3:
        return "expand"
    if state["analysis"].confidence < 0.7:
        return "expand"
    return "sufficient"

常见设计模式

1. Pipeline 模式

最简单的串行流水线：A → B → C。每步的输出是下一步的输入。

2. Router 模式

先分类再路由：根据输入类型选择不同的处理路径。

3. Fan-out/Fan-in 模式

并行执行多个子任务，汇总结果后继续。

4. Retry with Backoff

失败后重试，使用指数退避策略。

5. Circuit Breaker

连续失败达到阈值后熔断，避免级联故障。

常见坑

1. 过度编排: 简单任务不需要复杂工作流，直接单次 LLM 调用就够
2. 不做状态持久化: 长时间运行的工作流中断后无法恢复
3. 不做超时控制: 某个节点卡住导致整个工作流挂起
4. 错误处理缺失: 某个步骤失败后没有降级路径
5. 并行度不做限制: 同时启动太多并行任务导致资源耗尽

和其他概念的关系

• vs Agentic Workflow Patterns: Patterns 是编排的模式库
• vs 状态机: 状态机是编排的一种实现方式
• vs 图工作流: 图工作流是编排的底层模型
• vs Agent Engine: Engine 是编排中每个节点的执行者
• vs Agent Orchestration: 两者有交集但不等同；本页强调显式多步骤控制流，后者强调 agent / tool / handoff / context / policy 的系统级调度

学习检查

不看正文回答：

1. 工作流编排为什么不仅是“按顺序调用几个函数”？
2. Sequential、Parallel、Conditional 与 Loop 分别改变了什么控制流语义？
3. Workflow orchestration 与 Agent Orchestration 的核心边界是什么？
4. 为什么生产系统常采用“确定性 workflow skeleton + 动态 agent 节点”？
5. 哪些状态必须持久化，才能让长流程在失败后安全恢复？

总结

工作流编排的本质是把复杂的多步任务拆解成可管理、可监控、可恢复的流程。好的编排设计应该在确定性和灵活性之间找到平衡。

参考资料

• Building Effective Agents - Anthropic
• LangGraph Concepts
• Agentic Workflow Patterns