尧图网站建设 
Agno Workflow 实战指南六种控制流搭建可控多智能体流水线【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agnoAgno Workflow 是 Agno 多智能体框架内置的工作流编排引擎能把 Agent、Team 与普通函数串成一条可控流水线条件分支、循环执行、并行执行、路由与提前终止均为一等公民。本文以内容流水线为例演示如何逐层加装六种控制流。心智模型 把流程想象成一条传送带工作流编排的入口是Workflow类它的steps参数接收一列步骤。整条流水线可以想象成传送带输入从带首进入每个步骤从上游拿到结果、做完自己的加工再往下传。步骤不局限于 Agent 和 Team——普通 Python 函数、嵌套的子工作流都能直接排进队列。函数步骤接收StepInput内含input原始输入和previous_step_content上一步产出返回StepOutput上下游衔接就这样自然完成了。官方示例 sequence_of_steps.py 搭了一条研究 → 内容规划 → 写作的直线流水线同时演示了同步、异步与流式四种运行方式。最小形态如下workflow Workflow( nameblog-post, steps[ Step(nameresearch, agentresearcher), Step(namewrite, agentwriter), ], ) workflow.print_response(inputAI trends in 2026)后续所有控制流组件本质都是在直线上增加分叉、回卷、汇合的弯折能力解决什么问题对应组件顺序串联固定流程按序执行steps列表门禁校验二值规则决定某段是否跳过Condition多路分流按内容选择一条或多条路径Router重试回卷单次产出不稳定需重复直到达标Loop并发采集多个独立数据源串行等待浪费时间Parallel全局熔断发现风险时终止整条流水线stopTrue输出本节要点接受传送带这个隐喻后读后面所有组件时只需回答一个问题——它让传送带在哪里弯折。给流水线加判断力Condition、Router 与 CEL 三件套 ⚖️最简单的判断是二值门禁。Condition组件带一个evaluator评估函数返回真才执行内部步骤。condition_basic.py 的做法是研究 → 摘要 → 门禁 → 写文章评估函数扫描摘要里是否出现research indicatespercent等事实性措辞命中才插入事实核查 Agent避免无谓的模型调用Condition( namefact_check_gate, evaluatorneeds_fact_checking, # 读取 previous_step_content 判断 steps[fact_check_step], )需要兜底分支时用else_steps补上否则路径见 condition_with_else.py。判断从二值扩展到多路时换用Router声明一组choices由selector函数依据上下文动态返回要执行的步骤可一条也可多条。router_basic.py 中主题含 tech、AI 等关键词时路由到 HackerNews 研究员否则走通用网页研究员两路汇合后统一进入发布步骤。同目录还有嵌套路由、字符串选择器、媒体管线等变体。若规则足够简单可以连函数都不写。Agno 支持 CELCommon Expression Language一种声明式表达式语言evaluator直接传字符串即可例如evaluatorinput.contains(urgent)输入命中urgent走加急 Agent否则走常规 Agent样例见 cel_basic.py需pip install cel-python。CEL 还能引用previous_step、session_state等上下文对象Loop的end_condition同样适用——07_cel_expressions/loop/ 里的cel_iteration_limit.py就用current_iteration 2一个字符串完成了退出判断。本节要点Condition 管走不走Router 管走哪条CEL 管少写代码——三者可叠加在同一层。给流水线加并发力Loop 设上限与底线Parallel 摊开吞吐 ⚡先说为什么需要重复。大模型的单次输出并不可靠同一研究任务跑一遍可能只吐出一段空泛的摘要。Loop的设计思路不是不达标就重试而是给两个约束——end_condition给质量设上限max_iterations给失败设底线。loop_basic.py 把研究 HackerNews 研究网页两个步骤包进Loop(max_iterations3)评估函数检查任一输出是否超过 200 字符够厚实即提前退出进入创作三次都不达标则带有限产出退出既不会死循环也不会把半成品当成品往下传。若希望每轮结果累加而非覆盖参考loop_iterative_accumulation.py。再说为什么需要并行。两个研究任务互不依赖时串行等待纯属浪费。Parallel把所有分支同时发起全部完成后自动合并结果再进下一步。parallel_basic.py 的结构是并行研究阶段HN 网页→ 写作 → 审阅多源采集场景下整体耗时约减半steps[ Loop(steps[hn_step, web_step], end_conditionresearch_evaluator, max_iterations3), Parallel(hn_step, web_step, nameresearch phase), write_step, ],想给并行分支再套一层判断见 parallel_with_condition.py。本节要点Loop 的max_iterations是预算护栏而非装饰Parallel 的前提是分支间无共享写依赖。安全阀 ️提前终止与状态持久化Condition只能跳过一段真正的事故需要熔断整条流水线。步骤函数返回StepOutput(stopTrue)时引擎跳过所有剩余步骤并将运行标记为提前终止。early_stop_basic.py 里的部署管线很典型安全扫描返回 VULNERABLE门禁函数发出停止信号部署与监控配置一步都不会执行数据校验、内容质检管线也是同一模式。状态方面给Workflow挂上dbSqliteDb(session_table...)即可持久化会话支持跨运行续接session_state字典适合放跨运行的标志位例如 state_in_condition.py 里只对首次出现的用户打招呼就靠它实现。需要人工介入的环节步骤确认、用户输入、错误重试审批08_human_in_the_loop/ 目录按确认、决策树、超时等场景组织了一整套示例。本节要点stop 信号优于异常——它让提前结束成为可预期的正常路径而不是错误路径。组合技巧与避坑 组件组合的自由度比表面更高但有几个易踩的坑嵌套是双向的。Workflow 本身可以作为步骤排进另一个 Workflow见 06_advanced_concepts/ 下的 nested_workflow 系列Condition内部也能装Parallelcondition_with_parallel.py。嵌套层级建议不超过两层否则事件流难以追踪。Loop 必须带底线。只写end_condition不写max_iterations一旦评估函数永远返回假就会无限消耗 token。CEL 有隐式依赖。表达式写法依赖 cel-python 包引擎通过CEL_AVAILABLE检测未安装时直接退出在 CI 环境里要提前声明该依赖。并行分支的写竞争。多个 Parallel 分支同时读写session_state存在先后不可控写依赖放在 Parallel 合并点之后最稳妥。门禁函数的返回值要显式。通过路径也建议显式传stopFalse让继续和熔断在代码里都可见。本节要点控制流组合的边界不在引擎而在事件流的可读性——弯折越多越要留日志。动手顺序建议与源码入口按依赖关系排序四步走效率最高先跑通 sequence_of_steps.py需配置模型 API 密钥亲手改一处previous_step_content的用法理解步骤间数据怎么流。在流水线上加一个Condition门禁评估函数从关键词匹配起步再试else_steps。把串行研究段替换为Parallel或Loop对比两种改造在耗时与 token 上的差异。需要声明式规则时引入 CEL需要持久化与人工确认时再上数据库和 HITL。引擎实现集中在 libs/agno/agno/workflow/ 目录主循环在 workflow.py各控制流组件分别对应condition.py、loop.py、parallel.py、router.py、cel.py排查控制流行为时按文件名定位即可。收尾要点Step 打底判断、路由、循环、并行逐层加装CEL 压缩规则代码stop 信号兜底——六种控制流装齐多智能体流水线就有了生产级的骨架。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Agno Workflow 实战指南：六种控制流搭建可控多智能体流水线
Agno Workflow 实战指南六种控制流搭建可控多智能体流水线【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agnoAgno Workflow 是 Agno 多智能体框架内置的工作流编排引擎能把 Agent、Team 与普通函数串成一条可控流水线条件分支、循环执行、并行执行、路由与提前终止均为一等公民。本文以内容流水线为例演示如何逐层加装六种控制流。心智模型 把流程想象成一条传送带工作流编排的入口是Workflow类它的steps参数接收一列步骤。整条流水线可以想象成传送带输入从带首进入每个步骤从上游拿到结果、做完自己的加工再往下传。步骤不局限于 Agent 和 Team——普通 Python 函数、嵌套的子工作流都能直接排进队列。函数步骤接收StepInput内含input原始输入和previous_step_content上一步产出返回StepOutput上下游衔接就这样自然完成了。官方示例 sequence_of_steps.py 搭了一条研究 → 内容规划 → 写作的直线流水线同时演示了同步、异步与流式四种运行方式。最小形态如下workflow Workflow( nameblog-post, steps[ Step(nameresearch, agentresearcher), Step(namewrite, agentwriter), ], ) workflow.print_response(inputAI trends in 2026)后续所有控制流组件本质都是在直线上增加分叉、回卷、汇合的弯折能力解决什么问题对应组件顺序串联固定流程按序执行steps列表门禁校验二值规则决定某段是否跳过Condition多路分流按内容选择一条或多条路径Router重试回卷单次产出不稳定需重复直到达标Loop并发采集多个独立数据源串行等待浪费时间Parallel全局熔断发现风险时终止整条流水线stopTrue输出本节要点接受传送带这个隐喻后读后面所有组件时只需回答一个问题——它让传送带在哪里弯折。给流水线加判断力Condition、Router 与 CEL 三件套 ⚖️最简单的判断是二值门禁。Condition组件带一个evaluator评估函数返回真才执行内部步骤。condition_basic.py 的做法是研究 → 摘要 → 门禁 → 写文章评估函数扫描摘要里是否出现research indicatespercent等事实性措辞命中才插入事实核查 Agent避免无谓的模型调用Condition( namefact_check_gate, evaluatorneeds_fact_checking, # 读取 previous_step_content 判断 steps[fact_check_step], )需要兜底分支时用else_steps补上否则路径见 condition_with_else.py。判断从二值扩展到多路时换用Router声明一组choices由selector函数依据上下文动态返回要执行的步骤可一条也可多条。router_basic.py 中主题含 tech、AI 等关键词时路由到 HackerNews 研究员否则走通用网页研究员两路汇合后统一进入发布步骤。同目录还有嵌套路由、字符串选择器、媒体管线等变体。若规则足够简单可以连函数都不写。Agno 支持 CELCommon Expression Language一种声明式表达式语言evaluator直接传字符串即可例如evaluatorinput.contains(urgent)输入命中urgent走加急 Agent否则走常规 Agent样例见 cel_basic.py需pip install cel-python。CEL 还能引用previous_step、session_state等上下文对象Loop的end_condition同样适用——07_cel_expressions/loop/ 里的cel_iteration_limit.py就用current_iteration 2一个字符串完成了退出判断。本节要点Condition 管走不走Router 管走哪条CEL 管少写代码——三者可叠加在同一层。给流水线加并发力Loop 设上限与底线Parallel 摊开吞吐 ⚡先说为什么需要重复。大模型的单次输出并不可靠同一研究任务跑一遍可能只吐出一段空泛的摘要。Loop的设计思路不是不达标就重试而是给两个约束——end_condition给质量设上限max_iterations给失败设底线。loop_basic.py 把研究 HackerNews 研究网页两个步骤包进Loop(max_iterations3)评估函数检查任一输出是否超过 200 字符够厚实即提前退出进入创作三次都不达标则带有限产出退出既不会死循环也不会把半成品当成品往下传。若希望每轮结果累加而非覆盖参考loop_iterative_accumulation.py。再说为什么需要并行。两个研究任务互不依赖时串行等待纯属浪费。Parallel把所有分支同时发起全部完成后自动合并结果再进下一步。parallel_basic.py 的结构是并行研究阶段HN 网页→ 写作 → 审阅多源采集场景下整体耗时约减半steps[ Loop(steps[hn_step, web_step], end_conditionresearch_evaluator, max_iterations3), Parallel(hn_step, web_step, nameresearch phase), write_step, ],想给并行分支再套一层判断见 parallel_with_condition.py。本节要点Loop 的max_iterations是预算护栏而非装饰Parallel 的前提是分支间无共享写依赖。安全阀 ️提前终止与状态持久化Condition只能跳过一段真正的事故需要熔断整条流水线。步骤函数返回StepOutput(stopTrue)时引擎跳过所有剩余步骤并将运行标记为提前终止。early_stop_basic.py 里的部署管线很典型安全扫描返回 VULNERABLE门禁函数发出停止信号部署与监控配置一步都不会执行数据校验、内容质检管线也是同一模式。状态方面给Workflow挂上dbSqliteDb(session_table...)即可持久化会话支持跨运行续接session_state字典适合放跨运行的标志位例如 state_in_condition.py 里只对首次出现的用户打招呼就靠它实现。需要人工介入的环节步骤确认、用户输入、错误重试审批08_human_in_the_loop/ 目录按确认、决策树、超时等场景组织了一整套示例。本节要点stop 信号优于异常——它让提前结束成为可预期的正常路径而不是错误路径。组合技巧与避坑 组件组合的自由度比表面更高但有几个易踩的坑嵌套是双向的。Workflow 本身可以作为步骤排进另一个 Workflow见 06_advanced_concepts/ 下的 nested_workflow 系列Condition内部也能装Parallelcondition_with_parallel.py。嵌套层级建议不超过两层否则事件流难以追踪。Loop 必须带底线。只写end_condition不写max_iterations一旦评估函数永远返回假就会无限消耗 token。CEL 有隐式依赖。表达式写法依赖 cel-python 包引擎通过CEL_AVAILABLE检测未安装时直接退出在 CI 环境里要提前声明该依赖。并行分支的写竞争。多个 Parallel 分支同时读写session_state存在先后不可控写依赖放在 Parallel 合并点之后最稳妥。门禁函数的返回值要显式。通过路径也建议显式传stopFalse让继续和熔断在代码里都可见。本节要点控制流组合的边界不在引擎而在事件流的可读性——弯折越多越要留日志。动手顺序建议与源码入口按依赖关系排序四步走效率最高先跑通 sequence_of_steps.py需配置模型 API 密钥亲手改一处previous_step_content的用法理解步骤间数据怎么流。在流水线上加一个Condition门禁评估函数从关键词匹配起步再试else_steps。把串行研究段替换为Parallel或Loop对比两种改造在耗时与 token 上的差异。需要声明式规则时引入 CEL需要持久化与人工确认时再上数据库和 HITL。引擎实现集中在 libs/agno/agno/workflow/ 目录主循环在 workflow.py各控制流组件分别对应condition.py、loop.py、parallel.py、router.py、cel.py排查控制流行为时按文件名定位即可。收尾要点Step 打底判断、路由、循环、并行逐层加装CEL 压缩规则代码stop 信号兜底——六种控制流装齐多智能体流水线就有了生产级的骨架。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考