06 Agent 工作流编排

学习目标：能设计多步骤 Agent 工作流，支持任务拆解、分支、工具调用、人工确认和失败恢复。 重点度：必会（15 分）—— 全路线最高分之一 前置要求：Ch5 Agent Runtime

概述#

复杂任务不能靠单轮问答，需要拆解、分支、循环、人工确认和失败恢复——这就是工作流编排。

⚠️ 重要认知：多 Agent 不一定优于单 Agent。多 Agent 的通信成本、责任不清、上下文丢失、eval 困难都是真实代价。能用 workflow 解决的，别盲目上 multi-agent。

ReAct / Plan-and-Execute#

ReAct（Reason + Act）#

最简单的 Agent 模式：思考一步 → 执行一步 → 再思考 → 再执行。

text
Thought: 用户要分析 repo, 我先看 README
Action: read_file("README.md")
Observation: 这是一个 Node.js 项目...
Thought: 接下来看 package.json
Action: read_file("package.json")
Observation: ...
Thought: 已经收集足够信息, 可以总结了
Action: final_answer("...")

优势：简单、灵活、对开放任务友好。劣势：每一步都要调 LLM、慢、可能跑偏。

Plan-and-Execute#

先制定完整计划，再分步执行：

text
Plan:
  1. 读 README
  2. 读 package.json
  3. 分析 src 结构
  4. 总结报告

Execute:
  Step 1: read_file("README.md") → ...
  Step 2: read_file("package.json") → ...
  Step 3: list_dir("src") → ...
  Step 4: generate_summary(...) → ...

优势：可预测、可审计、可并行规划。劣势：计划可能跑偏，需要 re-plan 机制。

混合策略#

text
Plan-and-Execute 制定大方向
  ↓
执行时如遇意外, ReAct 灵活调整
  ↓
每 N 步重新评估 plan

Workflow / State Machine / Graph Orchestration#

Workflow（工作流）#

固定步骤序列：

text

[接收订单] → [校验库存] → [扣款] → [生成发货单] → [通知仓库]

适合流程明确的场景。优势是可预测、可审计；劣势是不灵活。

State Machine（状态机）#

带状态流转：

text
状态: draft → submitted → reviewing → approved → executed
                                ↓
                              rejected → draft

每个状态有:
  - 可执行动作
  - 自动流转条件
  - 进入/退出 hook

适合有明确状态流转的业务场景（审批流、订单流）。

Graph Orchestration（图编排）#

最灵活，节点 + 边 + 条件：

text
       ┌─────────┐
       │ analyze │
       └────┬────┘
            │
      ┌─────┴─────┐
      ↓           ↓
  ┌───────┐  ┌───────┐
  │ search│  │ query │
  └───┬───┘  └───┬───┘
      │          │
      └────┬─────┘
           ↓
      ┌─────────┐
      │ synthesize │
      └─────────┘

适合复杂分支并行场景。LangGraph、Dify Workflow、Coze Workflow 都是图编排引擎。

LangGraph / Dify / Coze Workflow#

LangGraph#

Python 库，把 Agent 工作流建模成图：

python
from langgraph.graph import StateGraph, END

def search_node(state):
    state["search_results"] = search(state["query"])
    return state

def answer_node(state):
    state["answer"] = llm.generate(state["search_results"])
    return state

graph = StateGraph(dict)
graph.add_node("search", search_node)
graph.add_node("answer", answer_node)
graph.add_edge("search", "answer")
graph.add_edge("answer", END)
graph.set_entry_point("search")

app = graph.compile()
result = app.invoke({"query": "什么是 Agent"})

特点：代码定义、可复杂分支、支持 checkpoint（断点恢复）。

Dify Workflow#

可视化编排，拖拽节点：

LLM 节点 / 工具节点 / 条件分支 / 循环 / 代码节点
可视化调试、版本管理、API 发布
适合非工程师搭建 Agent 工作流

Coze Workflow#

字节出品，类似 Dify：

国内合规友好
内置大量插件（搜索/数据库/IM）
Bot 发布到飞书/抖音/微信等渠道

选型决策#

需求	推荐
工程师写代码、复杂逻辑	LangGraph
非工程师搭建、可视化	Dify / Coze
国内业务、多渠道发布	Coze
海外业务、开源	Dify

Supervisor / Router / Worker Agent#

Supervisor 模式#

一个 Supervisor Agent 协调多个 Worker：

text
       ┌───────────┐
       │ Supervisor│  ← 拆任务、分派、汇总
       └─────┬─────┘
    ┌────────┼────────┐
    ↓        ↓        ↓
┌──────┐ ┌──────┐ ┌──────┐
│Worker│ │Worker│ │Worker│  ← 各干各的
│  1   │ │  2   │ │  3   │
└──────┘ └──────┘ └──────┘

优势：职责清晰、可并行、context 隔离。劣势：Supervisor 是单点、协调成本高。

Router 模式#

Router 决定请求分给哪个 Worker：

text
用户请求
  ↓
Router (轻量模型/规则): 判断属于哪个领域
  ↓
Worker A (财务专家) / Worker B (技术专家) / Worker C (HR)

优势：每个 Worker 专注一个领域、prompt 精简。劣势：Router 判断错就完蛋。

Worker 模式#

无协调者，多个 Worker 平等协作：

text

Worker 1 ↔ Worker 2 ↔ Worker 3
（通过 A2A 协议通信）

优势：去中心化、灵活。劣势：协作复杂、debug 困难。

Multi-Agent Collaboration（★A2A 依赖）#

多 Agent 何时有意义#

并行独立子任务：3 个文件独立分析 → 3 个 Worker 并行
专业分工：编码 + 审计 + 测试，各 Agent 专精
context 隔离：子任务上下文大，独立 Agent 跑不污染主上下文

多 Agent 何时不该用#

⚠️ 反模式：为了多 Agent 而多 Agent

简单串行任务 → 单 Agent + workflow
需要全局视角的决策 → 单 Agent 更好（多 Agent 视角割裂）
实时性要求高 → 多 Agent 通信延迟大

多 Agent 协作的真实代价#

代价	说明
通信成本	Agent 间消息要序列化、传输、解析，token 翻倍
责任不清	失败时不知道是哪个 Agent 错
上下文丢失	Agent 间共享状态难，信息衰减
eval 困难	多 Agent 轨迹复杂，评测爆炸
调试噩梦	出 bug 要追踪多个 Agent 的消息流

判断口诀：能用 workflow 别用 multi-agent；要用 multi-agent 必须并行独立子任务。

Human-in-the-Loop（HITL）#

为什么需要 HITL#

Agent 不是全自动就好——某些场景必须人工介入：

不可逆操作：删除、转账、部署、发邮件
高成本操作：一次操作花几百美元
合规要求：医疗/金融/法律需要人工签字
创意判断：审美、文案、设计选择
不确定性高：Agent 自己也不确定怎么办

HITL 模式#

1. Approval Gate（审批门）#

text
Agent: 我准备删除 user_id=123, 请确认
  ↓
用户: [批准] [拒绝] [查看详情]
  ↓
Agent: 收到批准/拒绝, 继续/调整

2. Interactive Q&A#

text
Agent: 我需要确认任务优先级, 高优先级能立即处理, 低优先级排队
  ↓
用户: 高优先级
  ↓
Agent: 已调整为高优先级, 开始处理

3. Plan Review#

text

Agent 生成计划 → 用户 review → 用户修改/批准 → Agent 执行

4. Result Review#

text

Agent 完成任务 → 用户 review 结果 → 用户接受/要求修改

HITL 体验设计#

不打断节奏：低风险自动通过，高风险才打断
可批量：重复操作支持批量审批
可异步：用户不在线时任务挂起，用户回来再继续
可解释：每次审批给足上下文，让用户能做判断

长任务恢复（Checkpoint / Resume）#

为什么需要 Checkpoint#

Agent 跑长任务时可能中断：

用户主动暂停
系统崩溃
预算耗尽
等人工审批

没有 Checkpoint，就要从头跑；有 Checkpoint，能从断点继续。

Checkpoint 设计#

python
def save_checkpoint(state):
    checkpoint = {
        "task_id": state.task_id,
        "current_step": state.step,
        "messages": state.messages,  # 或压缩后的
        "workspace_state": state.workspace,
        "plan": state.plan,
        "completed_steps": state.completed,
        "budget_used": state.budget_used,
    }
    db.save(checkpoint)

def resume(task_id):
    checkpoint = db.load(task_id)
    state = restore(checkpoint)
    return agent.continue_from(state)

Checkpoint 时机#

每完成 N 步（定期）
每个关键节点后（如 plan 完成、子任务完成）
等待人工审批前
预算预警时

LangGraph 的 Checkpoint#

LangGraph 内置 checkpoint 机制：

python
from langgraph.checkpoint.memory import MemorySaver

graph = build_graph()
app = graph.compile(checkpointer=MemorySaver())

# 第一次执行
config = {"configurable": {"thread_id": "task-1"}}
app.invoke({"query": "..."}, config)

# 中断后恢复
app.invoke(None, config)  # 从最近 checkpoint 继续

工作流设计模式总结#

模式选择决策树#

text
任务简单?
  └─ Yes → ReAct 单 Agent
任务可预测?
  └─ Yes → Workflow (固定步骤)
有状态流转?
  └─ Yes → State Machine
有复杂分支?
  └─ Yes → Graph (LangGraph)
有并行独立子任务?
  └─ Yes → Supervisor + Worker (multi-agent)
需要专业分工?
  └─ Yes → Router + Worker
跨系统 Agent 协作?
  └─ Yes → A2A 协议

长任务恢复策略#

关键节点存 Checkpoint
任务状态外置到 workspace（不依赖内存）
支持 Resume（从 Checkpoint 继续）
支持 Retry（重新跑某一步）
支持 Rollback（回退到某个 Checkpoint）

阶段产出#

完成本章后应该能：

区分 ReAct / Plan-and-Execute，知道何时用哪个
设计 Workflow / State Machine / Graph 三种编排
用 LangGraph 写图编排工作流（含 Checkpoint）
评估 Dify / Coze 选型，搭建可视化工作流
设计 Supervisor / Router / Worker 三种 multi-agent 模式
判断"何时该用 multi-agent、何时不该"
实现 4 种 HITL 模式（Approval / Q&A / Plan Review / Result Review）
实现 Checkpoint / Resume / Retry / Rollback

下一章 Ch7 进入 Workspace / Memory / State——Agent 执行复杂任务时不能只依赖上下文窗口，怎么管理会话状态、工作区文件、短期/长期记忆。