Agno Workflow工作流编排实战条件分支、循环、并行与CEL表达式打造自动化流水线【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agnoAgnoAgno Workflow是一款 Python 多智能体开发框架内置的Workflow 工作流引擎让你用代码把多个 Agent、Team 和普通函数编排成一条可控的自动化流水线支持顺序执行、条件分支、循环、并行、路由Router还能用 CEL 表达式一行搞定分支判断并可随时提前终止Early Stopping。本文带你从零基础看懂 Agno Workflow 的 5 大核心组件并附完整示例路径快速搭建属于你自己的智能体流水线。 先看懂全景图Agno Workflow 长什么样Agno Workflow 的核心思想很简单把复杂任务拆成一个个步骤Step再按控制流规则串起来。每个步骤可以是 Agent智能体、Team智能体团队、普通 Python 函数甚至是嵌套的子工作流。工作流的入口Workflow支持同步/异步、一次性输出/流式输出还能接入 SQLite 等数据库保存会话实现状态持久化。官方示例位于 cookbook/04_workflows/入口核心类可参见 libs/agno/agno/workflow/workflow.py。Agno Workflow 的 5 大核心组件一览组件作用一句话理解Step执行单元流水线上的一个工位Condition条件分支满足条件才走这段Loop循环执行不达标就反复重试Parallel并行执行多路同时干最后汇总Router智能路由根据内容自动选路 顺序步骤最简单的流水线最基础的 Workflow 就是步骤列表按顺序执行。在 sequence_of_steps.py 中作者搭建了研究 → 内容规划 → 写作的博客流水线steps[research_step, content_planning_step]每个步骤可以是 Agent、Team也可以直接是一个函数——函数通过读取StepInput、返回StepOutput参与流水线天然能访问前一步的输出。 小贴士任何步骤都能拿到previous_step_content上一步内容这让上一步查资料、下一步写文章的衔接非常自然。 条件分支只有需要时才执行Condition组件让流水线学会看情况办事。它带一个evaluator评估函数返回True才执行内部步骤否则直接跳过。经典案例事实核查门禁Fact-Check Gate。在 condition_basic.py 中流程是研究 → 摘要 →条件判断→ 写文章只有当摘要里出现数据显示百分比研究证明等事实性关键词时才插入事实核查 Agent 这一步否则直接跳过进入写作。这样既保证了内容严谨性又不浪费额外的大模型调用。Condition还支持else_steps否则分支基础用法condition_basic.py带 else 分支condition_with_else.py条件内并行condition_with_parallel.py 循环执行不达标就再来一次当任务质量无法一次保证时Loop组件可以让一组步骤反复执行直到end_condition评估函数判定够了或用max_iterations兜底防止死循环。在 loop_basic.py 中研究 HackerNews 研究网页两个步骤被包进一个Loop(max_iterations3)评估函数检查输出内容是否超过 200 字符——够长才退出循环去创作内容。更多变体带并行的循环loop_with_parallel.py累积式循环loop_iterative_accumulation.py⚡ 并行执行多路同时跑最后汇总如果两个研究任务互不依赖就没必要串行等待。Parallel组件把多个步骤同时发起全部完成后自动合并Merge结果再进入下一步。parallel_basic.py 里的内容创作流水线非常直观Parallel(研究HackerNews, 研究网页)并行跑完 → 写作步骤 → 审阅步骤。对于多源采集类任务这一步能显著缩短整体耗时。想进一步控制流程可看 parallel_with_condition.py并行 条件组合。 Router 智能路由按内容自动分流与Condition的是/否不同Router支持多路选择给它一组choices再由selector函数根据上下文动态决定走哪条路甚至多条路。例如 router_basic.py 中的智能研究工作流selector 检查主题是否包含 tech、AI、startup 等关键词——命中就路由到 HackerNews 研究员否则走通用网页研究员最后统一进入发布步骤。更多玩法嵌套路由、字符串选择器、媒体管线都在 05_conditional_branching/ 目录中。 CEL 表达式一行字符串搞定分支判断前面提到的evaluator/selector都可以写成 Python 函数但 Agno Workflow 还支持CELCommon Expression Language表达式——直接写一行字符串即可无需手写任何函数尤其适合规则化的简单判断。比如判断紧急请求的优先级路由示例见 cel_basic.py需先pip install cel-python条件写法evaluatorinput.contains(urgent)—— 输入含 urgent 走紧急处理 Agent否则走普通处理 Agent循环退出cel_iteration_limit.py用表达式限制迭代次数路由选择07_cel_expressions/router/ 演示了用 CEL 读取上一步输出、会话状态来选路CEL 表达式还能引用input、previous_step、session_state等上下文对象让条件、循环、路由三大控制点全部实现声明式配置是搭建自动化流水线时最省代码的利器。 Early Stopping随时叫停整个流水线除了局部跳过Agno Workflow 还支持全局提前终止当某个步骤触发StepStop()时剩余步骤将被直接跳过工作流标记为Early Stopped结束。典型场景输入明显不合规、中间结果已失败、或用户取消任务。示例见 06_advanced_concepts/early_stopping/ 目录。 上手路径与延伸阅读想动手实践按这个顺序走效率最高顺序执行01_basic_workflows/ —— 先跑通一条直线流水线条件分支02_conditional_execution/循环执行03_loop_execution/并行执行04_parallel_execution/路由与分支05_conditional_branching/CEL 表达式07_cel_expressions/更进阶的能力历史上下文、会话状态、结构化输入输出、Human-in-the-Loop 人工审批等集中在 06_advanced_concepts/ 与 08_human_in_the_loop/。工作流引擎源码位于 libs/agno/agno/workflow/可深入workflow.py、condition.py、loop.py、parallel.py、router.py等模块理解实现细节整体项目说明参见 README.md。 一句话总结Step 打底、Condition/Loop/Parallel/Router 控制走向、CEL 简化规则、Early Stop 兜底——掌握这 6 个概念你就能用 Agno Workflow 把任何多智能体业务编排成稳定可靠的自动化流水线。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考