Agno Workflow 工作流编排实战：条件分支与并行速查

Agno Workflow 工作流编排实战：条件分支与并行速查

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno

假设一条内容要经过采集、摘要、事实核查、写作、审阅。手写 if-else 链路，步骤一多就很难回答"哪一步执行了、在哪里分叉、中间结果去了哪"。Agno（Agno）的 Workflow 引擎用工作流编排解决这个问题：把任务拆成 Step（执行单元，相当于流水线上的一个工位），再用 Condition（条件分支）、Loop（循环）等控制结构串联，整条链路可观测、可暂停、可回放。

编排原语速览：五种组件一次讲清

五种原语彼此可以嵌套：Condition 里放 Router，Loop 里放 Parallel。每种原语的源码都在 libs/agno/agno/workflow/ 下独立成模块，逐个读没有负担。

内容生产：从采集到成稿的串并联组合

这是最常见的拓扑：两个互不依赖的采集任务并行，结果合并后再交给写作。Parallel（并行容器）同时发起内部步骤，全部完成后自动把各路输出汇总为下一步的previous_step_content，下游写作步骤不需要知道上游有几个来源。

组合的关键就在并行后的合并：

Parallel(steps=[hn_research_step, web_research_step])

完整示例见 parallel_basic.py，它演示了"并行研究 → 写作 → 审阅"的完整链路，同目录下还有并行加条件的组合写法。

质量门禁：条件判断 + 循环重试如何协作

质量诉求分两类。一类是"该不该跑"：摘要里出现"数据显示""百分比"这类事实性关键词时，才需要事实核查步骤——这是Condition，evaluator 返回 bool，决定进入 if 分支还是 else 分支。另一类是"跑到达标为止"：研究结果不足 200 字符就再来一轮——这是Loop，每轮结束后由 end_condition 评估是否退出，max_iterations（默认 3）兜底防止死循环。

condition_basic.py 里的门禁 evaluator 读step_input.previous_step_content做关键词匹配；loop_basic.py 的循环 evaluator 直接检查输出长度。两者可以叠用：Loop 内嵌 Condition，实现"本轮输出达标才进入审阅"这类复合门禁。

动态分流：多路路由与声明式规则

分支超过两条时，if-else 就开始难维护，这是Router（多路分发器：根据上下文动态选择执行哪条分支）的活。声明一组候选步骤choices，selector函数根据 StepInput 动态返回走哪条路，也可以配纯字符串选择器或人工选择。

router_basic.py 里 selector 检查主题是否包含 tech、AI 等关键词，命中走 HackerNews 研究员，未命中走通用网页研究员，最后统一进入发布步骤。嵌套路由、字符串选择器、媒体管线等进阶写法集中在 05_conditional_branching/ 目录。

跨场景进阶技巧：CEL、提前终止与人工确认

• 遇到分支规则都是简单的内容匹配 → 用 CEL 表达式（通用表达式语言：用一行字符串写判断逻辑），例如evaluator='input.contains("urgent")'，无需写函数。示例见 cel_basic.py，需pip install cel-python。
• 遇到输入明显不合规或中间结果已失败，不想再跑剩余步骤 → 在函数步骤里返回StepOutput(stop=True)，整条流水线标记为提前终止并跳过后续步骤，见 early_stop_basic.py。
• 遇到循环需要把本轮结果作为下一轮起点 → 设置Loop(forward_iteration_output=True)，默认每轮拿的是原始输入，不传上一轮结果。
• 遇到关键分支需要人拍板 → 给 Condition 或 Router 配置 human_review 确认项，流程会暂停等待人工决定再继续，示例在 08_human_in_the_loop/ 目录。
• 遇到多轮任务要跨次运行恢复 → Workflow 入口支持接入 SQLite 等数据库持久化会话，每步输出与运行状态可回放。

动手路线：五步能力递进

1. 先跑通顺序流水线：从 01_basic_workflows/ 选一条"研究 → 摘要 → 写作"直线流水线，理解 Step 与 StepOutput 如何在步骤间传递。
2. 加一个条件分支：把摘要后加事实核查门禁，参考 condition_basic.py，改几个关键词观察 evaluator 返回 True/False 的差异。
3. 加一个循环重试：把研究步骤包进Loop(max_iterations=3)，调 end_condition 感受不同退出时机。
4. 加并行：把两个独立研究源改成Parallel，对比串行与并行下输出合并的差别。
5. 加路由：把关键词 if-else 改写为Router(choices=...)，最后把 selector 换成 CEL 表达式，完成声明式收尾。

读源码时按执行链路来：workflow.py 是主循环，condition.py、loop.py、router.py 各对应一种控制结构，文件不大，一个下午能读完。

Workflow 引擎的真正价值不在组件数量，而在步骤之间有一份清晰的契约：每个步骤接收 StepInput、返回 StepOutput，控制结构只负责决定"何时执行哪个工位"。契约立住了，新增一个分支、一轮重试或一个人工检查点都只是局部改动。生产化时建议先接上流式事件与会话持久化，把可观测性做足，再叠加复杂路由。

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno