Agno Workflow 实战教程条件分支、循环与并行编排6 个构件搭出自动化执行链【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agnoAgno 是一个用 Python 构建、运行和管理 Agent 平台的开源框架其中的 Workflow 引擎可以把 Agent、Team 和普通函数组装成一条可控制流向的执行链。如果你正在处理多源数据要同时采集、结果质量要卡门禁、不同请求要走不同处理路径这类业务本文就是给你看的跟着三个真实场景走一遍你会掌握 Agno Workflow 的条件分支、循环重试、并行加速、智能路由、CEL 表达式声明和提前终止这几手编排能力。 一条执行链的四种基础构件先建立概念Workflow 里所有东西都是步骤步骤可以嵌套出控制结构。Step最小执行单元里面装 Agent、Team 或普通函数Steps顺序排列的步骤列表就是最朴素的执行链Condition/Loop/Parallel/Router四种控制结构分别管走不走、跑几遍、同时跑、走哪条路。入口类在 libs/agno/agno/workflow/workflow.py四个控制结构分别对应同目录下的 condition.py、loop.py、parallel.py、router.py。最简用法就是一个列表workflow Workflow(steps[Step(nameresearch, agentresearcher), writer])每个步骤执行时都能直接读取上游结果previous_step_content所以上一步查资料、下一步写文章这种衔接不需要你手动传参。想先跑通一条直线链路去 cookbook/04_workflows/01_basic_workflows/ 翻示例。⚡ 场景一多源内容采集用 Parallel 把等待时间砍半问题一篇文章需要刷 HackerNews 热点和抓网页资料两个来源串行跑要等两趟而这两件事毫无依赖。解法Parallel把互不依赖的步骤同时发起全部完成后自动合并Merge结果再交给下游的写作步骤。Parallel(research_hn, research_web)注意它的调用风格是变参形式刻意强调这几路是独立无序的。合并后的输出会作为下一步的输入继续流动。对应的完整流水线并行采集 → 写作 → 审阅在 cookbook/04_workflows/04_parallel_execution/其中parallel_with_condition.py还演示了并行分支里再嵌套条件的写法。 场景二质量门禁只有命中事实性内容才做核查问题不是每篇稿子都需要事实核查无脑加一步核查 Agent 就是白花模型调用费用。解法Condition挂一个evaluator评估函数返回True才执行steps里的内容还可以给else_steps指定否则分支避免非此即彼。Condition(evaluatorhas_factual_claims, steps[fact_checker], else_steps[skimmer])典型玩法是关键词门禁摘要里出现数据显示百分比这类事实性表述才触发核查步骤否则走轻量分支直接放行。判断条件可以是函数、布尔字面量也可以是一句 CEL 表达式后文细讲。三个递进示例依次在 cookbook/04_workflows/02_conditional_execution/基础版condition_basic.py、带 else 分支的condition_with_else.py、条件内再并行的condition_with_parallel.py。 场景三优先级分派让 Router 按内容选路问题客服请求里混着紧急工单和普通咨询你不想在业务代码里写一堆 if-else。解法Router支持多路选择。给它一组choices再给一个selector函数根据上下文决定走哪条路甚至可以同时选中多条。比如按主题关键词把研究任务分流给不同的研究员 Agent命中 tech/AI/startup 走专业研究员否则走通用研究员最后汇合到发布步骤。除了函数选择器还有两个省事模式selector 可以写成返回步骤名的 CEL 表达式或者把requires_user_input设为True暂停下来让人在界面上手动选路Human-in-the-Loop。示例集中在 cookbook/04_workflows/05_conditional_branching/嵌套路由、字符串选择器、媒体管线都有覆盖。 循环重试的两个退出判据问题一次性产出的研究结果经常太短、太敷衍你想让它再来一遍直到够用。解法Loop把一组步骤包起来反复执行退出只有两个判据必须配齐其一end_condition评估函数接收本轮所有步骤的输出返回True即结束。比如检查输出是否超过 200 字符够长才停max_iterations上限默认 3 轮作为保底机制防止死循环。两者都满足不了时以轮次上限为准。Loop(steps[research_hn, research_web], max_iterations3)如果希望上一轮的输出喂给下一轮继续加工而不是每轮都从原始输入重来还有carry_forward_output开关。两种变体值得对照着看带并行的loop_with_parallel.py和逐轮累积材料的loop_iterative_accumulation.py都在 cookbook/04_workflows/03_loop_execution/。 用一句表达式声明规则以及给链路装上急停规则判断如果都写成 Python 函数代码量会越堆越多。Agno Workflow 支持CELCommon Expression Language表达式evaluator、end_condition、selector三个控制点都可以直接传一句字符串需先pip install cel-python表达式里可引用input、previous_step_content、previous_step_outputs、session_state等上下文对象。条件判断evaluatorinput.contains(urgent)命中 urgent 走紧急处理 Agent否则走普通 Agent循环退出current_iteration 2、all_success这类写法直接限制迭代路由选择selector 写一句返回步骤名的表达式即可。完整的声明式写法示例在 cookbook/04_workflows/07_cel_expressions/condition、loop、router 三个子目录一一对应。另一手保命能力是提前终止当输入明显不合规或中间结果已失败时可以叫停整条链路剩余步骤直接跳过、执行链以提前停止状态结束而不是硬跑完浪费时间。直线步骤、条件、循环、并行四种结构各自的终止写法cookbook/04_workflows/06_advanced_concepts/early_stopping/ 目录里有四份对照示例basic / condition / loop / parallel。️ 按什么顺序上手一句话总结Step 打底Condition/Loop/Parallel/Router 管流向CEL 声明规则Early Stop 兜底——六个概念就是 Agno Workflow 编排的全部家底。下一步建议clone 仓库后git clone https://gitcode.com/GitHub_Trending/ag/agno按01_basic_workflows → 02_conditional_execution → 03_loop_execution → 04_parallel_execution的顺序把 cookbook/04_workflows/ 里的示例逐个跑起来遇到读不懂的行为再回头翻 libs/agno/agno/workflow/ 的对应源码文件比先啃文档高效得多。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考