
Agno Workflow 智能体工作流编排7 个真实场景从零搭出多智能体流水线【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agnoAgno 是一款 Python 多智能体开发框架内置的 Workflow 引擎就是智能体工作流编排的核心把多个 Agent、Team 和普通函数串成一条可控的多智能体流水线支持顺序步骤、条件分支、循环、并行、智能路由与提前终止。本文面向零基础新手按你搭流水线时真实会碰到的 7 个问题逐一拆解读完你能独立搭出第一条多智能体编排链路。把编排想象成闯关游戏每个步骤是一个关卡闯过去才解锁下一关条件是进门前的钥匙检查循环是反复挑战同一关并行是双屏同开两关路由是挑图选关提前终止则是直接退出副本。Agno Workflow 把这些规则做成了现成组件Step、Condition、Loop、Parallel、Router下面按问题逐个领。 多个 Agent 想按顺序接力第一步该写什么搭新流水线的第一个需求几乎都是先 A 后 BB 完再 C。这时只需要最基础的 Step 组件它代表工作流里的一个关卡可以放一个 Agent、一个 Team也可以直接放一个 Python 函数闯完本关自动把结果交给下一关。函数参与接力时有个约定它读取 StepInput——这是每关拿到的标准输入包里面装着上一关的输出——再返回 StepOutput。上关查资料、下关写文章的衔接就是这么来的。示例把研究 Team 和两个 Agent 依次接成一条接力线sequence_of_steps.py。Workflow 入口同时支持同步/异步、一次性输出/流式输出还能接 SQLite 存会话让流水线状态可持久化。 怎么让核查步骤只在真正需要时才触发事实核查 Agent 每次都跑太费钱每次都跳过又不放心。引入 Condition 组件它持有一个 evaluator——就是拿当前上下文算一下、返回 True/False 的评估函数——返回 True 才进入内部步骤False 直接跳过相当于关卡门口的钥匙检查没钥匙不进门。示例里流程是研究 → 摘要 → 核查 → 写作只有摘要出现数据显示百分比这类事实性表述时核查关才会被激活condition_basic.py。 输出质量不达标怎么让流程自己重试一次跑不完好输出长度或质量经常不达标。把一组步骤包进 Loop 组件每打完一轮由 end_condition 这个退出评估函数验收判定够了才放行到下一关再配上 max_iterations 限定最多重开次数防止无限刷关。示例把研究 HackerNews 研究网页两关包进Loop(max_iterations3)研究内容达到足够长才退出循环去创作loop_basic.py。⚡ 两路研究互不依赖能同时跑省时间吗两个互不依赖的研究任务串行排就得等第一路跑完才开第二路。换成 Parallel 组件多路关卡同时发起全部打完后自动合并Merge结果再放行到下一步等于双屏同时开两关。示例非常直观Parallel(研究HackerNews, 研究网页)并行跑完 → 写作 → 审阅多源采集类任务的总耗时被明显压短parallel_basic.py。 同一入口怎么按输入内容自动分给不同 AgentCondition 只有过/不过两种结果但真实业务常常要四选一甚至多选。这时上 Router 组件给它一组 choices候选关卡再由 selector 选择函数根据输入动态决定走哪条路甚至可以一次选多条相当于按地图提示挑关卡。示例里 selector 检查主题是否命中 tech、AI、startup 等关键词——命中走 HackerNews 研究员否则走通用网页研究员最后统一汇入发布步骤router_basic.py。 判断条件能不能不写函数、直接一行字符串搞定规则化的简单判断每次专门写个 evaluator 函数显得笨重。Agno 还支持 CELCommon Expression Language表达式——把判断规则直接写成一行字符串由引擎求值不用手写函数。比如紧急请求走紧急处理 Agent只需一行evaluatorinput.contains(urgent)。CEL 还能引用 input、上一步输出、session_state整条流程共享的状态字典让条件、循环、路由三个控制点全部变成声明式配置。前提是装一下依赖pip install cel-python。示例演示了紧急/普通请求的一行式分流cel_basic.py。 跑到一半发现输入不合规怎么立刻掐断整条链局部跳过之外还需要全局急停输入明显不合规、中间结果已失败时继续跑完剩余关卡纯属浪费模型调用。让该步骤返回StepOutput(stopTrue)工作流就会标记为 Early Stopped跳过所有剩余步骤直接结束——相当于触发机关后整个副本直接结算退出。示例是安全部署链安全扫描 Agent 判定有漏洞时提前掐断后面的部署、监控关卡一步都不执行early_stop_basic.py。 由浅入深的动手路径想真正上手建议按这条路线走先跑通一条直线接力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 读。把关卡串成链把判断写成规则智能体流水线就通了。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考