企业官网建设流程全解析

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组件与 CEL（Common 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。每个文件对应一种典型的路由决策数据源：

运行前置条件

原 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）：

1. 程序化选择（callable selector）：selector传入一个接收StepInput并返回 step / step 名列表的 Python 函数；
2. CEL 表达式选择（字符串 selector）：selector是一段返回分支 step 名的 CEL 表达式字符串；
3. 人工介入选择（HITL）：设置requires_user_input=True，暂停工作流让用户从choices中挑选。

本文聚焦第 2 种模式。Router的关键字段是choices（可供选择的分支Step列表）与selector（决定执行哪条分支），序列化时字符串类型的 selector 会被标记为selector_type="cel"（见 router.py 的to_dict实现）。

CEL selector 表达式内可访问的上下文变量与Condition一致，并额外多出step_choices（router.py、cel.py）：

CEL 表达式必须返回choices中某个分支的名字。底层求值由evaluate_cel_router_selector完成：它先通过_build_step_input_context（cel.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_expression（cel.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( name="Email Writer", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You write professional emails. Be concise and polished.", markdown=True, ) blog_agent = Agent( name="Blog Writer", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You write engaging blog posts with clear structure and headings.", markdown=True, ) tweet_agent = Agent( name="Tweet Writer", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You write punchy tweets. Keep it under 280 characters.", markdown=True, ) workflow = Workflow( name="CEL Additional Data Router", steps=[ Router( name="Content Format Router", selector="additional_data.route", choices=[ Step(name="Email Writer", agent=email_agent), Step(name="Blog Writer", agent=blog_agent), Step(name="Tweet Writer", agent=tweet_agent), ], ), ], ) if __name__ == "__main__": print("--- Route to email ---") workflow.print_response( input="Write about our new product launch.", additional_data={"route": "Email Writer"}, ) print() print("--- Route to tweet ---") workflow.print_response( input="Write 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( name="Classifier", model=OpenAIChat(id="gpt-5.6-luna"), instructions=( "Classify the request into exactly one category. " "Respond with only one word: BILLING, TECHNICAL, or GENERAL." ), markdown=False, ) billing_agent = Agent( name="Billing Support", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You handle billing inquiries. Help with invoices, payments, and subscriptions.", markdown=True, ) technical_agent = Agent( name="Technical Support", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You handle technical issues. Help with debugging and configuration.", markdown=True, ) general_agent = Agent( name="General Support", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You handle general inquiries.", markdown=True, ) workflow = Workflow( name="CEL Previous Step Outputs Router", steps=[ Step(name="Classify", agent=classifier), Router( name="Support 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(name="Billing Support", agent=billing_agent), Step(name="Technical Support", agent=technical_agent), Step(name="General Support", agent=general_agent), ], ), ], ) if __name__ == "__main__": print("--- Billing question ---") workflow.print_response(input="I was charged twice on my last invoice.") print() print("--- Technical question ---") workflow.print_response(input="My API keeps returning 503 errors.")

关键机制：

• 工作流先把Step(name="Classify", agent=classifier)放在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_outputs（router.py）中合并进previous_step_outputs，供后续 selector 读取。

值得注意：分类器设markdown=False且被要求“只回答一个词”，是为了保证输出干净、便于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( name="Detailed Analyst", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You provide detailed, in-depth analysis with examples and data.", markdown=True, ) brief_agent = Agent( name="Brief Analyst", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You provide brief, executive-summary style analysis. Keep it short.", markdown=True, ) workflow = Workflow( name="CEL Session State Router", steps=[ Router( name="Analysis Style Router", selector="session_state.preferred_handler", choices=[ Step(name="Detailed Analyst", agent=detailed_agent), Step(name="Brief Analyst", agent=brief_agent), ], ), ], session_state={"preferred_handler": "Brief Analyst"}, ) if __name__ == "__main__": print("--- Using session_state preference: Brief Analyst ---") workflow.print_response(input="Analyze the current state of cloud computing.") print() # 运行期切换偏好 workflow.session_state["preferred_handler"] = "Detailed Analyst" print("--- Changed preference to: Detailed Analyst ---") workflow.print_response(input="Analyze 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.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) video_agent = Agent( name="Video Specialist", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You specialize in video content creation and editing advice.", markdown=True, ) image_agent = Agent( name="Image Specialist", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You specialize in image design, photography, and visual content.", markdown=True, ) workflow = Workflow( name="CEL Ternary Router", steps=[ Router( name="Media Router", selector='input.contains("video") ? "Video Handler" : "Image Handler"', choices=[ Step(name="Video Handler", agent=video_agent), Step(name="Image Handler", agent=image_agent), ], ), ], ) if __name__ == "__main__": print("--- Video request ---") workflow.print_response(input="How do I edit a video for YouTube?") print() print("--- Image request ---") workflow.print_response(input="Help me design a logo for my startup.")

模式解析：

• input.contains("video")命中则走"Video Handler"，否则默认"Image Handler"；
• 这是纯关键字分流，无需额外 LLM 调用，成本最低、延迟最小；代价是只能识别硬编码关键词，无法理解语义（“剪辑”“渲染”等变体需扩展关键词或用前置分类步骤）。

七、用下标引用分支：step_choices让表达式更抗变更

前文所有表达式里都硬编码了分支名。当分支列表频繁增删、或你想避免手写名字造成拼写错误时，可用step_choices按下标位置引用分支。cel_using_step_choices.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) quick_analyzer = Agent( name="Quick Analyzer", model=OpenAIChat(id="gpt-5.6-luna"), instructions="Provide a brief, concise analysis of the topic.", markdown=True, ) detailed_analyzer = Agent( name="Detailed Analyzer", model=OpenAIChat(id="gpt-5.6-luna"), instructions="Provide a comprehensive, in-depth analysis of the topic.", markdown=True, ) workflow = Workflow( name="CEL Step Choices Router", steps=[ Router( name="Analysis Router", # step_choices[0] = "Quick Analysis"（第一个 choice） # step_choices[1] = "Detailed Analysis"（第二个 choice） selector='input.contains("quick") || input.contains("brief") ? step_choices[0] : step_choices[1]', choices=[ Step(name="Quick Analysis", agent=quick_analyzer), Step(name="Detailed Analysis", agent=detailed_analyzer), ], ), ], ) if __name__ == "__main__": print("=== Quick analysis request ===") workflow.print_response( input="Give me a quick overview of quantum computing.", stream=True ) print("\n" + "=" * 50 + "\n") print("=== Detailed analysis request ===") workflow.print_response(input="Explain quantum computing in detail.", stream=True)

机制与适用建议：

• step_choices是当前choices分支名的字符串列表（由evaluate_cel_router_selector注入，见 cel.py），因此step_choices[0]在运行时等于第一个Step.name，step_choices[1]等于第二个；
• 例如input含"quick"或"brief"时走step_choices[0]（Quick Analysis），否则走step_choices[1]；
• 示例还演示了stream=True的流式输出用法；
• 优点正如源码注释所述：避免步骤名拼写错误、提升表达式可维护性、支持按位置动态引用；代价是下标与顺序强耦合，插入或重排分支时必须同步检查表达式语义。

八、运行时解析与容错：selector 结果如何命中分支

理解以上 5 个例子的最终落点，是弄清“selector 算出的字符串如何变成真正被执行的 step”。链路如下（对应同步入口_route_steps，router.py；异步版_aroute_steps逻辑等价，router.py）：

1. 执行Router.execute时先调用_prepare_steps()，把choices里裸的Agent等对象包装为Step，并构建name -> step映射_step_name_map（router.py）；
2. 若selector是字符串，则调用evaluate_cel_router_selector求值，并把所有可选分支名作为step_choices传入；
3. 求值结果交给_resolve_selector_result（router.py）解析：字符串会先在_step_name_map中按名字查找；
4. 未命中的名字不会报错中断，而是记录 warning（Router selector returned unknown step name ...）并返回空列表，导致该轮 Router “完成 0 个结果”（no steps selected）；
5. CEL 求值自身失败（如语法错误、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(name="Branch A", model=OpenAIChat(id="gpt-5.6-luna"), instructions="...") branch_b = Agent(name="Branch B", model=OpenAIChat(id="gpt-5.6-luna"), instructions="...") workflow = Workflow( name="My Router Workflow", steps=[ Router( name="My Router", selector="<CEL 表达式，返回下列某个 Step.name>", choices=[Step(name="Branch A", agent=branch_a), Step(name="Branch B", agent=branch_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.py（Router数据类与执行链路）和 libs/agno/agno/workflow/cel.py（celpy封装、上下文构建、三类求值入口）中，深入阅读可看到 CEL selector 与 callable selector、HITL 选择的完整分支逻辑。

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