单步 Tool Call 解决的是「做一个动作」，而真实世界的任务往往需要 十几个动作按特定顺序组合执行。这就需要**流程编排（Workflow Orchestration）**——告诉 Agent 先做什么、后做什么、什么条件下分支、失败了怎么回退。
📑 目录
- Workflow：从单步到多步的跨越
- 常见的编排模式
- DAG vs 线性流水线
- 任务调度与并行执行
- 错误处理与重试
- 主流编排框架
Workflow：从单步到多步的跨越
一句话定义
Workflow = 将复杂的业务流程定义为一系列有序的任务步骤，包含依赖关系、条件分支、并行执行和异常处理。让 Agent 从「一次做一件事」进化为「自动完成一整条工作流」。
本质对比
单步 Function Calling：
用户：发一封邮件通知团队开会
→ 调用 send_email(to="team@xxx.com", subject="会议通知", …)
→ 1 步完成
Workflow（多步编排）：
用户：帮我安排下周的产品评审会议
→ 这是一个需要 7 个步骤的工作流：
① 查看团队成员日历（calendar.freebusy）
② 找出所有人都有空的时间段
③ 创建会议邀请（calendar.create_event）
④ 准备上次会议的 action item（docs.search）
⑤ 生成会议议程草案（llm.generate）
⑥ 发送通知邮件（email.send）
⑦ 在项目管理系统中创建跟进任务（jira.create_ticket）
这 7 步有先后顺序、有条件判断、可能失败需要重试
→ 这就是 Workflow 要解决的问题
常见的编排模式
1. 线性链式（Sequential Chain）
Step 1 → Step 2 → Step 3 → Step 4 → 完成
适用：固定流程，每步输出是下一步输入
示例：文章写作流：大纲 → 正文 → 校对 → 发布
2. 条件分支（Conditional Branching）
判断条件
┌────┴────┐
Yes No
↓ ↓
Path A Path B
\\ /
└──┬──┘
↓
合并
适用：需要根据中间结果做决策的场景
示例：代码审查 → 有 Bug? → 是：创建 Issue / 否：合并 PR
3. 并行执行（Parallel Execution）
Start
┌───┼───┐
↓ ↓ ↓
Task1 Task2 Task3 （同时执行）
└───┼───┘
↓
Merge
适用：互不依赖的子任务
示例：同时搜索 GitHub + Stack Overflow + 文档库
4. 循环迭代（Loop / Iteration）
┌─────────┐
│ Task │←── 满足退出条件？否 → 重试
└────┬────┘ 是 ↓
│ Exit
└── 返回重新执行
适用：需要反复尝试直到成功的场景
示例：Rerank 检索直到结果相关性 > 0.9
5. DAG（有向无环图）
┌─── A ──┐
│ ↓
└──→ C ←┘
↑ │
│ ↓
D ←─── E ─┘
↑
└── F
适用：复杂依赖关系的场景
示例：数据处理管道：采集→清洗→转换→加载→分析→报告
任务调度与并行执行
import asyncio
from concurrent.futures import ThreadPoolExecutor
# 并行执行独立的工具调用
async def parallel_tool_execution(tasks):
"""并行执行无依赖关系的任务"""
async def run_task(task):
if task.type == "api_call":
return await api_client.call(task.endpoint, task.params)
elif task.type == "search":
return await search_engine.query(task.query)
elif task.type == "llm_gen":
return await llm.generate(task.prompt)
# 所有任务并发执行
results = await asyncio.gather(*[run_task(t) for t in tasks])
return dict(zip([t.name for t in tasks], results))
# 使用：原来串行 15 秒 → 并行 3 秒
tasks = [
Task(name="search_docs", query="最新 API 文档"),
Task(name="search_web", type="search", query="同类产品方案"),
Task(name="summarize", type="llm_gen", prompt="总结以下内容…"),
]
results = await parallel_tool_execution(tasks)
调度策略选择
| 步骤间有依赖 | 串行 | 后面需要前面的结果 |
| 步骤互相独立 | 并行 | 大幅降低总耗时 |
| 某步可能很慢 | 异步 + 超时不阻塞其他路径 | |
| 需要资源隔离 | 队列 + Worker | 防止一个慢任务拖死整个流 |
错误处理与重试
class ResilientStep:
def __init__(self, name, action, max_retries=3, fallback=None):
self.name = name
self.action = action
self.max_retries = max_retries
self.fallback = fallback # 兜底方案
async def execute(self, context):
for attempt in range(self.max_retries):
try:
result = await self.action(context)
if self._validate(result):
return {"status": "success", "data": result}
raise ValueError("Result validation failed")
except Exception as e:
if attempt < self.max_retries – 1:
await asyncio.sleep(1 ** attempt) # 指数退避
continue
# 全部重试失败 → 执行兜底
if self.fallback:
return {"status": "fallback", "data": await self.fallback(context)}
return {"status": "failed", "error": f"{self.name} failed after retries"}
# 使用
workflow = [
ResilientStep("search_db", db_search, fallback=vector_search),
ResilientStep("call_llm", llm_generate, max_retries=2),
ResilientStep("send_result", notify_user),
]
主流编排框架
| LangChain/LangGraph | 功能全，生态大，学习曲线陡 | 复杂 Agent 应用 |
| LlamaIndex Workflows | 数据处理友好，RAG 流水线强 | 知识密集型应用 |
| CrewAI | 多 Agent 协作原生支持 | 多角色协作场景 |
| AutoGen | 微软出品，多 Agent 对话式 | 研究和多 Agent 实验 |
| Dify | 低代码可视化编排 | 快速搭建不想写代码 |
| 自定义轻量 | asyncio + 状态机 | 简单场景，最大灵活性 |
❌ 常见误区
- ❌ 编排越复杂越好 — 能用 3 步解决的就别设计 8 步，每一步都增加出错概率
- ❌ 所有步骤都要 Agent 决策 — 固定步骤用硬编码，只在真正需要灵活性的地方引入 LLM
- ❌ 忽略错误恢复 — 生产环境的 Workflow 必须考虑每一步失败的兜底方案
