Agno Workflow 工作流编排5 个控制流让多智能体流水线快速跑起来【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agnoAgno Workflow 是多智能体框架 Agno 内置的工作流编排引擎几行代码就能把 Agent、Team 和普通 Python 函数接成一条可控流水线分支、重试、并行顺序全部由你决定。为什么多智能体流水线需要工作流引擎多个 Agent 协作时你大概率会遇到这几个局面步骤顺序写死在主流程里想加一道质检只能改代码重写不知道从哪里切入。某个步骤的结果不达标只能人工盯着重跑没有不达标就再来一次的机制。三个数据源采集排成一条线后一个必须等前两个干完。中间结果没人把关坏数据一路传到最后一步才被发现。解法是控制流 步骤化把任务拆成步骤Step 是执行单元可以是 Agent、Team也可以是一个普通函数再用条件、循环、并行、路由这些控制流规定每个步骤何时跑、跑几遍、结果往哪走。Agno 的工作流引擎把这套编排工具内置好了。先跑通一条最小流水线让每一步把数据传给下一步结论先行Workflow只要一个步骤列表就按顺序执行每一步天然能拿到上一步的输出。def draft_reply(step_input: StepInput) - StepOutput: ticket step_input.input # 最初输入 analysis step_input.previous_step_content # 上一步输出 return StepOutput(contentf{ticket}: {analysis}) workflow Workflow( nameTicket Pipeline, steps[classify_step, analyze_step, draft_reply], )数据流很直白StepInput是进料单input是最初的输入previous_step_content是上一步的结果StepOutput是出料单它的content会自动变成下一步的previous_step_content。上一步做分析、下一步写回复的衔接就靠它。流水线支持同步、异步加streamTrue还能流式输出也可以接 SqliteDb 持久化会话完整玩法见 sequence_of_steps.py 和 workflow 源码。按决策类型学控制流要不要做、做到什么程度、走哪条路每种控制流其实对应一类决策。先看你要决定什么再选组件。要不要做用 Condition 加一道条件门禁适用有些步骤只在特定情况下才该执行。Condition包住若干步骤门禁由evaluator函数决定——它返回True才执行内部步骤否则整段跳过配一个else_steps参数还能补上不满足时改做这个。def needs_escalation(step_input: StepInput) - bool: return angry in (step_input.previous_step_content or ).lower() Condition( nameescalation_gate, evaluatorneeds_escalation, steps[escalate_step], )业务例子客服工单处理。情绪分析步骤之后上一级输出里出现angry才执行转人工步骤否则直接跳到发送回复草稿。门禁不开这一次大模型调用就省了。完整示例可参考 condition_basic.py。常见坑evaluator里只看previous_step_content不看input。如果你的判断依据是最初的用户输入读错对象就会在错误的步骤上做出判断。做到什么程度才算完用 Loop 质检复称适用结果质量无法一次保证需要反复执行到达标为止。Loop包住一组步骤每轮结束后由end_condition评估函数复称一次合格就出环不合格再跑一轮。务必给max_iterations兜底防止永远不达标导致死循环。Loop( nameclean_loop, steps[clean_step], end_conditionquality_check, # 质检合格返回 True max_iterations3, )业务例子数据清洗流水线。清洗步骤处理脏数据质检函数检查缺值和格式不合格就回到清洗步骤重来最多 3 次后强制进入入库。这个思路来自 loop_basic.py。常见坑end_condition只判断有没有输出不判断输出是否达标——第一轮就通过循环等于摆设。走哪条路Parallel 并行与 Router 分拣两个子问题任务互不依赖时先做哪些、路径该按内容怎么选。Parallel把多个步骤同时发起全部跑完后自动 Merge 结果再交给下一步Parallel( Step(namecollect_api, agentapi_collector), Step(namecollect_db, agentdb_collector), namecollect_phase, )多源采集场景API 拉取和数据库读取并发执行总耗时取决于最慢一路汇总步骤拿到合并结果。写法见 parallel_basic.py。Router像快递分拣中心choices里摆好若干条路线selector函数按内容决定包裹进哪个格口返回多条就同时执行多条路def route_ticket(step_input: StepInput) - List[Step]: if billing in step_input.input.lower(): return [billing_step] return [general_step] Router(nameticket_router, choices[billing_step, general_step], selectorroute_ticket)工单按主题分流账单类进billing_step其余进general_step。更多路由玩法见 router_basic.py。常见坑把有依赖的任务塞进Parallel。B 需要 A 的结果却和 A 并行B 只会拿到空输入。用 CEL 表达式一行写完条件不写函数结论规则简单时evaluator和selector都不用写函数一行字符串就够。CELCommon Expression Language一种标准表达式语法把判断直接变成声明式配置表达式里能引用input最初输入、previous_step上一步输出、session_state会话状态这些上下文对象条件、循环、路由三种控制点通用。Condition( nameurgent_check, evaluatorinput.contains(urgent), steps[urgent_step], else_steps[normal_step], )场景是工单优先级路由输入含 urgent 走紧急处理 Agent否则走普通处理 Agent一行替代十几行关键词匹配代码。注意需先执行pip install cel-python。更多变体见 cel_basic.py 和 07_cel_expressions 目录。给流水线装一个急停按钮用 StepStop 提前终止结论除了局部跳过整条流水线可以全局叫停。某个步骤触发StepStop()代码中等价于返回stopTrue的StepOutput剩余步骤直接跳过工作流提前结束。def security_gate(step_input: StepInput) - StepOutput: result step_input.previous_step_content or if VULNERABLE in result.upper(): return StepOutput(contentDeploy blocked, stopTrue) return StepOutput(contentOK, stopFalse)典型场景输入明显不合规、中间结果已失败比如安全扫描发现漏洞、用户取消任务。宁可让流水线在门口停下也不要等剩余步骤跑完再回滚。完整示例见 early_stopping 目录。选型速查表、新手误区与学习路径你的场景组件关键参数满足特定条件才执行Conditionevaluator、else_steps反复执行到质量达标Loopend_condition、max_iterations互不依赖的任务同时跑Parallelname按内容选择路线Routerchoices、selector提前终止整条流水线StepStop()stopTrue新手三条常见误区每个条件都写函数——规则能用一句话说清时直接写 CEL 表达式函数留着处理复杂逻辑。Loop 不设 max_iterations——模型总说还不够时没有兜底就是无限循环。把有依赖的任务塞进 Parallel——并行只服务于互不依赖的任务有先后就拆成串行步骤。动手练习按这个顺序来01_basic_workflows直线流水线 →02_conditional_execution门禁 →03_loop_execution重试 →04_parallel_execution并行 →05_conditional_branching分拣都在 cookbook/04_workflows/ 目录下。想理解引擎实现再读 libs/agno/agno/workflow/ 里的condition.py、loop.py、parallel.py、router.py四个文件即可。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/9/4 15:36:07 · 阅读 1.2k Agno Workflow 工作流编排：5 个控制流让多智能体流水线快速跑起来
Agno Workflow 工作流编排5 个控制流让多智能体流水线快速跑起来【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agnoAgno Workflow 是多智能体框架 Agno 内置的工作流编排引擎几行代码就能把 Agent、Team 和普通 Python 函数接成一条可控流水线分支、重试、并行顺序全部由你决定。为什么多智能体流水线需要工作流引擎多个 Agent 协作时你大概率会遇到这几个局面步骤顺序写死在主流程里想加一道质检只能改代码重写不知道从哪里切入。某个步骤的结果不达标只能人工盯着重跑没有不达标就再来一次的机制。三个数据源采集排成一条线后一个必须等前两个干完。中间结果没人把关坏数据一路传到最后一步才被发现。解法是控制流 步骤化把任务拆成步骤Step 是执行单元可以是 Agent、Team也可以是一个普通函数再用条件、循环、并行、路由这些控制流规定每个步骤何时跑、跑几遍、结果往哪走。Agno 的工作流引擎把这套编排工具内置好了。先跑通一条最小流水线让每一步把数据传给下一步结论先行Workflow只要一个步骤列表就按顺序执行每一步天然能拿到上一步的输出。def draft_reply(step_input: StepInput) - StepOutput: ticket step_input.input # 最初输入 analysis step_input.previous_step_content # 上一步输出 return StepOutput(contentf{ticket}: {analysis}) workflow Workflow( nameTicket Pipeline, steps[classify_step, analyze_step, draft_reply], )数据流很直白StepInput是进料单input是最初的输入previous_step_content是上一步的结果StepOutput是出料单它的content会自动变成下一步的previous_step_content。上一步做分析、下一步写回复的衔接就靠它。流水线支持同步、异步加streamTrue还能流式输出也可以接 SqliteDb 持久化会话完整玩法见 sequence_of_steps.py 和 workflow 源码。按决策类型学控制流要不要做、做到什么程度、走哪条路每种控制流其实对应一类决策。先看你要决定什么再选组件。要不要做用 Condition 加一道条件门禁适用有些步骤只在特定情况下才该执行。Condition包住若干步骤门禁由evaluator函数决定——它返回True才执行内部步骤否则整段跳过配一个else_steps参数还能补上不满足时改做这个。def needs_escalation(step_input: StepInput) - bool: return angry in (step_input.previous_step_content or ).lower() Condition( nameescalation_gate, evaluatorneeds_escalation, steps[escalate_step], )业务例子客服工单处理。情绪分析步骤之后上一级输出里出现angry才执行转人工步骤否则直接跳到发送回复草稿。门禁不开这一次大模型调用就省了。完整示例可参考 condition_basic.py。常见坑evaluator里只看previous_step_content不看input。如果你的判断依据是最初的用户输入读错对象就会在错误的步骤上做出判断。做到什么程度才算完用 Loop 质检复称适用结果质量无法一次保证需要反复执行到达标为止。Loop包住一组步骤每轮结束后由end_condition评估函数复称一次合格就出环不合格再跑一轮。务必给max_iterations兜底防止永远不达标导致死循环。Loop( nameclean_loop, steps[clean_step], end_conditionquality_check, # 质检合格返回 True max_iterations3, )业务例子数据清洗流水线。清洗步骤处理脏数据质检函数检查缺值和格式不合格就回到清洗步骤重来最多 3 次后强制进入入库。这个思路来自 loop_basic.py。常见坑end_condition只判断有没有输出不判断输出是否达标——第一轮就通过循环等于摆设。走哪条路Parallel 并行与 Router 分拣两个子问题任务互不依赖时先做哪些、路径该按内容怎么选。Parallel把多个步骤同时发起全部跑完后自动 Merge 结果再交给下一步Parallel( Step(namecollect_api, agentapi_collector), Step(namecollect_db, agentdb_collector), namecollect_phase, )多源采集场景API 拉取和数据库读取并发执行总耗时取决于最慢一路汇总步骤拿到合并结果。写法见 parallel_basic.py。Router像快递分拣中心choices里摆好若干条路线selector函数按内容决定包裹进哪个格口返回多条就同时执行多条路def route_ticket(step_input: StepInput) - List[Step]: if billing in step_input.input.lower(): return [billing_step] return [general_step] Router(nameticket_router, choices[billing_step, general_step], selectorroute_ticket)工单按主题分流账单类进billing_step其余进general_step。更多路由玩法见 router_basic.py。常见坑把有依赖的任务塞进Parallel。B 需要 A 的结果却和 A 并行B 只会拿到空输入。用 CEL 表达式一行写完条件不写函数结论规则简单时evaluator和selector都不用写函数一行字符串就够。CELCommon Expression Language一种标准表达式语法把判断直接变成声明式配置表达式里能引用input最初输入、previous_step上一步输出、session_state会话状态这些上下文对象条件、循环、路由三种控制点通用。Condition( nameurgent_check, evaluatorinput.contains(urgent), steps[urgent_step], else_steps[normal_step], )场景是工单优先级路由输入含 urgent 走紧急处理 Agent否则走普通处理 Agent一行替代十几行关键词匹配代码。注意需先执行pip install cel-python。更多变体见 cel_basic.py 和 07_cel_expressions 目录。给流水线装一个急停按钮用 StepStop 提前终止结论除了局部跳过整条流水线可以全局叫停。某个步骤触发StepStop()代码中等价于返回stopTrue的StepOutput剩余步骤直接跳过工作流提前结束。def security_gate(step_input: StepInput) - StepOutput: result step_input.previous_step_content or if VULNERABLE in result.upper(): return StepOutput(contentDeploy blocked, stopTrue) return StepOutput(contentOK, stopFalse)典型场景输入明显不合规、中间结果已失败比如安全扫描发现漏洞、用户取消任务。宁可让流水线在门口停下也不要等剩余步骤跑完再回滚。完整示例见 early_stopping 目录。选型速查表、新手误区与学习路径你的场景组件关键参数满足特定条件才执行Conditionevaluator、else_steps反复执行到质量达标Loopend_condition、max_iterations互不依赖的任务同时跑Parallelname按内容选择路线Routerchoices、selector提前终止整条流水线StepStop()stopTrue新手三条常见误区每个条件都写函数——规则能用一句话说清时直接写 CEL 表达式函数留着处理复杂逻辑。Loop 不设 max_iterations——模型总说还不够时没有兜底就是无限循环。把有依赖的任务塞进 Parallel——并行只服务于互不依赖的任务有先后就拆成串行步骤。动手练习按这个顺序来01_basic_workflows直线流水线 →02_conditional_execution门禁 →03_loop_execution重试 →04_parallel_execution并行 →05_conditional_branching分拣都在 cookbook/04_workflows/ 目录下。想理解引擎实现再读 libs/agno/agno/workflow/ 里的condition.py、loop.py、parallel.py、router.py四个文件即可。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考