Agno Workflow 工作流编排6 个控制点让多智能体流水线跑起来【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno你让一个 Agent 一口气调研 写作 审核结果时好时坏自己写 if/else 去补代码越补越乱。Agno 的 Workflow 引擎解决的就是这个问题用代码把 Agent、Team 和普通 Python 函数排成一条流水线并且随时控制它往哪走、走几圈、什么时候停。 心智模型先记住一条产线把整条 Workflow 想成一条工厂产线。物料输入数据从一头进经过一个个工位步骤从另一头产出成品。区别只在于有的工位带质检台不达标就退回返工区有的路段拆成几条并行通道同时开工有的路口有分拣员决定走哪条道产线末端还有一枚紧急停止按钮。工位之间靠传送带交接物料——前一个工位产出的内容后一个工位天然拿得到。这套引擎的全部能力都建立在步骤 传送带交接这个骨架上控制流只是给产线加质检台、分叉口和按钮。官方示例集中在 cookbook/04_workflows/ 目录。 串联多角色内容流水线让数据自己往后流你要搭调研 → 规划 → 写作这类多角色协作流程又不想在每两个角色之间手写数据搬运代码。每个步骤可以是 Agent、Team也可以直接是一个函数。函数步骤签名只接收一个StepInput返回StepOutput而StepInput里带着上一步的内容所以上一步查到的资料会自动出现在下一步的视野里。比如博客流水线一个 Team 步骤HackerNews 研究员 网页搜索员先做调研接着规划 Agent 排内容日历最后写作 Agent 出稿中间没有任何手动传参。完整代码在 sequence_of_steps 示例函数步骤的写法看 02_step_with_function 目录。 给流水线加质量门禁只在需要时插入额外检查你不想让每篇稿子都白白跑一遍事实核查只想在摘要里出现了数据、百分比、研究结论这类表述时才核查。这就是条件分支要解决的事给Condition配一个evaluator评估函数返回 True 才执行里面的步骤否则整段跳过。示例 condition_basic.py 里的门禁规则就是摘要含事实性关键词才放行给核查 Agent。判断不成立时你想走另一条路用else_steps即可写法见 condition_with_else.py。如果规则本身很机械——比如输入含 urgent 就走加急通道——其实可以不写任何函数改用 CEL 声明式表达式Condition( evaluatorinput.contains(urgent), steps[urgent_step], )表达式里还能引用previous_step、session_state等上下文一行字符串替代整个判断函数。更多写法在 07_cel_expressions 目录需pip install cel-python。 结果不稳就自动重试多源数据并发采集你发现模型调研的输出时长时短单次结果不可靠或者两个数据源串行跑太慢。对前者Loop把一组步骤包起来反复执行end_condition评估函数说了算它返回 True 才放行。示例 loop_basic.py 里调研 HackerNews 调研网页被包进一个Loop(max_iterations3)退出条件检查产出内容是否超过 200 字符。两者关系其实很清晰end_condition是正常出口max_iterations是防死循环的保险丝出口坏了保险丝兜底。对后者Parallel把互不依赖的步骤同时发起全部完成后自动把各路结果合并再喂给下一步——parallel_basic.py 就是两路调研并发 → 合并 → 写作 → 审阅的标准结构。 按内容智能分流出问题时提前叫停你的调研入口会收到各种主题科技、创业、通用话题各该交给不同专家同时你也不希望一条明显违规的输入把后续所有大模型调用都烧掉。分流交给Router给它一组choices每条路对应一个步骤再由selector函数看上下文动态挑路。router_basic.py 的 selector 检查主题关键词含 tech/AI/startup 就发给 HackerNews 研究员否则走通用网页研究员最后汇回发布步骤。叫停则是全局级的任何步骤返回StepOutput(stopTrue)整条流水线立即截断后面的部署、监控步骤全部跳过。early_stopping 目录 里的示例是安全门禁扫描 Agent 报出 VULNERABLE 时——return StepOutput(content发现严重漏洞终止部署, stopTrue)部署与监控 Agent 一次都不会被调用直接省掉对应的模型开销。说白了提前终止省的不是几行代码是真金白银的 token。 快速上手按依赖顺序过一遍 cookbook你不需要通读全部示例按依赖关系排的阅读顺序效率最高01_basic_workflows先跑通直线流水线把步骤间怎么传数据吃透02_conditional_execution条件分支与 else 分支03_loop_execution循环与退出条件04_parallel_execution并行执行与结果合并05_conditional_branchingRouter 路由与嵌套分支07_cel_expressions把判断改写成声明式表达式再往上进阶能力都集中在两个目录06_advanced_concepts 里有历史上下文、会话状态session_state、结构化输入输出、嵌套子工作流等子目录08_human_in_the_loop 演示人工审批如何插进流水线。️ 下一步把第一条流水线跑起来现在就打开condition_basic.py配好模型 API key输入一段带数据显示的摘要和不带事实表述的摘要各跑一次观察门禁是否在两种情况下给出不同行为——这是理解条件分支最快的一分钟。想深入实现细节引擎源码在 libs/agno/agno/workflow/ 目录workflow.py、condition.py、loop.py、parallel.py、router.py五个文件各对应一节里讲的一个控制点04_workflows 的 README 则汇总了所有示例的索引。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考