agno Workflow Router 实战用 CEL 表达式驱动工作流动态路由与分支选择【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本篇技术指南以 agno 仓库cookbook/04_workflows/07_cel_expressions/router/下的 5 个可运行示例为核心讲解如何在Workflow中用Router组件与 CELCommon Expression Language表达式实现“由数据决定执行哪条分支”的动态路由。读完本文你将掌握Router.selector的五种典型写法additional_data、前序步骤输出、session_state、三元表达式、step_choices索引理解其底层求值机制与上下文变量并能在你自己的多 Agent 工作流中直接复用这些模式。一、示例目录速览5 个路由场景5 种数据来源cookbook/04_workflows/07_cel_expressions/router/下的示例全部是可独立运行的工作流脚本统一展示了同一个核心能力Router用一段 CEL 表达式作为selector根据运行时的不同数据来源决定执行choices中的哪一个Step。每个文件对应一种典型的路由决策数据源文件演示点决策数据来源cel_additional_data_route.pyadditional data route调用方传入的additional_data.route上游/UI 决定分支cel_previous_step_route.pyprevious step route前序命名步骤分类器的输出cel_session_state_route.pysession state route会话状态中的持久化偏好cel_ternary.pyternary用户input文本内容关键字cel_using_step_choices.pystep choices基于choices列表下标动态引用分支运行前置条件原 README 列出的前提与仓库其他示例一致需要依次满足激活 demo 环境.venvs/demo/bin/python通过direnv allow加载 API Keys需要本地存在.envrc文件安装cel-python示例脚本均以from agno.workflow import CEL_AVAILABLE做环境探测未安装时会打印CEL is not available. Install with: pip install cel-python并退出。pip install cel-python .venvs/demo/bin/python cookbook/04_workflows/07_cel_expressions/router/cel_ternary.py二、先理解 Router三种选择机制与 CEL 上下文Router是 agnoWorkflow中负责“动态选路”的组件定义在 libs/agno/agno/workflow/router.py。从它的类文档与字段可以归纳出三种工作模式router.py程序化选择callable selectorselector传入一个接收StepInput并返回 step / step 名列表的 Python 函数CEL 表达式选择字符串 selectorselector是一段返回分支 step 名的 CEL 表达式字符串人工介入选择HITL设置requires_user_inputTrue暂停工作流让用户从choices中挑选。本文聚焦第 2 种模式。Router的关键字段是choices可供选择的分支Step列表与selector决定执行哪条分支序列化时字符串类型的 selector 会被标记为selector_typecel见 router.py 的to_dict实现。CEL selector 表达式内可访问的上下文变量与Condition一致并额外多出step_choicesrouter.py、cel.py上下文变量类型含义inputstring本次工作流输入的字符串形式previous_step_contentstring上一步骤的输出内容previous_step_outputsmap所有已完成步骤step_name - content的映射additional_datamap调用工作流时传入的附加数据session_statemap会话状态字典step_choiceslist(string)当前choices中各分支的 step 名列表CEL 表达式必须返回choices中某个分支的名字。底层求值由evaluate_cel_router_selector完成它先通过_build_step_input_contextcel.py把StepInput与session_state组装成input / previous_step_content / previous_step_outputs / additional_data / session_state上下文再注入step_choices最后用_evaluate_cel_string求值并强转为字符串cel.py。Python 原生值在求值前统一经_to_cel转换为 CEL 类型cel.py因此布尔、整数、字符串、列表、字典都能在表达式中直接使用。字符串是否被当作 CEL 表达式判断由is_cel_expressioncel.py完成纯 Python 标识符如函数名my_evaluator返回False而包含.、()、?、比较/逻辑运算符、引号等 CEL 特征 token 时返回True。三、路由由上游决定additional_data.route第一个场景解决的是“路由决策发生在工作流之外”的情况例如由 UI 表单或上层编排器指定本次要写邮件、博客还是推文。cel_additional_data_route.py 完整代码如下from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.workflow import CEL_AVAILABLE, Step, Workflow from agno.workflow.router import Router if not CEL_AVAILABLE: print(CEL is not available. Install with: pip install cel-python) exit(1) email_agent Agent( nameEmail Writer, modelOpenAIChat(idgpt-5.6-luna), instructionsYou write professional emails. Be concise and polished., markdownTrue, ) blog_agent Agent( nameBlog Writer, modelOpenAIChat(idgpt-5.6-luna), instructionsYou write engaging blog posts with clear structure and headings., markdownTrue, ) tweet_agent Agent( nameTweet Writer, modelOpenAIChat(idgpt-5.6-luna), instructionsYou write punchy tweets. Keep it under 280 characters., markdownTrue, ) workflow Workflow( nameCEL Additional Data Router, steps[ Router( nameContent Format Router, selectoradditional_data.route, choices[ Step(nameEmail Writer, agentemail_agent), Step(nameBlog Writer, agentblog_agent), Step(nameTweet Writer, agenttweet_agent), ], ), ], ) if __name__ __main__: print(--- Route to email ---) workflow.print_response( inputWrite about our new product launch., additional_data{route: Email Writer}, ) print() print(--- Route to tweet ---) workflow.print_response( inputWrite about our new product launch., additional_data{route: Tweet Writer}, )要点拆解selector 为additional_data.route即直接读取additional_data字典的route键三次调用传同一个input仅靠additional_data{route: ...}切换分支说明路由与主输入内容解耦传入的值如Email Writer必须与choices中Step.name精确一致否则会走不到任何分支详见下文“解析与容错”。四、路由由分类结果决定previous_step_outputs 嵌套三元当路由依赖一个“先分类、再处理”的前置步骤时可用previous_step_outputs按步骤名取到分类器输出。cel_previous_step_route.py 的完整定义如下from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.workflow import CEL_AVAILABLE, Step, Workflow from agno.workflow.router import Router if not CEL_AVAILABLE: print(CEL is not available. Install with: pip install cel-python) exit(1) classifier Agent( nameClassifier, modelOpenAIChat(idgpt-5.6-luna), instructions( Classify the request into exactly one category. Respond with only one word: BILLING, TECHNICAL, or GENERAL. ), markdownFalse, ) billing_agent Agent( nameBilling Support, modelOpenAIChat(idgpt-5.6-luna), instructionsYou handle billing inquiries. Help with invoices, payments, and subscriptions., markdownTrue, ) technical_agent Agent( nameTechnical Support, modelOpenAIChat(idgpt-5.6-luna), instructionsYou handle technical issues. Help with debugging and configuration., markdownTrue, ) general_agent Agent( nameGeneral Support, modelOpenAIChat(idgpt-5.6-luna), instructionsYou handle general inquiries., markdownTrue, ) workflow Workflow( nameCEL Previous Step Outputs Router, steps[ Step(nameClassify, agentclassifier), Router( nameSupport Router, # 通过 previous_step_outputs 映射按步骤名访问分类器输出 selector( previous_step_outputs.Classify.contains(BILLING) ? Billing Support : previous_step_outputs.Classify.contains(TECHNICAL) ? Technical Support : General Support ), choices[ Step(nameBilling Support, agentbilling_agent), Step(nameTechnical Support, agenttechnical_agent), Step(nameGeneral Support, agentgeneral_agent), ], ), ], ) if __name__ __main__: print(--- Billing question ---) workflow.print_response(inputI was charged twice on my last invoice.) print() print(--- Technical question ---) workflow.print_response(inputMy API keeps returning 503 errors.)关键机制工作流先把Step(nameClassify, agentclassifier)放在Router之前执行previous_step_outputs.Classify通过步骤名访问分类器输出CEL 的 map 字段访问语法然后调用.contains(BILLING)做子串匹配多层? :构成嵌套三元表达式链语义上等价于 if/elif/else先命中BILLING走 Billing Support再命中TECHNICAL走 Technical Support否则默认 General Support之所以能按名取数是因为 Router 执行链会把已执行步骤的输出按step_name - StepOutput汇总到router_step_outputs并在_update_step_input_from_outputsrouter.py中合并进previous_step_outputs供后续 selector 读取。值得注意分类器设markdownFalse且被要求“只回答一个词”是为了保证输出干净、便于contains精确命中。实际使用时若输出带格式建议在 CEL 前做归一化或用更宽松的关键词。五、路由偏好跨会话持久session_state如果希望路由偏好跨多次运行保持不变例如某个用户始终想要“简洁版分析”可把它写入session_state。cel_session_state_route.py 展示了两种切换方式from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.workflow import CEL_AVAILABLE, Step, Workflow from agno.workflow.router import Router if not CEL_AVAILABLE: print(CEL is not available. Install with: pip install cel-python) exit(1) detailed_agent Agent( nameDetailed Analyst, modelOpenAIChat(idgpt-5.6-luna), instructionsYou provide detailed, in-depth analysis with examples and data., markdownTrue, ) brief_agent Agent( nameBrief Analyst, modelOpenAIChat(idgpt-5.6-luna), instructionsYou provide brief, executive-summary style analysis. Keep it short., markdownTrue, ) workflow Workflow( nameCEL Session State Router, steps[ Router( nameAnalysis Style Router, selectorsession_state.preferred_handler, choices[ Step(nameDetailed Analyst, agentdetailed_agent), Step(nameBrief Analyst, agentbrief_agent), ], ), ], session_state{preferred_handler: Brief Analyst}, ) if __name__ __main__: print(--- Using session_state preference: Brief Analyst ---) workflow.print_response(inputAnalyze the current state of cloud computing.) print() # 运行期切换偏好 workflow.session_state[preferred_handler] Detailed Analyst print(--- Changed preference to: Detailed Analyst ---) workflow.print_response(inputAnalyze the current state of cloud computing.)要点拆解路由决策完全来自session_state.preferred_handler与每次提问内容无关偏好初始值通过Workflow(..., session_state{...})注入代码展示了运行期动态改状态再复用同一个 workflow 对象的写法workflow.session_state[preferred_handler] Detailed Analyst之后再次print_response第二次运行即路由到 Detailed Analyst在带WorkflowSession持久化的场景中同一模式可让“用户偏好”跨多次会话自动恢复。六、按输入内容即时分流CEL 三元表达式当不需要前置步骤仅凭本次input文本即可分流时直接在 selector 中对input用三元表达式即可。cel_ternary.pyfrom agno.agent import Agent from agno.models.openai import OpenAIChat from agno.workflow import CEL_AVAILABLE, Step, Workflow from agno.workflow.router import Router if not CEL_AVAILABLE: print(CEL is not available. Install with: pip install cel-python) exit(1) video_agent Agent( nameVideo Specialist, modelOpenAIChat(idgpt-5.6-luna), instructionsYou specialize in video content creation and editing advice., markdownTrue, ) image_agent Agent( nameImage Specialist, modelOpenAIChat(idgpt-5.6-luna), instructionsYou specialize in image design, photography, and visual content., markdownTrue, ) workflow Workflow( nameCEL Ternary Router, steps[ Router( nameMedia Router, selectorinput.contains(video) ? Video Handler : Image Handler, choices[ Step(nameVideo Handler, agentvideo_agent), Step(nameImage Handler, agentimage_agent), ], ), ], ) if __name__ __main__: print(--- Video request ---) workflow.print_response(inputHow do I edit a video for YouTube?) print() print(--- Image request ---) workflow.print_response(inputHelp me design a logo for my startup.)模式解析input.contains(video)命中则走Video Handler否则默认Image Handler这是纯关键字分流无需额外 LLM 调用成本最低、延迟最小代价是只能识别硬编码关键词无法理解语义“剪辑”“渲染”等变体需扩展关键词或用前置分类步骤。七、用下标引用分支step_choices让表达式更抗变更前文所有表达式里都硬编码了分支名。当分支列表频繁增删、或你想避免手写名字造成拼写错误时可用step_choices按下标位置引用分支。cel_using_step_choices.pyfrom agno.agent import Agent from agno.models.openai import OpenAIChat from agno.workflow import CEL_AVAILABLE, Step, Workflow from agno.workflow.router import Router if not CEL_AVAILABLE: print(CEL is not available. Install with: pip install cel-python) exit(1) quick_analyzer Agent( nameQuick Analyzer, modelOpenAIChat(idgpt-5.6-luna), instructionsProvide a brief, concise analysis of the topic., markdownTrue, ) detailed_analyzer Agent( nameDetailed Analyzer, modelOpenAIChat(idgpt-5.6-luna), instructionsProvide a comprehensive, in-depth analysis of the topic., markdownTrue, ) workflow Workflow( nameCEL Step Choices Router, steps[ Router( nameAnalysis Router, # step_choices[0] Quick Analysis第一个 choice # step_choices[1] Detailed Analysis第二个 choice selectorinput.contains(quick) || input.contains(brief) ? step_choices[0] : step_choices[1], choices[ Step(nameQuick Analysis, agentquick_analyzer), Step(nameDetailed Analysis, agentdetailed_analyzer), ], ), ], ) if __name__ __main__: print( Quick analysis request ) workflow.print_response( inputGive me a quick overview of quantum computing., streamTrue ) print(\n * 50 \n) print( Detailed analysis request ) workflow.print_response(inputExplain quantum computing in detail., streamTrue)机制与适用建议step_choices是当前choices分支名的字符串列表由evaluate_cel_router_selector注入见 cel.py因此step_choices[0]在运行时等于第一个Step.namestep_choices[1]等于第二个例如input含quick或brief时走step_choices[0]Quick Analysis否则走step_choices[1]示例还演示了streamTrue的流式输出用法优点正如源码注释所述避免步骤名拼写错误、提升表达式可维护性、支持按位置动态引用代价是下标与顺序强耦合插入或重排分支时必须同步检查表达式语义。八、运行时解析与容错selector 结果如何命中分支理解以上 5 个例子的最终落点是弄清“selector 算出的字符串如何变成真正被执行的 step”。链路如下对应同步入口_route_stepsrouter.py异步版_aroute_steps逻辑等价router.py执行Router.execute时先调用_prepare_steps()把choices里裸的Agent等对象包装为Step并构建name - step映射_step_name_maprouter.py若selector是字符串则调用evaluate_cel_router_selector求值并把所有可选分支名作为step_choices传入求值结果交给_resolve_selector_resultrouter.py解析字符串会先在_step_name_map中按名字查找未命中的名字不会报错中断而是记录 warningRouter selector returned unknown step name ...并返回空列表导致该轮 Router “完成 0 个结果”no steps selectedCEL 求值自身失败如语法错误、cel-python 未安装时同样会被捕获并返回空选择同时打印异常日志见_route_steps中的 try/except。这解释了为什么所有示例都强调“表达式返回的名字必须与choices中某个Step.name精确一致”也提示了排错时的首要检查点对照RouterExecutionCompletedEvent/日志中的selected_steps或控制台 warning确认返回值与_step_name_map键集合是否匹配。此外evaluate_cel_router_selector会把求值结果用_evaluate_cel_string强转字符串cel.py所以必须保证表达式最终落在返回字符串的语义上直接取 map 值或三元分支均返回 string而不是布尔或其他类型。九、一个最小的可运行骨架把 5 个例子的共同结构抽出来便得到可以直接改造的最小骨架from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.workflow import Step, Workflow from agno.workflow.router import Router branch_a Agent(nameBranch A, modelOpenAIChat(idgpt-5.6-luna), instructions...) branch_b Agent(nameBranch B, modelOpenAIChat(idgpt-5.6-luna), instructions...) workflow Workflow( nameMy Router Workflow, steps[ Router( nameMy Router, selectorCEL 表达式返回下列某个 Step.name, choices[Step(nameBranch A, agentbranch_a), Step(nameBranch B, agentbranch_b)], ), ], ) workflow.print_response(input...)写表达式时对照前文的上下文变量表选择数据源外部指定用additional_data、先判后处理用previous_step_outputs、跨会话偏好用session_state、纯内容分流用input三元、抗变更引用用step_choices。十、延伸阅读与验证本目录配套 TEST_LOG.md逐文件记录运行与预期行为校验状态可据此逐个执行脚本核对路由是否命中预期分支CEL 表达式同样可用于Condition条件判断与Loop循环终止等步骤相关示例见 07_cel_expressions/condition 与 07_cel_expressions/loop它们与 Router 共享同一套celpy求值内核核心实现均在 libs/agno/agno/workflow/router.pyRouter数据类与执行链路和 libs/agno/agno/workflow/cel.pycelpy封装、上下文构建、三类求值入口中深入阅读可看到 CEL selector 与 callable selector、HITL 选择的完整分支逻辑。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考