Agno Workflow 实战教程：条件分支、循环与并行编排，6 个构件搭出自动化执行链
【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno
Agno 是一个用 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(name="research", agent=researcher), 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(evaluator=has_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_iterations=3)如果希望上一轮的输出喂给下一轮继续加工（而不是每轮都从原始输入重来），还有carry_forward_output开关。两种变体值得对照着看：带并行的loop_with_parallel.py和逐轮累积材料的loop_iterative_accumulation.py，都在 cookbook/04_workflows/03_loop_execution/。
📝 用一句表达式声明规则，以及给链路装上急停
规则判断如果都写成 Python 函数，代码量会越堆越多。Agno Workflow 支持CEL（Common Expression Language）表达式：evaluator、end_condition、selector三个控制点都可以直接传一句字符串（需先pip install cel-python），表达式里可引用input、previous_step_content、previous_step_outputs、session_state等上下文对象。
- 条件判断：
evaluator='input.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），仅供参考
