Agno Workflow 工作流编排完全指南如何把多智能体编排成可控的 Python 自动化流水线【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno想象这样一个场景你要做一个内容生产系统——用户丢进一个选题系统先去 HackerNews 和网页两路找素材再摘要、再写初稿最后发布。如果每一步都交给一个自由发挥的智能体结果往往不可复现有时漏了核查有时研究没够深就急着动笔。Agno Workflow 工作流编排解决的就是这个问题把每个环节固化为步骤Step用Workflow把它们串成一条有入口、有出口、每步可观测的流水线而不是让一群 Agent自己商量着办。 Step 与 Workflow先建立流水线的心智模型把Workflow想象成工厂车间Step就是工位。每个工位上可以站三种工人一个Agent、一个Team多智能体团队或者一个普通的 Python 函数——函数步骤接收StepInput、返回StepOutput逻辑完全由你掌控。引擎源码在 libs/agno/agno/workflow/workflow.py各步骤类型定义在 libs/agno/agno/workflow/step.py。步骤之间天然可传递上下文任何一步都能读到previous_step_content上一步的产出和前序所有步骤的输出。说白了上一步查资料、下一步写文章这种衔接不需要你手动搬运数据流水线自带传送带。这里有个关键设计Workflow同时支持同步/异步执行和一次性/流式输出还能挂上 SQLite 等存储做会话持久化。也就是说你搭的不只是一段脚本而是一个可以中断恢复、可追溯的生产级执行单元。 控制流工具箱条件、循环、并行分别管什么顺序流水线只能处理直线任务真实业务里总有三类岔路Agno Workflow 工作流编排给了三个对应手法条件分支能不能跳过某步Condition带一个evaluator评估函数返回True才执行内部步骤否则直接跳过还支持else_steps走否则分支。典型场景摘要里出现数据显示研究证明这类含数字断言的表述时才插入事实核查 Agent 这一步纯观点类内容直接进写作环节。示例见 condition_basic.py带否则分支的写法在 condition_with_else.py。循环质量不达标就返工。Loop把一组步骤包起来反复执行由end_condition评估函数判定何时收手max_iterations兜底防死循环。比如研究 HackerNews 研究网页两步跑一轮输出不足 200 字符就再来一轮最多三轮。见 loop_basic.py 和累积式写法 loop_iterative_accumulation.py。并行互不依赖就别串行等。Parallel同时发起多条路径全部完成后自动合并结果再进下一步。两路素材采集并行跑整体耗时约等于最慢的那路而非两路之和。见 parallel_basic.py。组件解决的场景关键参数Condition某步可有可无evaluator/else_stepsLoop结果不达标需重试end_condition/max_iterationsParallel多路任务无依赖自动 Merge 汇总Router多路选其一或数路choices/selector 经验提示条件和循环都可以嵌套进Parallel的某条支路里先并行分叉、支路内再按条件收敛是处理多源采集 异质后处理的常用组合参考 condition_with_parallel.py。 智能决策层Router 多路路由与 CEL 声明式规则前面讲的都是是/否但选题分发常常是多路选择科技话题走技术研究员创业话题走商业研究员其余走通用网页研究员。Router组件吃一组choices由selector函数根据上下文动态挑路甚至可以一次选中多条。router_basic.py 里 selector 扫描主题是否含 tech、AI、startup 等关键词来定路由嵌套路由、字符串选择器等变体都在 05_conditional_branching/ 目录下。这里有个省代码的利器CEL 表达式Common Expression Language。evaluator和selector不必写 Python 函数直接传一行字符串就行。比如判断紧急请求走加急通道evaluatorinput.contains(urgent)CEL 表达式能引用input、previous_step、session_state等上下文对象意味着条件、循环退出、路由选择三个决策点都能声明式配置——业务规则变了改字符串即可不用动代码。完整示例分三个子目录07_cel_expressions/condition/、loop/、router/需pip install cel-python。 安全与容错中间一步炸了怎么办流水线最怕的不是某一步失败而是失败后还在闷头跑完剩下所有步骤——白烧 Token甚至发布烂结果。Agno Workflow 提供StepStop()做全局提前终止任一步骤触发它剩余步骤直接跳过整个工作流以 Early Stopped 状态收尾。典型触发时机输入明显不合规、中间结果已不可用、用户主动取消。四种触发位置的对照示例都在 06_advanced_concepts/early_stopping/ 目录基础触发 early_stop_basic.py、条件内触发 early_stop_condition.py、循环内触发 early_stop_loop.py、并行支路内触发 early_stop_parallel.py。生产部署时还建议配套三件东西Guardrails在步骤入口做提示注入等安全拦截见 06_advanced_concepts/guardrails/run_control目录下的取消运行、事件存储能力应对长任务治理长任务场景则用 background_execution/ 和 long_running/ 里的后台执行与断点续跑方案。 实践路线从跑通第一条流水线到生产级能力按下面顺序走每一步都有可运行的官方示例全部位于 cookbook/04_workflows/顺序执行01_basic_workflows/ —— 从 sequence_of_steps.py 跑通研究 → 规划 → 写作直线流水线再看 step_with_function.py 学会用普通函数当步骤条件分支02_conditional_execution/ —— 掌握Condition与else_steps循环执行03_loop_execution/ —— 学会用end_conditionmax_iterations做质量闸门并行执行04_parallel_execution/ —— 搭多源采集的并行段路由与分支05_conditional_branching/ ——Router多路分发与嵌套路由CEL 表达式07_cel_expressions/ —— 把分支规则从代码降级为配置进阶能力06_advanced_concepts/ —— 会话状态、结构化输入输出、嵌套工作流workflow_as_a_step/人工审批看 08_human_in_the_loop/想深挖实现细节引擎核心模块都在 libs/agno/agno/workflow/ 目录workflow.py是执行引擎condition.py、loop.py、parallel.py、router.py各对应一种控制流cel.py处理表达式解析。一句话收尾把任务想成车间排产——工位Step排好工序岔路Condition/Loop/Parallel/Router定好策略规则CEL写在纸上而非焊死在机器里再给每条线装一个急停按钮StepStop智能体流水线就从演示代码变成了能上生产的系统。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Agno Workflow 工作流编排完全指南：如何把多智能体编排成可控的 Python 自动化流水线
Agno Workflow 工作流编排完全指南如何把多智能体编排成可控的 Python 自动化流水线【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno想象这样一个场景你要做一个内容生产系统——用户丢进一个选题系统先去 HackerNews 和网页两路找素材再摘要、再写初稿最后发布。如果每一步都交给一个自由发挥的智能体结果往往不可复现有时漏了核查有时研究没够深就急着动笔。Agno Workflow 工作流编排解决的就是这个问题把每个环节固化为步骤Step用Workflow把它们串成一条有入口、有出口、每步可观测的流水线而不是让一群 Agent自己商量着办。 Step 与 Workflow先建立流水线的心智模型把Workflow想象成工厂车间Step就是工位。每个工位上可以站三种工人一个Agent、一个Team多智能体团队或者一个普通的 Python 函数——函数步骤接收StepInput、返回StepOutput逻辑完全由你掌控。引擎源码在 libs/agno/agno/workflow/workflow.py各步骤类型定义在 libs/agno/agno/workflow/step.py。步骤之间天然可传递上下文任何一步都能读到previous_step_content上一步的产出和前序所有步骤的输出。说白了上一步查资料、下一步写文章这种衔接不需要你手动搬运数据流水线自带传送带。这里有个关键设计Workflow同时支持同步/异步执行和一次性/流式输出还能挂上 SQLite 等存储做会话持久化。也就是说你搭的不只是一段脚本而是一个可以中断恢复、可追溯的生产级执行单元。 控制流工具箱条件、循环、并行分别管什么顺序流水线只能处理直线任务真实业务里总有三类岔路Agno Workflow 工作流编排给了三个对应手法条件分支能不能跳过某步Condition带一个evaluator评估函数返回True才执行内部步骤否则直接跳过还支持else_steps走否则分支。典型场景摘要里出现数据显示研究证明这类含数字断言的表述时才插入事实核查 Agent 这一步纯观点类内容直接进写作环节。示例见 condition_basic.py带否则分支的写法在 condition_with_else.py。循环质量不达标就返工。Loop把一组步骤包起来反复执行由end_condition评估函数判定何时收手max_iterations兜底防死循环。比如研究 HackerNews 研究网页两步跑一轮输出不足 200 字符就再来一轮最多三轮。见 loop_basic.py 和累积式写法 loop_iterative_accumulation.py。并行互不依赖就别串行等。Parallel同时发起多条路径全部完成后自动合并结果再进下一步。两路素材采集并行跑整体耗时约等于最慢的那路而非两路之和。见 parallel_basic.py。组件解决的场景关键参数Condition某步可有可无evaluator/else_stepsLoop结果不达标需重试end_condition/max_iterationsParallel多路任务无依赖自动 Merge 汇总Router多路选其一或数路choices/selector 经验提示条件和循环都可以嵌套进Parallel的某条支路里先并行分叉、支路内再按条件收敛是处理多源采集 异质后处理的常用组合参考 condition_with_parallel.py。 智能决策层Router 多路路由与 CEL 声明式规则前面讲的都是是/否但选题分发常常是多路选择科技话题走技术研究员创业话题走商业研究员其余走通用网页研究员。Router组件吃一组choices由selector函数根据上下文动态挑路甚至可以一次选中多条。router_basic.py 里 selector 扫描主题是否含 tech、AI、startup 等关键词来定路由嵌套路由、字符串选择器等变体都在 05_conditional_branching/ 目录下。这里有个省代码的利器CEL 表达式Common Expression Language。evaluator和selector不必写 Python 函数直接传一行字符串就行。比如判断紧急请求走加急通道evaluatorinput.contains(urgent)CEL 表达式能引用input、previous_step、session_state等上下文对象意味着条件、循环退出、路由选择三个决策点都能声明式配置——业务规则变了改字符串即可不用动代码。完整示例分三个子目录07_cel_expressions/condition/、loop/、router/需pip install cel-python。 安全与容错中间一步炸了怎么办流水线最怕的不是某一步失败而是失败后还在闷头跑完剩下所有步骤——白烧 Token甚至发布烂结果。Agno Workflow 提供StepStop()做全局提前终止任一步骤触发它剩余步骤直接跳过整个工作流以 Early Stopped 状态收尾。典型触发时机输入明显不合规、中间结果已不可用、用户主动取消。四种触发位置的对照示例都在 06_advanced_concepts/early_stopping/ 目录基础触发 early_stop_basic.py、条件内触发 early_stop_condition.py、循环内触发 early_stop_loop.py、并行支路内触发 early_stop_parallel.py。生产部署时还建议配套三件东西Guardrails在步骤入口做提示注入等安全拦截见 06_advanced_concepts/guardrails/run_control目录下的取消运行、事件存储能力应对长任务治理长任务场景则用 background_execution/ 和 long_running/ 里的后台执行与断点续跑方案。 实践路线从跑通第一条流水线到生产级能力按下面顺序走每一步都有可运行的官方示例全部位于 cookbook/04_workflows/顺序执行01_basic_workflows/ —— 从 sequence_of_steps.py 跑通研究 → 规划 → 写作直线流水线再看 step_with_function.py 学会用普通函数当步骤条件分支02_conditional_execution/ —— 掌握Condition与else_steps循环执行03_loop_execution/ —— 学会用end_conditionmax_iterations做质量闸门并行执行04_parallel_execution/ —— 搭多源采集的并行段路由与分支05_conditional_branching/ ——Router多路分发与嵌套路由CEL 表达式07_cel_expressions/ —— 把分支规则从代码降级为配置进阶能力06_advanced_concepts/ —— 会话状态、结构化输入输出、嵌套工作流workflow_as_a_step/人工审批看 08_human_in_the_loop/想深挖实现细节引擎核心模块都在 libs/agno/agno/workflow/ 目录workflow.py是执行引擎condition.py、loop.py、parallel.py、router.py各对应一种控制流cel.py处理表达式解析。一句话收尾把任务想成车间排产——工位Step排好工序岔路Condition/Loop/Parallel/Router定好策略规则CEL写在纸上而非焊死在机器里再给每条线装一个急停按钮StepStop智能体流水线就从演示代码变成了能上生产的系统。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考