Agno Workflow 工作流编排实战条件分支与并行速查【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno假设一条内容要经过采集、摘要、事实核查、写作、审阅。手写 if-else 链路步骤一多就很难回答哪一步执行了、在哪里分叉、中间结果去了哪。AgnoAgno的 Workflow 引擎用工作流编排解决这个问题把任务拆成 Step执行单元相当于流水线上的一个工位再用 Condition条件分支、Loop循环等控制结构串联整条链路可观测、可暂停、可回放。编排原语速览五种组件一次讲清原语解决什么问题对应类或参数顺序步骤Step前一步结果原样传给下一步一切流水线的底座Workflow(steps[...])读StepInput.previous_step_contentCondition 条件分支满足条件才执行一段否则跳过或走 else 分支Condition(evaluator..., else_steps...)Loop 循环结果不达标就重复执行有上限兜底Loop(max_iterations..., end_condition...)Parallel 并行互不依赖的任务同时跑全部完成后自动合并Parallel(steps[...])Router 多路分发按输入或上一步内容动态选择走哪条路Router(choices..., selector...)五种原语彼此可以嵌套Condition 里放 RouterLoop 里放 Parallel。每种原语的源码都在 libs/agno/agno/workflow/ 下独立成模块逐个读没有负担。内容生产从采集到成稿的串并联组合这是最常见的拓扑两个互不依赖的采集任务并行结果合并后再交给写作。Parallel并行容器同时发起内部步骤全部完成后自动把各路输出汇总为下一步的previous_step_content下游写作步骤不需要知道上游有几个来源。组合的关键就在并行后的合并Parallel(steps[hn_research_step, web_research_step])完整示例见 parallel_basic.py它演示了并行研究 → 写作 → 审阅的完整链路同目录下还有并行加条件的组合写法。质量门禁条件判断 循环重试如何协作质量诉求分两类。一类是该不该跑摘要里出现数据显示百分比这类事实性关键词时才需要事实核查步骤——这是Conditionevaluator 返回 bool决定进入 if 分支还是 else 分支。另一类是跑到达标为止研究结果不足 200 字符就再来一轮——这是Loop每轮结束后由 end_condition 评估是否退出max_iterations默认 3兜底防止死循环。condition_basic.py 里的门禁 evaluator 读step_input.previous_step_content做关键词匹配loop_basic.py 的循环 evaluator 直接检查输出长度。两者可以叠用Loop 内嵌 Condition实现本轮输出达标才进入审阅这类复合门禁。动态分流多路路由与声明式规则分支超过两条时if-else 就开始难维护这是Router多路分发器根据上下文动态选择执行哪条分支的活。声明一组候选步骤choicesselector函数根据 StepInput 动态返回走哪条路也可以配纯字符串选择器或人工选择。router_basic.py 里 selector 检查主题是否包含 tech、AI 等关键词命中走 HackerNews 研究员未命中走通用网页研究员最后统一进入发布步骤。嵌套路由、字符串选择器、媒体管线等进阶写法集中在 05_conditional_branching/ 目录。跨场景进阶技巧CEL、提前终止与人工确认遇到分支规则都是简单的内容匹配 → 用 CEL 表达式通用表达式语言用一行字符串写判断逻辑例如evaluatorinput.contains(urgent)无需写函数。示例见 cel_basic.py需pip install cel-python。遇到输入明显不合规或中间结果已失败不想再跑剩余步骤 → 在函数步骤里返回StepOutput(stopTrue)整条流水线标记为提前终止并跳过后续步骤见 early_stop_basic.py。遇到循环需要把本轮结果作为下一轮起点 → 设置Loop(forward_iteration_outputTrue)默认每轮拿的是原始输入不传上一轮结果。遇到关键分支需要人拍板 → 给 Condition 或 Router 配置 human_review 确认项流程会暂停等待人工决定再继续示例在 08_human_in_the_loop/ 目录。遇到多轮任务要跨次运行恢复 → Workflow 入口支持接入 SQLite 等数据库持久化会话每步输出与运行状态可回放。动手路线五步能力递进先跑通顺序流水线从 01_basic_workflows/ 选一条研究 → 摘要 → 写作直线流水线理解 Step 与 StepOutput 如何在步骤间传递。加一个条件分支把摘要后加事实核查门禁参考 condition_basic.py改几个关键词观察 evaluator 返回 True/False 的差异。加一个循环重试把研究步骤包进Loop(max_iterations3)调 end_condition 感受不同退出时机。加并行把两个独立研究源改成Parallel对比串行与并行下输出合并的差别。加路由把关键词 if-else 改写为Router(choices...)最后把 selector 换成 CEL 表达式完成声明式收尾。读源码时按执行链路来workflow.py 是主循环condition.py、loop.py、router.py 各对应一种控制结构文件不大一个下午能读完。Workflow 引擎的真正价值不在组件数量而在步骤之间有一份清晰的契约每个步骤接收 StepInput、返回 StepOutput控制结构只负责决定何时执行哪个工位。契约立住了新增一个分支、一轮重试或一个人工检查点都只是局部改动。生产化时建议先接上流式事件与会话持久化把可观测性做足再叠加复杂路由。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考