Agno Workflow 实战：智能体流水线编排完整指南

Agno Workflow 实战：智能体流水线编排完整指南

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno

想让多个 Agent 依次完成"研究 → 起草 → 审阅"，传统做法往往是一坨 if-else 大函数，上下文靠全局变量传递，出了错都不知道挂在哪一步。Agno Workflow 编排正是解决这件事的：它把 Agent、Team 和普通 Python 函数统一封装成 Step（步骤），按控制流规则串成一条可控流水线，支持顺序执行、条件分支、循环、并行、路由和提前终止，且可接入 SQLite 等数据库持久化会话状态。

⚡ 多路任务互不依赖，为什么要串行跑？

同时采集多个信息源时，串行等待就是在浪费钱和时间。Parallel 组件把多个步骤同时发起，全部完成后自动合并结果再交给下一步。官方示例 parallel_basic.py 里"研究 HackerNews + 研究网页"并行跑完才进入写作，多源采集类任务的耗时直接减半；它还能和 Condition 组合，实现"命中才并行执行"。

任务间无依赖就交给并行，这是压缩流水线总耗时最直接的抓手。

结果不达标怎么办？给流水线套一个重试环

大模型单次输出质量不稳定，与其靠运气，不如套上 Loop：一组步骤反复执行，直到 end_condition 函数判定"够了"才退出，再用 max_iterations 兜底防死循环。loop_basic.py 把两个研究步骤包进循环，以"输出超过 200 字符"为退出条件，最多试 3 轮后放行给写作 Agent。反过来，如果输入明显不合规或用户取消了任务，可以在步骤里触发 StepStop() 直接 Early Stop，剩余步骤全部跳过，工作流标记为提前终止结束。

Loop 负责"不够好就再来"，StepStop 负责"没救了就收手"，一进一出让流水线不会失控。

分支该怎么走？把判断交给"调度台"

两条路的选择用 Condition：写一个 evaluator 函数，返回 True 才执行内部步骤，否则跳过，还可以配 else_steps 走另一条路。典型场景是质量门禁——摘要里出现"数据显示""百分比"这类事实性关键词时才插入事实核查 Agent，不命中就省下这次模型调用。路径超过两条时换成 Router：给它一组 choices，由 selector 函数根据主题关键词动态决定走哪条，比如命中 tech、AI 就路由给 HackerNews 研究员，否则走通用网页研究员。更省事的是 CEL 表达式：一行字符串如input.contains("urgent")就能声明式地完成条件、循环和路由判断，还能引用上一步输出与会话状态，示例见 cel_basic.py。

Condition 管"是/否"，Router 管"多选一"，CEL 让这两者一行配置搞定。

从零搭一条内容流水线

以"博客自动生产"为例，按四步走：

1. 最小可用配置：把 Agent、Team 甚至普通函数直接丢进Workflow(steps=[研究步骤, 规划步骤, 写作步骤])。函数步骤读取StepInput、返回StepOutput即可参与流水线，天然拿到上一步的previous_step_content，完整代码参考 sequence_of_steps.py。
2. 关键参数：给 Workflow 传db=SqliteDb(...)持久化会话；步骤间衔接全靠previous_step_content，不用自己传参。
3. 进阶玩法：需要重试就在中间插 Loop，需要分流就换 Router，简单判断直接写 CEL 字符串。
4. 常见坑：Loop 忘了写 max_iterations 容易空转；并行分支的结果是合并后到达下一步的，别假设顺序；CEL 能力依赖 cel-python，缺了要先安装。

先跑通直线，再加控制流——一步只引入一个新概念，出问题才好定位。

按需选用速查

上手路线

1. 跑通直线：在 01_basic_workflows 里把 sequence_of_steps.py 跑起来，验证方式是看到"研究、规划"两步依次产出结果。
2. 加条件：02_conditional_execution 里改 evaluator 的关键词列表，验证某输入走、某输入跳。
3. 加循环：03_loop_execution 中调小 max_iterations，观察重试轮数是否被限制。
4. 加分支与路由：05_conditional_branching 里换主题输入，看请求是否路由到不同研究员。
5. 上声明式配置：07_cel_expressions 用 CEL 字符串替换手写判断函数，对比代码量变化。

适合谁，不适合谁

它适合已经在用 Agno 搭 Agent、想把"多个智能体 + 固定业务流程"落成稳定自动化服务的开发者；不适合流程完全由模型自主决定、不需要确定性控制流的场景——那种情况直接用 Agent 或 Team 就够了。引擎源码在 libs/agno/agno/workflow/ 下，想深入可对照 workflow.py、condition.py、loop.py 等模块读实现。

Step 打底、条件与循环控质量、并行与路由控走向、CEL 省代码、Early Stop 兜底——六个概念记住，任何多智能体业务都能编排成流水线。

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno