Agno Workflow 条件分支实战用 Router 构建动态选择的多 Agent 工作流【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本篇技术指南以 agno 仓库中cookbook/04_workflows/05_conditional_branching目录为核心围绕Router路由器这一条件分支组件展开讲解如何在 agno Workflow 中根据输入动态选择要执行的 Step、Loop 或 Steps 序列。读完本文你将掌握 Router 的三种选择模式函数选择器、CEL 表达式、人工介入、四种返回值形态字符串、Step 对象、Step 列表、嵌套列表以及如何组合Loop与Steps构建自适应路由并了解该目录下 8 个示例脚本的测试验证结果。目录概览条件分支实验场的构成cookbook/04_workflows/05_conditional_branching目录是一个可运行的 Workflow 条件分支示例集合由三部分构成README.md声明目录范围Scope、8 个示例文件清单及运行前提8 个.py示例脚本覆盖路由的多种形态TEST_LOG.md2026-02-08 生成的自动化测试日志逐文件记录执行状态、运行方式、超时阈值与结果摘要。示例文件一览文件演示主题测试状态loop_in_choices.py将 Loop 作为 Router 的 choice 之一PASSnested_choices.pyRouter choices 中的嵌套列表PASSrouter_basic.py基于主题的简易路由基础版FAIL35s 超时router_with_loop.py路由到 Loop 的深度技术调研FAIL35s 超时selector_media_pipeline.py图片/视频生成管线的路由选择FAIL35s 超时selector_types.pyselector 的多种返回形态PASSstep_choices_parameter.pyselector 中使用step_choices参数PASSstring_selector.py字符串形式的 selector返回 step 名称PASS运行前提根据 README.md 的 Prerequisites运行这些示例需要激活演示虚拟环境.venvs/demo/bin/python通过direnv allow加载 API Key需要本地存在.envrc文件。测试日志也印证了这一运行方式所有条目均以.venvs/demo/bin/python执行模式为 normal非流式打包测试超时阈值统一为 35s。Router 组件条件分支的核心抽象条件分支在 agno Workflow 中由Router组件承载。从源码看router.py 中Router是一个 dataclass其核心职责是根据输入动态选择要执行的步骤官方注释明确给出了三种工作模式程序化选择Programmatic selection使用selector函数决定执行哪些步骤CEL 表达式选择使用返回 step 名称的 CEL 表达式字符串HITL 选择Human-in-the-loop设置requires_user_inputTrue暂停工作流让用户从choices中选择。Router 的关键字段dataclass class Router: # 可被选择执行的所有步骤 choices: WorkflowSteps # 选择器函数或 CEL 表达式HITL 模式下可省略 selector: Optional[Union[Callable, str]] None name: Optional[str] None description: Optional[str] None human_review: HumanReview field(default_factoryHumanReview)其中WorkflowSteps允许 choices 中包含Callable、Step、Steps、Loop、Parallel、Condition、Router甚至嵌套的Workflow见 router.py这意味着路由分支本身可以是任意复杂度的执行单元。选择器可用上下文无论使用函数还是 CEL 表达式selector 都能读取以下上下文见 router.py 与 cel.pyinput工作流输入字符串previous_step_content上一步的内容previous_step_outputs此前所有步骤的名称到内容的映射additional_data传递给工作流的附加数据session_state会话状态映射step_choices当前 selector 可选择的 step 名称列表仅 Router 可用。CEL 表达式示例源码文档中给出input.contains(video) ? video_step : image_stepadditional_data.routeprevious_step_outputs.classifier.contains(billing) ? Billing : SupportCEL 表达式必须返回choices中的某个 step 名称。注意使用 CEL 需要安装cel-pythonpip install cel-python否则会在_route_steps中报错并返回空列表见 router.py。selector 的四种返回值形态selector 函数的返回类型被定义为Union[str, Step, List[Step]]本目录的示例逐一展示了这些形态的实战用法。形态一返回字符串step 名称string_selector.py 演示了最简洁的写法——selector 直接返回 step 的名称字符串def route_by_topic(step_input: StepInput) - Union[str, Step, List[Step]]: topic step_input.input.lower() if tech in topic or ai in topic or software in topic: return Tech Research if business in topic or market in topic or finance in topic: return Business Research return General Research workflow Workflow( nameExpert Routing (String Selector), steps[ Router( nameTopic Router, selectorroute_by_topic, choices[tech_step, business_step, general_step], ), ], )底层实现中字符串结果会通过_step_name_map由_prepare_steps构建的名称到 Step 的映射解析为对应的 Step 对象若返回了未知名称Router 会打印 warning 并返回空列表见 router.py因此务必保证返回值与 choices 中 Step 的name完全一致。形态二使用 step_choices 参数动态选择step_choices_parameter.py 演示了 selector 的第二参数step_choices——Router 会在运行时通过inspect.signature检测 selector 是否声明了该参数若声明则把准备好的 steps 列表注入见 router.pydef dynamic_selector( step_input: StepInput, step_choices: list, ) - Union[str, Step, List[Step]]: user_input step_input.input.lower() step_map {s.name: s for s in step_choices if hasattr(s, name) and s.name} print(fAvailable steps: {list(step_map.keys())}) if research in user_input: return researcher if write in user_input: return step_map.get(writer, step_choices[0]) if full in user_input: return [step_map[researcher], step_map[writer], step_map[reviewer]] return step_choices[0]这段代码展示了三种能力按名称返回字符串、按名称从映射中取 Step 对象、以及返回多个 Step 的列表以实现研究员→写手→审阅者的全流程串行执行。形态三直接返回 Step 对象与列表selector_types.py 将上述形态集中在一个文件中验证依次运行三个工作流字符串 selectorroute_by_topic——按主题路由到 Tech / Business / General 三个专家step_choices 参数dynamic_selector——按research/write/full关键词动态选择嵌套 choicesnested_selector——返回step_choices[1]这样的嵌套列表项。形态四嵌套列表 choicesnested_choices.py 单独演示了嵌套形态。当 choices 中出现列表时如choices[step_a, [step_b, step_c]]_prepare_steps会将其自动包装为一个Steps容器steps_group_{index}从而把一次选择变成顺序执行多个 Step见 router.pyworkflow Workflow( nameNested Choices Routing, steps[ Router( nameNested Router, selectornested_selector, choices[step_a, [step_b, step_c]], ), ], )当用户输入包含 single 时只执行step_a否则执行[step_b, step_c]组成的顺序序列。在 choices 中组合 Loop 与 StepsRouter 的 choices 不限于简单 Step还可以是循环与管线序列。Loop 作为路由分支loop_in_choices.pyloop_in_choices.py 将Loop组件直接作为 Router 的 choice 之一构建了一个快速回答 / 草稿 / 迭代润色三选一路由refinement_loop Loop( namerefinement_loop, steps[Step(namerefine_step, agentrefiner)], max_iterations2, ) def loop_selector( step_input: StepInput, step_choices: list, ) - Union[str, Step, List[Step]]: user_input step_input.input.lower() if quick in user_input: return step_choices[0] if refine in user_input or polish in user_input: return [step_choices[1], step_choices[2]] return step_choices[1] workflow Workflow( nameLoop Choice Routing, steps[ Router( nameContent Router, selectorloop_selector, choices[quick_response, draft_writer, refinement_loop], ), ], )这里Loop的max_iterations2限制迭代上限。从 loop.py 源码可见Loop默认max_iterations3并支持end_condition回调函数或 CEL 表达式提前结束循环——这正是下一个示例的核心。路由到深度调研循环router_with_loop.pyrouter_with_loop.py 演示了简单主题走单次 Web 调研深度科技主题走迭代循环调研的自适应策略。关键在于Loop的end_condition回调def research_quality_check(outputs: List[StepOutput]) - bool: if not outputs: return False for output in outputs: if output.content and len(output.content) 300: print(f[PASS] Research quality check passed - found substantial content ({len(output.content)} chars)) return True print([FAIL] Research quality check failed - need more substantial research) return False deep_tech_research_loop Loop( nameDeep Tech Research Loop, steps[research_hackernews], end_conditionresearch_quality_check, max_iterations3, descriptionPerform iterative deep research on tech topics, )路由规则是当主题命中deep_tech_keywords如 ai developments、blockchain technology、github trends或包含 tech 且词数超过 3 个时返回[deep_tech_research_loop]否则返回[research_web]。该示例同时验证了同步print_response与异步aprint_response两种运行方式。媒体生成管线路由selector_media_pipeline.pyselector_media_pipeline.py 将 Router 用于多模态媒体生成根据用户消息是否包含 video 或 image 关键词路由到视频或图片生成管线。两条管线均由Steps容器顺序串联image_sequence Steps( nameimage_generation, descriptionComplete image generation and analysis workflow, steps[generate_image_step, describe_image_step], ) video_sequence Steps( namevideo_generation, descriptionComplete video production and analysis workflow, steps[generate_video_step, describe_video_step], ) def media_sequence_selector(step_input: StepInput) - List[Step]: if not step_input.input or not isinstance(step_input.input, str): return [image_sequence] message_lower step_input.input.lower() if video in message_lower: return [video_sequence] if image in message_lower: return [image_sequence] return [image_sequence]该示例还定义了MediaRequestPydantic 模型含topic、content_type、prompt、style、duration、resolution字段作为结构化输入参考说明路由前可以先用结构化模型承载媒体请求参数。真实场景智能研究发布工作流router_basic.pyrouter_basic.py 是条件分支最贴近业务的示例一个自动选择研究方式并发布内容的工作流包含两个顺序步骤——Router研究策略路由与publish_content发布 Step。路由规则将tech_keywords涵盖 startup、programming、ai、machine learning、software、blockchain、open source、github 等 14 个关键词与用户输入做子串匹配命中则走 HackerNews 研究HackerNewsTools否则走通用 Web 研究WebSearchTools最后统一交给内容发布 Agent 排版输出workflow Workflow( nameIntelligent Research Workflow, descriptionAutomatically selects the best research method based on topic, then publishes content, steps[ Router( nameresearch_strategy_router, selectorresearch_router, choices[research_hackernews, research_web], descriptionIntelligently selects research method based on topic, ), publish_content, ], )示例同时演示了四种运行方式同步print_response、同步流式streamTrue、异步aprint_responseasyncio.run包裹与异步流式可满足不同场景的调用需求。测试日志解读与结果分析TEST_LOG.md 是理解本目录运行状态的第一手数据。它记录了 8 个示例的自动化执行结果其中 5 个 PASS、3 个 FAIL失败原因高度一致——35s 超时示例状态结果摘要原文loop_in_choices.pyPASSExecuted successfully输出涉及 Jupyter 数据可视化内容nested_choices.pyPASSExecuted successfully耗时 2.8srouter_basic.pyFAILTimed out after 35s日志停在 TOOL METRICS 调试输出router_with_loop.pyFAILTimed out after 35s日志停在创建 gpt-5.6-luna 的 async OpenAI clientselector_media_pipeline.pyFAILTimed out after 35s日志停在generate_image(prompt...)selector_types.pyPASSExecuted successfully耗时 2.4sstep_choices_parameter.pyPASSExecuted successfully耗时 1.7sstring_selector.pyPASSExecuted successfully耗时 8.4s失败原因分析三个 FAIL 的示例并非代码逻辑错误而是运行特征所致涉及外部工具与多模型调用router_basic.py调用了HackerNewsTools与WebSearchTools联网工具selector_media_pipeline.py调用了OpenAIToolsgpt-image-1与GeminiTools(vertexaiTrue)媒体生成工具。外部 API 的响应延迟容易超出 35s 阈值涉及模型初始化与长输出router_with_loop.py的日志停留在Creating new async OpenAI client for model gpt-5.6-luna说明在模型客户端初始化阶段即超时——这与示例中模型 ID 为gpt-5.6-lunaOpenAI 兼容接口有关示例运行依赖可用的 API Key 环境对比验证同为 Router 示例的string_selector.py8.4s与nested_choices.py2.8s能快速通过因为它们不依赖外部工具调用进一步印证了超时与外部依赖相关。对读者的启示本地复现时若脚本因 35s 超时而中断可先确认.envrc中 API Key 已正确加载direnv allow或适当延长超时时间后再运行涉及工具调用的三个示例。总结cookbook/04_workflows/05_conditional_branching用 8 个可运行示例完整覆盖了 agno Workflow 条件分支的编程模型Router通过selector动态选择choices中的执行单元支持函数选择器、CEL 表达式与 HITL 三种模式selector 可以返回字符串名称、Step 对象、Step 列表或嵌套列表choices 则可以容纳Step、Steps、Loop、Parallel、Condition、Router乃至嵌套 Workflow见 router.py。测试日志表明纯逻辑路由示例字符串选择、嵌套、step_choices均可在数秒内稳定通过而依赖外部工具与媒体生成模型的示例则需更充分的环境配置与超时预算。如果希望进一步深挖可以继续阅读 agno 源码 中_route_steps/_aroute_steps/execute/aexecute的路由与串行链式执行实现或查看cookbook/04_workflows下 02_conditional_execution、07_cel_expressions 等相邻目录掌握条件执行与 CEL 表达式的更多细节。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考