工作流编排：Agent的神经系统
单个Agent的能力有限。复杂任务需要多个Agent协作——有人负责搜索，有人负责推理，有人负责验证。工作流编排就是定义这些Agent如何协作的"神经系统"。
编排模式演进
模式一：静态DAG（有向无环图）
最简单的编排方式——预定义执行流程：
[收集信息] → [分析数据] → [生成报告] → [审核质量]
优点：
- 确定性强，易于调试
- 执行可预测
- 性能好（可并行优化不依赖的节点）
局限：
- 无法处理动态决策（发现需要额外搜索时无法回溯）
- 流程僵化，不适配多变场景
模式二：条件分支DAG
在DAG基础上加入条件判断：
[收集信息] → [信息充分?]
├─ 是 → [分析数据]
└─ 否 → [补充搜索] → [分析数据]
优点：
- 支持基本决策逻辑
- 仍保持可预测性
局限：
- 分支数有限时有效
- 复杂决策导致图过于复杂
模式三：状态机
状态机允许循环和回退：
状态: {current_step, context, results}
转移规则:
collecting → (if enough info) → analyzing
collecting → (if not enough) → collecting
analyzing → (if clear) → generating
analyzing → (if need more data) → collecting
generating → (if quality ok) → done
generating → (if quality low) → analyzing
优点：
- 支持循环和重试
- 状态可持久化
- 更接近真实工作流
局限：
- 状态爆炸风险
- 仍需预定义转移规则
模式四：自适应流程
让LLM决定下一步做什么：
循环:
当前状态 = {任务, 已完成步骤, 可用工具, 结果}
下一步 = LLM(当前状态, 可选步骤列表)
执行下一步
更新状态
如果完成 → 退出
优点：
- 最大灵活性
- 可处理未预料的情况
局限：
- 不可预测
- 难以调试
- 可能进入无限循环
- Token消耗大
实践架构
混合编排：分层设计
实践中最有效的是混合架构——高层用DAG保证可控性，低层用自适应流程处理细节：
[Phase 1: 信息收集] (DAG, 可并行)
├── Agent A: 搜索网页
├── Agent B: 查询数据库
└── Agent C: 读取文件
[Phase 2: 分析推理] (自适应)
└── LLM Agent:
- 根据收集的信息自主分析
- 如果发现信息缺口，可回退到Phase 1
- 多轮推理直到得出结论
[Phase 3: 结果验证] (DAG)
├── 验证Agent: 检查事实性
└── 格式Agent: 确保格式正确
[Phase 4: 输出] (固定)
└── 生成最终报告
实现方案
class Workflow:
def __init__(self):
self.phases = []
self.state = {}
def add_phase(self, name, executor, condition=None):
self.phases.append({
'name': name,
'executor': executor,
'condition': condition
})
async def run(self, input_data):
self.state = {'input': input_data, 'results': {}}
for i, phase in enumerate(self.phases):
# 检查条件
if phase['condition'] and not phase['condition'](self.state):
continue
# 执行
result = await phase['executor'](self.state)
self.state['results'][phase['name']] = result
# 自适应：允许LLM决定是否回退
if result.get('retry'):
# 回退到指定phase
i = self.find_phase(result['retry_to']) - 1
return self.state
并发控制
Agent并发
多个Agent并行执行可以大幅提升速度：
# 并行执行独立任务
tasks = [
search_agent.search("AI芯片市场"),
data_agent.query("chip_revenue"),
doc_agent.read("market_report.pdf")
]
results = await asyncio.gather(*tasks)
资源限制
并发需要控制资源使用：
- LLM API并发：避免触发rate limit
- 工具调用并发：外部API有调用频率限制
- 内存：每个Agent的上下文占内存
DAG到执行图
将DAG转换为可执行的并发计划：
DAG:
A → C
B → C
B → D
并行执行计划:
Step 1: [A, B] (并行)
Step 2: [C, D] (B完成后，C和D可并行如果D不依赖C)
错误处理
重试策略
@retry(max_attempts=3, backoff=exponential)
async def call_agent(agent, input):
result = await agent.run(input)
if result.error:
raise AgentError(result.error)
return result
降级策略
- Agent失败 → 使用更简单的Prompt重试
- 工具不可用 → 使用替代工具
- LLM超时 → 使用更快的模型（如从GPT-4降级到GPT-3.5）
- 所有重试失败 → 跳过该步骤，在结果中标注缺失
Checkpoint机制
长流程需要checkpoint：
# 每完成一个phase后保存状态
await self.save_checkpoint(self.state)
# 恢复时从最近checkpoint开始
state = await self.load_checkpoint()
workflow.resume_from(state)
可观测性
日志记录
[2026-07-16 10:00:01] Phase: information_collection
[2026-07-16 10:00:01] Agent: web_search, Query: "AI芯片2025"
[2026-07-16 10:00:03] Agent: web_search, Results: 5 items
[2026-07-16 10:00:03] Agent: db_query, Query: "chip_revenue"
[2026-07-16 10:00:05] Agent: db_query, Results: 12 rows
[2026-07-16 10:00:05] Phase: information_collection, Status: complete
追踪
使用OpenTelemetry追踪Agent执行：
- 每个Agent调用记录为span
- 标记输入、输出、耗时、token数
- 可视化执行DAG
评估指标
- 端到端成功率：任务最终成功比例
- 平均步数：完成任务需要的平均步骤数
- 平均耗时：端到端延迟
- 成本：每任务平均token消耗
- 重试率：需要重试的步骤比例
总结
Agent工作流编排正在从"写死流程"走向"自适应流程"。但完全自适应的Agent在实际生产中仍不够可靠。最务实的方案是混合编排——在关键路径上用DAG保证可控性，在细节执行上给Agent自主决策空间。编排的核心不是"怎么自动执行"，而是"怎么在出错时优雅地处理"。