如何用 Agno Workflow 搭建灵活的 Agent 流水线：6 种编排控制方法完整指南

如何用 Agno Workflow 搭建灵活的 Agent 流水线：6 种编排控制方法完整指南

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

把多个 Agent 接在一条线上不难，难的是流程走向：什么时候该分叉、什么时候该并排跑、什么时候该回头重试、什么时候该整体停下。Agno 框架内置的 Workflow 编排模块就是为此设计的：任务被拆成一个个 Step（步骤），再由 Condition、Loop、Parallel、Router、CEL 表达式和提前终止这 6 种控制方式决定执行路径。本文从最小执行单元讲起，逐步叠加每种控制结构，带你搭出一条可控的 Agent 流水线。

步骤之间如何传递数据：Step 与 StepInput 的基本约定

一个 Step 是一次完整作业。创建时它必须、且只能指定一种执行器：一个 Agent、一个 Team、一个普通函数，或一个嵌套的子 Workflow，四者并存会直接报错（校验逻辑在 libs/agno/agno/workflow/step.py 的_validate_executor_config）。Workflow 按列表顺序逐个执行 Step，并且自动把上一步的产出封装成下一步的输入，不需要你手动搬运数据。

每个 Step 收到的输入是StepInput对象，几个关键字段值得记住：

• input：工作流启动时的原始输入；
• previous_step_content：紧邻上一步的产出内容；
• previous_step_outputs：以步骤名为键的全量历史输出，可以用get_step_content("step_name")按名取任意一步的结果；
• additional_data：调用方额外传入的字典数据。

函数也可以成为步骤：只要它接收StepInput并返回StepOutput。Step 还自带max_retries（默认 3 次重试）和skip_on_failure两个容错开关。一个最小的直线工作流长这样：

workflow = Workflow(steps=[research_step, outline_step, writing_step])

先跑通直线，再谈控制。示例集中在 cookbook/04_workflows/01_basic_workflows/ 目录，01_sequence_of_steps/sequence_of_steps.py展示了纯 Agent 序列，02_step_with_function/step_with_function.py演示函数步骤。

条件分支与多路分发：Condition 和 Router 各管多少路

两条路的分叉交给 Condition，多路的分发交给 Router。两者都接收一个"裁判"：Condition 的evaluator返回布尔值，为真执行steps，为假则走可选的else_steps，整个 Condition 被跳过；Router 的selector从一组choices候选中挑出一个或多个实际执行。

Condition 适合做"过不过"的二元判断，比如检查上一步产出里是否出现需要人工确认的表述：

condition = Condition( steps=[verify_step], evaluator=lambda inp: "待核实" in (inp.previous_step_content or ""), else_steps=[skip_step], )

Condition 内部还能再嵌套 Parallel 或 Loop，分支里做并行完全合法，见 condition_with_parallel.py。

Router 适合"三选一"甚至"多选几"的分发：selector 是个函数或字符串，按输入动态选路。命中科技类主题走专用研究员、其余走通用研究员，就是 router_basic.py 的结构。如果不想写判断逻辑，把human_review里的requires_user_input置为 True，工作流会暂停，把候选列表展示给你来选。

更多变体（嵌套路由、字符串选择器、媒体管线分流）都在 cookbook/04_workflows/05_conditional_branching/ 下。

并行执行如何压缩总耗时：Parallel 的多路同时起跑

多个相互独立的数据源没必要排队跑。Parallel 把多条路径同时启动，等全部完成后自动把各路StepOutput合并成一份结果交给下游，归并代码一行都不用你写。整段并行区间的耗时约等于最慢那条路径，而不是各路径之和。

构造方式用位置参数，第一个位置参数可以传自定义名字：

gather = Parallel( "multi_source_research", hn_step, web_step, )

一个细节：并行分支内部也可以再放 Condition，先分流再汇聚，parallel_with_condition.py 演示了这个组合。源码在 libs/agno/agno/workflow/parallel.py，可以看到它用线程池驱动各路径、并处理会话状态的合并。

如何配置循环直到产出达标：Loop 的退出条件与累积模式

大模型单次产出的质量有波动，"不达标就重跑"几乎每个管线都会用到。Loop 把一组步骤包起来反复执行：每轮跑完后调用end_condition评估函数，它接收本轮全部 StepOutput 的列表并返回布尔值，返回真才跳出循环；max_iterations（缺省 3）设定轮数上限，避免评估函数永远返回假。

retry = Loop( steps=[draft_step], max_iterations=3, end_condition=lambda outs: all(len(o.content or "") > 300 for o in outs), )

还有一个容易忽略的开关forward_iteration_output：默认每轮都拿到最初那份输入；置为 True 后，上一轮的产出会作为下一轮的输入，形成"越改越好"的累积式循环，loop_iterative_accumulation.py 就是这个模式。Loop 内部同样允许嵌套 Parallel，实现"每轮多路并发、整体按轮迭代"。

实现细节见 libs/agno/agno/workflow/loop.py。

不写 Python 函数，用 CEL 表达式声明判断规则

判断逻辑如果只是一次字符串包含、数值比较，专门写一个函数有点重。CEL（Common Expression Language）表达式允许你直接传一行字符串，Condition 的 evaluator、Loop 的 end_condition、Router 的 selector 三个控制点都支持。使用前安装依赖：

pip install cel-python

表达式统一可以访问这些上下文变量：input（原始输入）、previous_step_content、previous_step_outputs、additional_data、session_state；Router 里额外有step_choices（可选步骤名列表）。几个典型写法：

• Condition：'input.contains("urgent")'，输入含 urgent 即走加急分支；
• Loop：'all_success && current_iteration >= 2'，本轮全部成功且已跑满两轮才退出，可用变量还包括max_iterations、last_step_content、step_outputs；
• Router：'input.contains("video") ? "video_step" : "image_step"'，三目运算直接返回步骤名。

三类 CEL 示例分别放在 cookbook/04_workflows/07_cel_expressions/condition/、loop/、router/ 三个子目录，对照着改规则最快。

检测到严重问题时如何立即停止：StepOutput 的 stop 信号

跳过某段流程是 Condition 的职责，但"发现输入不合规、中间结果已失败、用户取消任务"这类情况需要整个工作流立即收场。方式很直接：任何步骤返回带stop=True的 StepOutput，剩余步骤全部跳过，工作流以 Early Stopped 状态结束。

if "VULNERABLE" in (step_input.previous_step_content or ""): return StepOutput(content="Deployment blocked by security scan", stop=True)

early_stop_basic.py的部署管线就是这种用法：安全扫描之后接一个函数步骤，检查结果异常就发停止信号。在 Condition、Loop、Parallel 内部触发停止的完整示例集在 cookbook/04_workflows/06_advanced_concepts/early_stopping/ 子目录，stop字段的定义在 libs/agno/agno/workflow/types.py 的StepOutput中。

示例导航：按顺序过一遍 cookbook 的 6 个目录

建议的学习路径与 cookbook 目录编号基本对齐：

遇到想深挖执行细节的问题，直接读 libs/agno/agno/workflow/ 下的workflow.py、condition.py、loop.py、parallel.py、router.py、cel.py，每个文件对应一种控制结构的完整实现。本地跑示例前，先确认你已经在 01_basic_workflows/ 里验证过previous_step_content能正常读到上一步内容——这一步通了，后面所有控制结构都有数据基础。

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