为什么需要工作流编排

单个 Agent 调用一次工具、回答一个问题，看起来已经足够聪明。但一旦任务变成「帮我订一张下周去上海的机票，顺便把会议纪要整理成周报发给团队」，单步调用就不够了：你需要先查日历、再查航班、比价、下单、读纪要、生成文档、发邮件，中间任何一步失败都要决定是重试、换方案还是找人确认。这就是工作流编排要解决的问题——把一串不确定的步骤，组织成可观测、可重试、可中断的流水线。

工作流编排和让大模型自己「想想下一步做什么」不是一回事。前者是程序员预先画好节点图，模型只在每个节点里做决策；后者是模型自己决定下一步调什么工具。前者可控、可调试、可审计；后者灵活但容易跑偏。真实业务里，两条路通常混用：主干流程用编排固定下来，节点内部用模型自由发挥。

工作流编排的价值不是限制模型，而是给模型的自由装上边界：每一步都知道自己在哪、下一步去哪、失败了怎么办。
本文核心知识点
  • 工作流编排把多步任务拆成节点图，每个节点是一次模型调用或工具调用。
  • 状态传递节点之间通过结构化状态对象交换数据，而不是靠模型自由发挥。
  • 分支与循环让流程能根据上一步结果走不同路径，或重复执行直到满足条件。
  • 错误重试与人机协同保证自动化流程在遇到异常时不会直接崩掉，而是可控地恢复或转人工。
Agent 工作流配图：中央六边形核心连接日历、信封、齿轮与数据库四个工具节点，蓝色渐变扁平科技风格
一个典型的 Agent 工作流：中央编排核心把任务拆给日历、邮件、工具和数据库节点，节点之间通过状态对象传递结果。

节点、边与状态：工作流的三要素

任何工作流引擎，不管叫 LangGraph、Dify 还是 Airflow，本质上都在做三件事：定义节点、连接节点、在节点之间传状态。节点是一个可执行单元——可以是一次大模型调用、一次 API 请求、一段普通代码；边是节点之间的转移条件；状态是贯穿整个流程的共享数据结构。

要素作用典型实现
节点执行一个具体动作LLM 调用、工具函数、HTTP 请求、人工审批
边决定下一步去哪固定顺序、条件分支、循环回边、并行扇出
状态在节点间传递数据字典对象、消息列表、结构化 JSON

状态设计是工作流里最容易被低估的一环。新手喜欢让每个节点都把上下文全塞进 prompt 里，结果流程一长就开始丢字段、串数据。成熟的做法是显式定义状态结构：哪些字段由谁写入、谁读取、什么时机更新，让每个节点只关心自己负责的那几个字段。

# 一个最小的工作流状态与节点定义
state = {
    "user_query": "帮我订下周去上海的机票",
    "calendar_events": [],
    "flights": [],
    "selected_flight": None,
    "approval_required": False,
}

def query_calendar(state):
    state["calendar_events"] = calendar.search(next_week=True)
    return state

def search_flights(state):
    state["flights"] = flight_api.search(state["user_query"])
    return state

def choose_best(state):
    state["selected_flight"] = pick_cheapest(state["flights"])
    state["approval_required"] = state["selected_flight"]["price"] > 3000
    return state

这段伪代码里，三个节点各自只读写状态里自己关心的字段。后续加节点、改顺序，都不会影响其他节点的逻辑——这就是状态显式化带来的解耦。

分支、循环与并行执行

真实任务很少是一条直线。查完航班要判断价格是否超预算，超了就要找用户确认；搜索结果可能要反复过滤直到数量合适；多个独立查询可以同时发出去省时间。这些场景分别对应分支、循环和并行三种控制结构。

  • 条件分支：根据状态里某个字段的值，走不同的下游节点。例如审批标志为真时进人工节点，否则直接下单。
  • 循环回边：某个节点执行完不走向终点，而是回到前面的节点再跑一轮，直到满足退出条件。适合「反复改写直到格式正确」这类任务。
  • 并行扇出：一个节点同时触发多个分支，等全部完成后再汇聚。适合同时查多个数据源、同时跑多个分析任务。

设计并行时要特别注意合并策略：是等所有分支都成功，还是只要一个成功就继续？是把结果拼在一起，还是互相校验？这些决策直接决定了流程的韧性。

# 分支与并行的示意
graph = {
    "start": "query_calendar",
    "query_calendar": "search_flights",
    "search_flights": "choose_best",
    "choose_best": {
        True: "human_approval",       # 价格超预算，走人工
        False: "book_ticket",         # 直接下单
    },
    "human_approval": {
        "approved": "book_ticket",
        "rejected": "try_alternate",  # 换方案重来
    },
    "try_alternate": "search_flights",  # 循环回边
}

把分支条件显式画出来之后，整个流程的路径就一目了然了。调试时也能直接定位到「这个分支为什么走到了这里」，而不是在一大段 prompt 输出里猜。

错误处理与重试策略

Agent 工作流跑在真实世界里，外部 API 会超时、模型会幻觉、用户输入会莫名其妙。没有错误处理的工作流，第一次遇到异常就整条崩掉。工程上常用的手段有三层：节点内重试、节点间降级、全局兜底。

节点内重试针对的是瞬时错误：网络抖动、限流、偶发 500。设定一个最大重试次数和退避间隔，失败就等一会再试。节点间降级针对的是永久错误：这个工具就是调不通，那就换一个工具，或者用一个备选方案把流程继续走下去。全局兜底则是所有尝试都失败后，把当前状态存下来，通知人来接手。

# 带重试与降级的节点执行
def call_flight_api(state, retries=3):
    for attempt in range(retries):
        try:
            state["flights"] = flight_api.search(state["user_query"])
            return state
        except TimeoutError:
            sleep(2 ** attempt)   # 指数退避
    # 三次都失败，降级到备选数据源
    state["flights"] = backup_flight_source.search(state["user_query"])
    state["degraded"] = True
    return state

一个容易被忽略的细节是：重试必须是幂等的。如果第一次调用已经下了单，第二次重试又下了一单，那就不是容错而是事故。写节点之前先确认：这个操作重复执行多次，会不会产生副作用？

人机协同与人工审核节点

完全自动化的 Agent 在演示里很酷，但在真实业务里，涉及花钱、发消息、改数据的操作，几乎都需要人来点头。工作流编排的优势恰恰在这里：你可以在任何位置插一个「等待人工确认」节点，流程暂停在那里，等人点了按钮再继续。

人工节点不只是「等一下」。它要把当前状态里的关键信息整理成人能看懂的摘要，要给人提供明确的操作选项（批准、拒绝、修改后继续），还要设定一个超时时间——人忘了处理，流程不能永远挂着。

设计人工节点的三条经验
  • 信息摘要把当前节点的结论和将要执行的动作说清楚，不要让人对着原始 JSON 猜。
  • 操作有限给人 2-3 个明确选项，而不是开放输入框，降低决策成本。
  • 超时兜底设定最长等待时间，超时后自动走安全路径（比如取消而不是继续）。

人机协同的本质不是「让机器做人做的事」，而是把机器擅长的高频标准化动作和人擅长的判断性决策，在同一条流程里衔接起来。