Agno Workflow 工作流实战指南用 6 个控制流概念搭建稳定的多智能体流水线【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agnoAgno Workflow 工作流编排让你用代码把 Agent、Team 和普通 Python 函数编排成一条可控的自动化流水线顺序执行、条件分支、循环重试、并行加速、智能路由还能随时提前终止。这篇文章先帮你建立心智地图再按场景逐层展开。 一、先建立全景任务拆成 Step控制流决定走向一句话讲清它的核心思想把复杂任务拆成一个个步骤Step再按控制流规则把它们串起来。每个 Step 里可以放一个 Agent、一个 Team、一个普通 Python 函数甚至一个嵌套的子工作流输入是一句字符串或 Pydantic 结构化数据末端输出最终结果。整套体系只有 5 个核心组件先记住这张对照表组件在流水线中的角色一句话理解Step执行单元一个工位放 Agent、Team 或函数Condition条件分支条件满足才走内部步骤否则跳过Loop循环不达标就反复重跑Parallel并行多路同时干完成后自动汇总Router路由根据上下文动态选路可多路入口核心类在 libs/agno/agno/workflow/workflow.py官方示例集中在 cookbook/04_workflows/。 二、从一条直线搭到会分叉、能重试、可加速流水线的复杂度是被场景逼出来的先跑通直线再按需加判断、重试和并行每一步都有对应示例可以照跑。最短可运行的顺序流水线最简单的形态就是一条直线步骤排成列表依次执行。sequence_of_steps.py 里的博客流水线是研究 → 内容规划 → 写作steps[research_step, content_planning_step]就是最完整的一条最短线。三个关键点任何步骤都可以是 Agent、Team 或纯函数函数步骤读取StepInput输入、返回StepOutput就能和智能体步骤无缝混排每个步骤天然能拿到previous_step_content上一步查资料、下一步写文章的衔接不用额外传参。 建议先到 01_basic_workflows/ 目录里把这条直线跑通后面所有变体都是在这条线上加控制节点。会看情况条件与路由怎么选流程一旦需要分岔工具只有两个Condition负责是/否判断Router负责多路选择。Condition 只做二选一。它带一个evaluator评估函数返回True才执行内部步骤否则直接跳过还可以挂else_steps处理否则分支。官方示例 condition_basic.py 是一道事实核查门禁研究和摘要之后只有摘要里出现数据显示百分比这类事实性表述才插入事实核查 Agent否则直接进写作环节省掉一次大模型调用。带 else 分支和条件内并行的变体可看 condition_with_else.py 与 condition_with_parallel.py。Router 处理多路选择。给它一组choices再由selector函数根据上下文动态决定走哪条路甚至可以同时选多条。比如 router_basic.py 的研究工作流主题命中 tech、AI 等关键词就路由给 HackerNews 研究员否则走通用网页研究员最后统一进发布步骤。嵌套路由、字符串选择器、媒体管线等玩法都在 05_conditional_branching/ 目录里。怎么选只有做/不做两种结果用 Condition处理方案是多个明确分支用 Router——用 Condition 层层套是/否可读性远不如一个 Router 干净。会重试循环兜住质量下限质量无法一次保证时把一组步骤包进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。 三、少写代码CEL 表达式与全局提前终止这两件事的共同目标是用更少的代码把规则管起来判断声明式化异常全局兜底。一行字符串完成判断CEL 表达式前面所有evaluator/selector都可以写成 Python 函数但 Agno Workflow 还支持 CELCommon Expression Language把规则直接写成一行字符串不用手写任何函数。使用前先执行pip install cel-python。条件判断evaluatorinput.contains(urgent)输入含 urgent 走紧急处理 Agent否则走普通处理 Agent示例见 cel_basic.py循环退出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/——把规则判断变成声明式配置更进阶的能力集中在 06_advanced_concepts/历史上下文、会话状态、结构化输入输出、运行控制等和 08_human_in_the_loop/人工确认、审批、决策树想深入实现细节可以读 libs/agno/agno/workflow/ 下的condition.py、loop.py、parallel.py、router.py等模块项目整体介绍见 README.md。控制流想清楚之后多智能体业务就不再是一堆散落的模型调用而是一条能测试、能叫停、能交接的线——剩下的事只是把每个工位上的 Step 写扎实。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考