Agno Workflow 工作流编排:7 个控制原语,把多智能体流水线拧成可控机器【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno把三个 Agent 按顺序串起来谁都会,真正的门槛在后面:让流水线自己判断该不该核查、要不要重试、派给哪个处理队列、何时一脚刹车。Agno Workflow 的工作流编排把这四件事收敛成代码里几个可组合的原语,下面按控制场景拆开讲。 控制原语全景:流水线里的七个开关先用 30 秒建立心智模型。除 Step 之外的原语,干的都是同一件事:决定这一步以什么方式继续。原语它替你决定的事换成产线的说法Step最小执行体,接 Agent、Team 或普通函数一道工序Condition评估函数返回 True 才放行,否则跳过安检门Loop反复执行,直到 end_condition 满足或 max_iterations 兜底返工台Parallel多分支并发,全部完成后自动合并结果并道汇流Routerselector 从 choices 里挑一条或多条路岔路口CEL一行表达式字符串替代 evaluator/selector 函数免代码规则StepStop()任意步骤触发即终止整条流水线紧急制动直线流水线只用 Step 就能跑:每个步骤通过previous_step_content拿到上一步的输出,上一步取数、下一步清洗这类衔接不需要任何胶水代码。 客服工单分拣:该不该做、做几遍、走哪条路决定流水线成熟度的不是执行速度,而是它在每个岔路口做的判断。给高危工单加一道安检门工单流水线大致是接单、分类、自动回复。在分类后插入一个 Condition:评估函数扫一遍工单正文,命中退款投诉法务等词就走人工升级分支,否则放行自动回复。门禁的价值恰恰在于普通工单不再唤起昂贵的处理 Agent。最小写法可看 condition_basic.py 里的事实核查门禁;想给否分支也安排去处,看 condition_with_else.py。让 Loop 给回复草稿上返工台自动生成的回复经常一次不达标。loop_basic.py 把两个采集步骤包进 Loop:评估函数只在输出足够长时放行(示例阈值为两百字符),max_iterations 兜底防止无限返工。可抄的套路是先写清退出标准、再谈重试;要多轮之间携带中间结果,参考 loop_iterative_accumulation.py。Router 把工单派进不同处理队列Condition 只回答是否,工单却常分三四类:退款进财务队列,技术故障进工程师队列,其余归通用客服。router_basic.py 演示 choices selector 的标准多路分发;nested_choices.py 把 Router 嵌进 Router 的分支里,适配先按领域、再按优先级的两级分拣。⚡ 多源 ETL 提速:并行合并与声明式规则步骤之间没有依赖,串行就是纯粹浪费;规则简单到一行能写完,手写函数就是多余代码。并行采集后的合并顺序数据管道同时从新闻、社交、论坛三个源取数。parallel_basic.py 用Parallel(step_a, step_b, nameResearch Phase)并发取数,全部分支跑完后结果自动汇合,再流入后续的清洗与校验步骤,无需手工拼装;耗时由最慢的一支决定。想给并行后加质量门禁,参考 parallel_with_condition.py 的组合写法。用一行 CEL 字符串替代手写分支函数规则化的判断,evaluator 和 selector 都可以直接写成 CEL 表达式字符串(需先pip install cel-python)。cel_basic.py 是最小演示:Condition( nameUrgent Check, evaluatorinput.contains(urgent), steps[Step(agenturgent_handler)], else_steps[Step(agentnormal_handler)], )表达式能引用input、previous_step、session_state等上下文对象。循环退出条件同样可以表达式化:cel_iteration_limit.py 用一行限制迭代次数,07_cel_expressions/router/ 收录了按上一步输出选路的完整示例。条件、循环、路由三大控制点全部能落成配置字符串,不写一个函数定义。 提前终止:给整条流水线装一脚刹车StepStop()是全局信号:某一步触发它,剩余步骤整体跳过,工作流标记为 Early Stopped 结束。典型时机是输入明显不合规、中间结果已判定失败、用户取消任务——此时不应让管道继续烧完后面的模型调用。最小触发方式,以及与条件、循环、并行的组合场景,见 early_stopping/ 目录下的四个示例。️ 动手路线图:四步把控制流装进管道直线跑通:从 01_basic_workflows/ 开始,让steps列表先跑起来,弄清步骤之间如何传递上下文;门禁与返工:在 02_conditional_execution/ 给管道加 Condition 分支,再到 03_loop_execution/ 补上重试;并发与路由:在 04_parallel_execution/ 用 Parallel 压缩采集耗时,再用 05_conditional_branching/ 的 Router 做多路分发;声明式与源码:在 07_cel_expressions/ 把手写函数换成表达式,然后按 workflow.py、condition.py、loop.py、parallel.py、router.py 的顺序读 libs/agno/agno/workflow/ 源码。一条能上生产的流水线,不是跑得最快的那条,而是每一步都写好了退路的那条。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Agno Workflow 工作流编排:7 个控制原语,把多智能体流水线拧成可控机器
Agno Workflow 工作流编排:7 个控制原语,把多智能体流水线拧成可控机器【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno把三个 Agent 按顺序串起来谁都会,真正的门槛在后面:让流水线自己判断该不该核查、要不要重试、派给哪个处理队列、何时一脚刹车。Agno Workflow 的工作流编排把这四件事收敛成代码里几个可组合的原语,下面按控制场景拆开讲。 控制原语全景:流水线里的七个开关先用 30 秒建立心智模型。除 Step 之外的原语,干的都是同一件事:决定这一步以什么方式继续。原语它替你决定的事换成产线的说法Step最小执行体,接 Agent、Team 或普通函数一道工序Condition评估函数返回 True 才放行,否则跳过安检门Loop反复执行,直到 end_condition 满足或 max_iterations 兜底返工台Parallel多分支并发,全部完成后自动合并结果并道汇流Routerselector 从 choices 里挑一条或多条路岔路口CEL一行表达式字符串替代 evaluator/selector 函数免代码规则StepStop()任意步骤触发即终止整条流水线紧急制动直线流水线只用 Step 就能跑:每个步骤通过previous_step_content拿到上一步的输出,上一步取数、下一步清洗这类衔接不需要任何胶水代码。 客服工单分拣:该不该做、做几遍、走哪条路决定流水线成熟度的不是执行速度,而是它在每个岔路口做的判断。给高危工单加一道安检门工单流水线大致是接单、分类、自动回复。在分类后插入一个 Condition:评估函数扫一遍工单正文,命中退款投诉法务等词就走人工升级分支,否则放行自动回复。门禁的价值恰恰在于普通工单不再唤起昂贵的处理 Agent。最小写法可看 condition_basic.py 里的事实核查门禁;想给否分支也安排去处,看 condition_with_else.py。让 Loop 给回复草稿上返工台自动生成的回复经常一次不达标。loop_basic.py 把两个采集步骤包进 Loop:评估函数只在输出足够长时放行(示例阈值为两百字符),max_iterations 兜底防止无限返工。可抄的套路是先写清退出标准、再谈重试;要多轮之间携带中间结果,参考 loop_iterative_accumulation.py。Router 把工单派进不同处理队列Condition 只回答是否,工单却常分三四类:退款进财务队列,技术故障进工程师队列,其余归通用客服。router_basic.py 演示 choices selector 的标准多路分发;nested_choices.py 把 Router 嵌进 Router 的分支里,适配先按领域、再按优先级的两级分拣。⚡ 多源 ETL 提速:并行合并与声明式规则步骤之间没有依赖,串行就是纯粹浪费;规则简单到一行能写完,手写函数就是多余代码。并行采集后的合并顺序数据管道同时从新闻、社交、论坛三个源取数。parallel_basic.py 用Parallel(step_a, step_b, nameResearch Phase)并发取数,全部分支跑完后结果自动汇合,再流入后续的清洗与校验步骤,无需手工拼装;耗时由最慢的一支决定。想给并行后加质量门禁,参考 parallel_with_condition.py 的组合写法。用一行 CEL 字符串替代手写分支函数规则化的判断,evaluator 和 selector 都可以直接写成 CEL 表达式字符串(需先pip install cel-python)。cel_basic.py 是最小演示:Condition( nameUrgent Check, evaluatorinput.contains(urgent), steps[Step(agenturgent_handler)], else_steps[Step(agentnormal_handler)], )表达式能引用input、previous_step、session_state等上下文对象。循环退出条件同样可以表达式化:cel_iteration_limit.py 用一行限制迭代次数,07_cel_expressions/router/ 收录了按上一步输出选路的完整示例。条件、循环、路由三大控制点全部能落成配置字符串,不写一个函数定义。 提前终止:给整条流水线装一脚刹车StepStop()是全局信号:某一步触发它,剩余步骤整体跳过,工作流标记为 Early Stopped 结束。典型时机是输入明显不合规、中间结果已判定失败、用户取消任务——此时不应让管道继续烧完后面的模型调用。最小触发方式,以及与条件、循环、并行的组合场景,见 early_stopping/ 目录下的四个示例。️ 动手路线图:四步把控制流装进管道直线跑通:从 01_basic_workflows/ 开始,让steps列表先跑起来,弄清步骤之间如何传递上下文;门禁与返工:在 02_conditional_execution/ 给管道加 Condition 分支,再到 03_loop_execution/ 补上重试;并发与路由:在 04_parallel_execution/ 用 Parallel 压缩采集耗时,再用 05_conditional_branching/ 的 Router 做多路分发;声明式与源码:在 07_cel_expressions/ 把手写函数换成表达式,然后按 workflow.py、condition.py、loop.py、parallel.py、router.py 的顺序读 libs/agno/agno/workflow/ 源码。一条能上生产的流水线,不是跑得最快的那条,而是每一步都写好了退路的那条。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
