如何用 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], evaluatorlambda 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_iterations3, end_conditionlambda 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 表达式声明判断规则判断逻辑如果只是一次字符串包含、数值比较专门写一个函数有点重。CELCommon Expression Language表达式允许你直接传一行字符串Condition 的 evaluator、Loop 的 end_condition、Router 的 selector 三个控制点都支持。使用前安装依赖pip install cel-python表达式统一可以访问这些上下文变量input原始输入、previous_step_content、previous_step_outputs、additional_data、session_stateRouter 里额外有step_choices可选步骤名列表。几个典型写法Conditioninput.contains(urgent)输入含 urgent 即走加急分支Loopall_success current_iteration 2本轮全部成功且已跑满两轮才退出可用变量还包括max_iterations、last_step_content、step_outputsRouterinput.contains(video) ? video_step : image_step三目运算直接返回步骤名。三类 CEL 示例分别放在 cookbook/04_workflows/07_cel_expressions/condition/、loop/、router/ 三个子目录对照着改规则最快。检测到严重问题时如何立即停止StepOutput 的 stop 信号跳过某段流程是 Condition 的职责但发现输入不合规、中间结果已失败、用户取消任务这类情况需要整个工作流立即收场。方式很直接任何步骤返回带stopTrue的 StepOutput剩余步骤全部跳过工作流以 Early Stopped 状态结束。if VULNERABLE in (step_input.previous_step_content or ): return StepOutput(contentDeployment blocked by security scan, stopTrue)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 目录编号基本对齐目录覆盖能力01_basic_workflows/顺序执行、函数步骤、嵌套 Workflow02_conditional_execution/Condition 基础、else 分支、并行嵌套03_loop_execution/循环退出条件、累积式循环04_parallel_execution/多路并行与并行内条件05_conditional_branching/Router 选路、嵌套与字符串选择器07_cel_expressions/声明式判断、退出与路由规则遇到想深挖执行细节的问题直接读 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考