把多个智能体串成一条可控流水线：Agno Workflow 的 5 种控制流玩法

把多个智能体串成一条可控流水线Agno Workflow 的 5 种控制流玩法【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno先看一个常见的纠结场景新闻生成任务里研究 Agent 查完资料后要不要再走一遍事实核查这取决于摘要里有没有出现数据。如果你用过程式代码写if/else 的分支逻辑会跟模型调用搅在一起越写越乱。Agno Workflow 就是为这类编排问题设计的它把复杂任务拆成一个个 Step步骤用控制流规则把它们串起来Agent、Team 和普通函数都能进入同一条流水线。 核心模型Step 与输入输出如何传递Step是流水线上唯一的执行单元你可以把它理解成地铁线路里的一站流程是线路列车停在哪、往哪开由线路设计决定。一个 Step 的内容可以是一个Agent、一个Team也可以是一个普通函数——函数只要接收StepInput、返回StepOutput就能和智能体站在同一条线上。步骤之间的传递靠previous_step_content任何步骤都能直接读到上一步的输出先查资料、再写文章这种衔接不需要你手动传变量。入口对象Workflow支持同步和异步两种调用方式输出可以一次性返回也可以流式返回接入 SQLite 等存储后流水线的会话还能持久化、可回放。想先看最朴素的接线方式读 sequence_of_steps.py研究、规划、写作三步顺序排列每一步是不同类型的执行器。 四种执行模式怎么选下表把四种模式放在一起方便你按场景对号入座模式适用场景关键参数直线串联研究 → 规划 → 写作的线性流水线steps分叉与选路事实核查门禁、按主题分流evaluator、else_steps、choices、selector循环重试输出质量一次无法保证end_condition、max_iterations并行分流多个互不依赖的研究任务Parallel内步骤、自动合并直线串联先把流水线跑通这是流程的默认形态把步骤排成一个列表依次执行。同一个目录下的 sequence_with_functions.py 展示了函数如何成为步骤——你不必为每个环节都建一个 Agent。实操建议先用直线把整条链路跑通确认输入输出符合预期再考虑加分支。分叉与选路Condition 管二选一Router 管多路Condition处理是/否二分evaluator函数返回True时执行内部步骤否则跳过。condition_basic.py 里的门禁逻辑很典型研究 → 摘要 → 判断 → 写作只有摘要中出现数据显示研究证明这类事实性关键词时才插入事实核查 Agent省掉一次不必要的模型调用想让否则那一路也有动作可以配置else_steps见 condition_with_else.py。Router处理多路按内容选路给它一组choices由selector函数根据上下文动态挑一条或几条路径。router_basic.py 里主题命中 tech、AI 等关键词就路由到 HackerNews 研究员否则走通用网页研究员最后统一汇入发布步骤。选择标准只有两种情况用Condition三条以上互斥路径、且路径依赖内容判断时用Router。循环重试退出条件与迭代上限怎么配合Loop把一组步骤反复执行直到end_condition评估函数判定够了。loop_basic.py 中两个研究步骤被包进Loop评估函数检查输出是否超过 200 字符同时设置了max_iterations3——退出条件再怎么不满足三圈之后也会强制收手。这两个参数是质量门槛 成本保险的关系实操中建议永远同时写。并行分流分叉、并发、再合并Parallel把多个步骤同时发起全部完成后自动合并Merge结果再进入下一步。parallel_basic.py 里HackerNews 研究与网页研究两个互不依赖的任务同时跑然后进入写作步骤——总耗时取决于最慢的一路而不是各路径之和。如果并行分支内部还要做判断参考 parallel_with_condition.py。 声明式判断用一行 CEL 表达式替代评估函数前面提到的evaluator/selector通常要手写 Python 函数Agno Workflow 还允许直接传一行 CEL 表达式字符串需先pip install cel-python。比如输入含 urgent 就走紧急处理判断只有一行evaluatorinput.contains(urgent)cel_basic.py 把这行表达式配上else_steps就得到一个完整的优先级分流流程。表达式里还能引用input、previous_step、session_state等上下文对象循环退出、路由选路同样适用end_conditioncurrent_iteration 2出现在 cel_iteration_limit.py路由示例集中在 router/ 目录。实操建议规则简单就用 CEL 让流程定义保持可读业务逻辑复杂再退回函数。 中断与兜底两套安全机制Early Stopping是全局断路器任意步骤返回StepOutput(stopTrue)剩余步骤直接跳过流程以Early Stopped状态收尾。early_stopping/ 目录里的 early_stop_basic.py 给了典型示范安全扫描发现 VULNERABLE 时门禁函数返回stopTrue部署与监控步骤一步都不会执行。输入明显不合规、中间结果已失败、用户取消任务都属于该叫停的场景。循环上限是局部兜底max_iterations约束的是单点成本Early Stopping 约束的是整条链路两者一个管一段、一个管全程。设计时想清楚风险在哪一层就在哪一层设闸。 实践路径官方示例顺序与源码入口按这个顺序动手前一步是后一步的地基直线串联01_basic_workflows/条件执行02_conditional_execution/循环执行03_loop_execution/并行执行04_parallel_execution/路由与分支05_conditional_branching/CEL 表达式07_cel_expressions/想理解引擎内部如何驱动这些控制流源码在 libs/agno/agno/workflow/按workflow.py、condition.py、loop.py、parallel.py、router.py的顺序读每个文件对应一种执行模式。历史上下文、会话状态、人工介入等进阶玩法分别在 06_advanced_concepts/ 和 08_human_in_the_loop/ 里展开等你跑通上面六步之后再去看。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考