工作流编排与停止证据
Router 选择路径,Supervisor 分配并汇总,handoff 转移控制权;它们都不应绕过任务状态、权限和预算。工作流完成需要目标对应的外部证据,例如测试通过、引用存在或工单状态已更新。模型输出“完成”只是候选事件。
第一个可运行实验:用验收证据结束工作流
保存为 workflow.py,用 Python 3.12 运行。
python
def finish(state):
if state['cancelled']:
return 'cancelled'
if state['steps'] >= state['max_steps']:
return 'budget_exhausted'
if state['permission_denied']:
return 'permission_denied'
if state['evidence'].get('tests_passed') and state['evidence'].get('citations'):
return 'completed'
return 'running'
base = {'cancelled': False, 'steps': 2, 'max_steps': 5,
'permission_denied': False, 'evidence': {}}
assert finish(base) == 'running'
assert finish({**base, 'evidence': {'tests_passed': True, 'citations': ['doc:1']}}) == 'completed'
assert finish({**base, 'steps': 5}) == 'budget_exhausted'
print('workflow stop evidence passed')多个子任务可以并行,但写入同一资源必须有独占范围或版本检查。Supervisor 汇总时要保留每个子任务来源和失败状态,不能把缺失结果拼成成功报告。handoff 前交接任务目标、已验证事实、未解决问题、权限和剩余预算。
变体实验
让一个子任务引用为空,要求整体不能完成。再让权限拒绝和测试通过同时出现,确认安全拒绝优先。
练习
- 为 handoff 设计最小交接包。
- 设计部分成功、人工接管和整体失败的汇总状态。
- 说明何时固定 pipeline 比动态 Agent 更可靠。