Agno Workflow 完整指南：4 个控制点让流水线又快又可控

Agno Workflow 完整指南4 个控制点让流水线又快又可控【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno把研究、成稿、核查三个 Agent 串成一条线只需几行代码难的是后面两个独立的数据源为什么要排队等摘要里没有事实性表述时核查环节为什么要照跑不误输入明显不合规时整条线为什么要跑完才收场Agno Workflow 就是为这些控制点准备的编排引擎先跑通直线流程再依次装上门禁、重试、并行、路由和全局叫停。数据如何在 Step 之间流动先搭骨架。一个 Step 是一次完整作业必须指定且只指定一个执行器Agent、Team、普通函数或嵌套的 Workflow。函数式步骤的签名是接收 StepInput、返回 StepOutput天然能嵌进流水线。Workflow 按列表顺序执行每步的输出自动成为下一步的输入。关键是 StepInput 里两个字段定义在 libs/agno/agno/workflow/types.pyprevious_step_content是上一步的内容previous_step_outputs是按步骤名索引的全量输出。拿到它之后前一步查资料、后一步写文章这种衔接不需要任何手动搬运。嵌套容器里的子步骤输出也能通过get_step_output(step_name)递归查到并行分支的各路输出取出来是一个按名映射的字典。workflow Workflow( steps[research_step, outline_step, writing_step] )如何给关键步骤加质量门禁不解决会怎样每份摘要都过一遍事实核查既多花钱又拖慢整条线。Condition 就是干这个的门禁它带一个evaluator返回 bool 的函数和一个steps列表为真才执行否则整段跳过再配上else_steps就能走否则分支。evaluator收到的正是 StepInput典型做法是检查previous_step_content里有没有出现数据显示83%这类事实性关键词命中才放行核查 Agent。门禁内部照样可以再套 Loop 或 Parallel条件里做并行并不冲突。gate Condition( evaluatorlambda si: 数据显示 in str(si.previous_step_content or ), steps[fact_check_step], else_steps[light_review_step], )最小示例见 cookbook/04_workflows/02_conditional_execution/condition_basic.py否则分支的写法在 condition_with_else.py。循环重试怎么写才不会跑不完大模型一次产出的质量波动大不达标就重跑几乎是刚需。Loop 把一组步骤包起来反复执行每轮结束调用end_condition评估函数接收本轮全部 StepOutput 列表返回 bool返回真才退出max_iterations兜底防死循环缺省就是 3。这里最容易踩坑的是给 end_condition 写完美标准——模型很难稳定达到结果每次都烧满 3 轮。把门槛降到信息量够了就过更实际。想让下一轮利用上一轮的结果做增量改进把forward_iteration_output置为 True上一轮输出会转发成下一轮输入默认 False每轮都拿最初的 StepInput。research_loop Loop( steps[hn_step, web_step], max_iterations3, end_conditionresearch_evaluator, )cookbook/04_workflows/03_loop_execution/ 里loop_basic.py检查本轮输出长度loop_iterative_accumulation.py演示累积式改进实现细节在 libs/agno/agno/workflow/loop.py。并行分支如何省掉串行等待两个数据源互相独立为什么要串行串行的总耗时是各路径之和Parallel 让多条路径同时开跑全部完成后自动把各路 StepOutput 合并成交给下游的一份结果——你不用写任何归并代码。它支持位置参数写法第一个位置参数还能传名字Parallel(采集, step_a, step_b)。它和 Condition 可以嵌套并行分支内部再按条件分流也不违和。多源采集类任务的总耗时约等于最慢那条路径而不是累加。research Parallel( collect, Step(nameHN 研究, agenthn_researcher), Step(name网页研究, agentweb_researcher), )cookbook/04_workflows/04_parallel_execution/parallel_basic.py 把查 HackerNews 查网页丢进 Parallel 后接写作和审阅parallel_with_condition.py 展示分支内部的条件判断。Router 多路分发怎么选路Condition 只够二选一真实业务常是命中科技关键词走 HN 研究员其余走通用研究员这种多路分发。Router 接收一组choices候选步骤和一个selector函数或 CEL 字符串根据 StepInput 动态决定执行哪条路一次选中多条也合法。router_basic.py的 selector 检查主题是否包含 tech、AI、startup 等词命中路由到 HackerNews 研究员否则走通用网页研究员两条路最后汇入发布步骤。规则化的选路可以直接写 CEL 字符串返回步骤名拿不准的路径可以设requires_user_inputTrue工作流会暂停让人来挑。嵌套路由、字符串选择器、媒体管线等变体集中在 cookbook/04_workflows/05_conditional_branching/。CEL 表达式与提前终止声明式规则和全局叫停evaluator和selector都能写 Python 函数但规则化判断写函数太重。装上cel-python后条件、循环、路由三个控制点可以统一用一行 CEL 字符串配置Condition 的 evaluator 给input.contains(urgent)输入含 urgent 就走紧急处理Loop 的end_condition能用current_iteration、max_iterations、all_success、last_step_content这些上下文写退出规则例如all_success current_iteration 2Router 的 selector 直接返回步骤名。三个点共用的上下文有input、previous_step_content、previous_step_outputs、additional_data、session_stateRouter 里另有step_choices。提前终止是另一层控制。局部跳过是 Condition 的事全局叫停靠stopTrue任何步骤返回带stopTrue的 StepOutput剩余步骤直接跳过整条工作流以 Early Stopped 状态结束。early_stop_basic.py的部署管线里安全扫描之后一个函数步骤检查previous_step_content发现 VULNERABLE 就终止return StepOutput(content[ALERT] Deployment blocked, stopTrue)典型场景输入明显不合规、中间结果已失败、用户取消。在 Condition、Loop、Parallel 内部触发停止的写法都在 cookbook/04_workflows/06_advanced_concepts/ 的early_stopping/子目录。上手路径先跑 cookbook/04_workflows/01_basic_workflows/确认直线流水线能读到previous_step_content再依次过 02 条件分支、03 循环、04 并行三个目录给管线装上门禁、重试和并行遇到多路分发翻 cookbook/04_workflows/05_conditional_branching/规则化判断多就直接上 cookbook/04_workflows/07_cel_expressions/最后读 06_advanced_concepts 的early_stopping/给整条线加全局叫停想抠执行细节按 libs/agno/agno/workflow/workflow.py、condition.py、loop.py、parallel.py、router.py 的顺序逐文件读。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考