agno Workflow 条件分支实战：用 CEL 表达式驱动 Condition 做智能路由

agno Workflow 条件分支实战：用 CEL 表达式驱动 Condition 做智能路由

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

本文围绕 agno（agno）Workflow 的 CEL 条件分支能力展开。你将理解Condition步骤如何用 CEL（Common Expression Language）表达式替代手写判断函数，掌握其暴露的input、previous_step_content、previous_step_outputs、additional_data、session_state五大上下文变量，并基于 cookbook/04_workflows/07_cel_expressions/condition 目录下的 5 个可运行示例，落地「输入内容路由」「按优先级分流」「先分类再路由」「按步骤名条件放行」「基于会话状态的重试」五类典型场景。读完你可以直接把这些表达式写法迁移到自己的 Workflow 中，并能从源码层面理解求值与分支的执行机制。

目录定位与示例全景

该关联文档所在目录是 Workflow 使用 CEL 表达式实现条件执行的专项示例集，位于仓库 cookbook/04_workflows/07_cel_expressions/condition。整个07_cel_expressions模块按用途拆分为三个子目录：

本文档目录下共 5 个可运行示例，逐一对应Condition求值环境中可用的一种上下文变量：

对应本目录的 TEST_LOG.md 记录了这些示例的实测运行日志，可作为行为验证参考。

运行前置条件

按文档说明，运行这些示例需要准备三件事：

1. 激活 demo 虚拟环境：仓库约定使用.venvs/demo/bin/python作为示例解释器；
2. 加载 API Key：执行direnv allow，前提是本地存在.envrc文件（direnv 会在进入目录时自动注入环境变量）；
3. 安装 CEL 支持库：pip install cel-python，所有示例底层依赖celpy。

实际上，每个示例脚本顶部都自带了一次「可用性守卫」：

from agno.workflow import CEL_AVAILABLE, Condition, Step, Workflow if not CEL_AVAILABLE: print("CEL is not available. Install with: pip install cel-python") exit(1)

CEL_AVAILABLE并非硬编码常量，而是由 cel.py 在导入时探测celpy是否安装成功得出的标志位。当cel-python缺失时它会退化为False，示例脚本据此优雅退出；同样的守卫也存在于框架内部——若你在未安装cel-python时强行传入 CEL 表达式，Condition会在求值时记录错误日志并安全返回False（走 else 分支），而不是让整个 Workflow 崩溃（详见 condition.py）。

CEL 在 agno Workflow 中的角色

agno 的 Workflow 由多种步骤类型编排而成，Condition是其中专门负责条件分支的一类。从源码看（condition.py），Condition是一个 dataclass，核心字段如下：

• steps：条件为真时执行的步骤列表；
• evaluator：条件求值器，支持三种形态——返回布尔值的可调用函数、布尔字面量（True/False）、或CEL 表达式字符串，默认值为True；
• else_steps：条件为假且非空时执行的分支（可省略）；
• name/description：步骤命名与描述；
• human_review：人工审核配置（决定分支前的确认时机与拒绝策略）。

在三种 evaluator 形态中，CEL 表达式最具表达力：你无需写 Python 函数，只需一行声明式字符串即可描述分支条件。官方实现里以注释形式列出的五个求值上下文变量（condition.py），正是本文 5 个示例逐一演示的对象：

值得注意的一点是框架对字符串 evaluator 的判定策略。在 cel.py 的is_cel_expression()中：若字符串是纯 Python 标识符（如my_evaluator），会被当作注册表函数名处理；只有包含.、(、比较运算符、逻辑运算符、字面量或引号等特征时，才判定为 CEL 表达式。这决定了你在evaluator里写"priority > 5"是表达式、写"evaluate_priority"是函数引用。

分支语义与条件行为

阅读任何示例前，先明确Condition的执行语义（condition.py）：

• 求值结果为True→ 顺序执行steps（if 分支）；
• 求值结果为False且提供了非空else_steps→ 执行else_steps（else 分支）；
• 求值结果为False且未提供else_steps→ 记录「条件未满足，跳过 N 个步骤」并继续后续流程（跳过而非报错）。

case 4 正是第三种语义的典型：没有 else 分支的Condition起到「门卫 / 关卡」作用，不满足条件就直接放行到下一个步骤。

案例一：按用户输入内容路由（input）

cel_basic.py 演示最直接的一类路由：用input.contains()判断请求是否紧急，紧急走专门的 Agent，否则走常规 Agent。

from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.workflow import CEL_AVAILABLE, Condition, Step, Workflow if not CEL_AVAILABLE: print("CEL is not available. Install with: pip install cel-python") exit(1) urgent_handler = Agent( name="Urgent Handler", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You handle urgent requests with high priority. Be concise and action-oriented.", markdown=True, ) normal_handler = Agent( name="Normal Handler", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You handle normal requests thoroughly and thoughtfully.", markdown=True, ) workflow = Workflow( name="CEL Input Routing", steps=[ Condition( name="Urgent Check", evaluator='input.contains("urgent")', steps=[ Step(name="Handle Urgent", agent=urgent_handler), ], else_steps=[ Step(name="Handle Normal", agent=normal_handler), ], ), ], ) if __name__ == "__main__": print("--- Urgent request ---") workflow.print_response( input="This is an urgent request - please help immediately!" ) print() print("--- Normal request ---") workflow.print_response(input="I have a general question about your services.")

要点拆解：

• input在 CEL 环境中是字符串类型，contains()是 CEL 字符串的标准成员方法（大小写敏感的子串判断）；
• if / else 两个分支各挂一个Step，每个Step绑定一个 Agent；
• 运行时会先后用「urgent 请求」与「普通请求」两段输入验证分支切换，符合文档的「急迫请求 + 普通请求」双路径意图；
• 这里的Agent在运行时会被自动包成Step——Condition._prepare_steps()会将裸 Agent、Team、内嵌 Workflow 与可调用对象统一封装为可执行步骤（condition.py），因此steps里既可以显式写Step，也可以直接放 Agent。

案例二：按附加数据分流（additional_data）

很多时候路由依据并不在用户自然语言里，而在结构化的业务字段中。cel_additional_data.py 通过additional_data.priority数值实现优先级门控：高于 5 走高优先级 Agent，否则走普通 Agent。

from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.workflow import CEL_AVAILABLE, Condition, Step, Workflow if not CEL_AVAILABLE: print("CEL is not available. Install with: pip install cel-python") exit(1) high_priority_agent = Agent( name="High Priority Agent", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You handle high-priority tasks. Be thorough and detailed.", markdown=True, ) low_priority_agent = Agent( name="Low Priority Agent", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You handle standard tasks. Be helpful and concise.", markdown=True, ) workflow = Workflow( name="CEL Priority Routing", steps=[ Condition( name="Priority Gate", evaluator="additional_data.priority > 5", steps=[ Step(name="High Priority", agent=high_priority_agent), ], else_steps=[ Step(name="Low Priority", agent=low_priority_agent), ], ), ], ) if __name__ == "__main__": print("--- High priority (8) ---") workflow.print_response( input="Review this critical security report.", additional_data={"priority": 8}, ) print() print("--- Low priority (2) ---") workflow.print_response( input="Update the FAQ page.", additional_data={"priority": 2}, )

要点拆解：

• 附加数据通过print_response(..., additional_data={"priority": 8})传入，在 CEL 端表现为字典类型的additional_data，因此支持additional_data.priority的点号取字段与> 5数值比较；
• 同一段代码跑出两条对照路径：priority=8 触发高优 Agent，priority=2 落到低优 Agent；
• 该模式同样适用于 CEL 字典的字符串取值，例如additional_data["region"] == "cn"。

案例三：先分类再路由（previous_step_content）

当「用户说什么」不足以下判断时，可以先安排一个专职分类 Agent 产出结构化结论，再由Condition依据上一轮的输出决定去向。cel_previous_step.py 用 Classifier 把请求分成 TECHNICAL / GENERAL 两类后路由：

from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.workflow import CEL_AVAILABLE, Condition, Step, Workflow 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 as either TECHNICAL or GENERAL. " "Respond with exactly one word: TECHNICAL or GENERAL." ), markdown=False, ) technical_agent = Agent( name="Technical Support", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You are a technical support specialist. Provide detailed technical help.", markdown=True, ) general_agent = Agent( name="General Support", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You handle general inquiries. Be friendly and helpful.", markdown=True, ) workflow = Workflow( name="CEL Classify and Route", steps=[ Step(name="Classify", agent=classifier), Condition( name="Route by Classification", evaluator='previous_step_content.contains("TECHNICAL")', steps=[ Step(name="Technical Help", agent=technical_agent), ], else_steps=[ Step(name="General Help", agent=general_agent), ], ), ], ) if __name__ == "__main__": print("--- Technical question ---") workflow.print_response( input="My API returns 500 errors when I send POST requests with JSON payloads." ) print() print("--- General question ---") workflow.print_response(input="What are your business hours?")

要点拆解：

• 这是真正的多步流水线：Step("Classify")先执行，其输出被框架自动写入后续步骤可见的上下文；
• previous_step_content在求值期被绑定为「上一步输出内容」字符串（condition.py 展示了 content 如何从单步输出或步骤列表中取出），因此可直接调用.contains("TECHNICAL")；
• 为了让字符串匹配可靠，Classifier 的 instructions 被刻意约束为“只输出一个词 TECHNICAL 或 GENERAL”，这是 CEL 字符串匹配类条件能够稳定工作的关键工程实践；
• 注意markdown=False用在分类器上、markdown=True用于输出型 Agent，避免格式噪音污染单词语义。

案例四：按名称引用历史步骤的输出（previous_step_outputs）

previous_step_content只能看到“上一步”，而长流水线需要按名回溯任意步骤。cel_previous_step_outputs.py 实现了「研究 → 安全检查（可跳过）→ 发布」的安全发布流水线，用映射类型previous_step_outputs.Research精确引用名为Research的步骤：

from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.workflow import CEL_AVAILABLE, Condition, Step, Workflow if not CEL_AVAILABLE: print("CEL is not available. Install with: pip install cel-python") exit(1) researcher = Agent( name="Researcher", model=OpenAIChat(id="gpt-5.6-luna"), instructions="Research the topic. If the topic involves safety risks, include SAFETY_REVIEW_NEEDED in your response.", markdown=True, ) safety_reviewer = Agent( name="Safety Reviewer", model=OpenAIChat(id="gpt-5.6-luna"), instructions="Review the research for safety concerns and provide recommendations.", markdown=True, ) publisher = Agent( name="Publisher", model=OpenAIChat(id="gpt-5.6-luna"), instructions="Prepare the research for publication.", markdown=True, ) workflow = Workflow( name="CEL Previous Step Outputs Condition", steps=[ Step(name="Research", agent=researcher), Condition( name="Safety Check", # Check the Research step output by name evaluator='previous_step_outputs.Research.contains("SAFETY_REVIEW_NEEDED")', steps=[ Step(name="Safety Review", agent=safety_reviewer), ], ), Step(name="Publish", agent=publisher), ], ) if __name__ == "__main__": print("--- Safe topic (skips safety review) ---") workflow.print_response(input="Write about gardening tips for beginners.") print() print("--- Safety-sensitive topic (triggers safety review) ---") workflow.print_response( input="Write about handling hazardous chemicals in a home lab." )

要点拆解：

• previous_step_outputs是一个「步骤名 → 输出内容字符串」的映射（condition.py 演示了它在步骤链上不断累积更新的机制），CEL 中可用点号按名取字段，等价于previous_step_outputs["Research"]；
• 这里的Condition没有 else_steps，落在第三种语义上：安全主题命中SAFETY_REVIEW_NEEDED时插入一次 Safety Review，普通主题则跳过该分支、直接进入Publish——一个零额外开销的“按需审批”关卡；
• 代码中的注释同样印证了该例的教学意图：# Check the Research step output by name；
• 该模式天然适配多 Agent 协作治理场景，如「内容发布前安全审查」「代码合入前走风险复核」。

案例五：基于会话状态实现重试逻辑（session_state）

最后这个示例把 CEL 与session_state结合，实现跨多次运行累积的计数式重试。cel_session_state.py 定义了三个子步骤组件，并用session_state.retry_count作为分流依据：

from agno.agent import Agent from agno.models.openai import OpenAIChat from agno.run import RunContext from agno.workflow import ( CEL_AVAILABLE, Condition, Step, StepInput, StepOutput, Workflow, ) if not CEL_AVAILABLE: print("CEL is not available. Install with: pip install cel-python") exit(1) def increment_retry_count(step_input: StepInput, run_context: RunContext) -> StepOutput: """Increment retry count in session state.""" current_count = run_context.session_state.get("retry_count", 0) run_context.session_state["retry_count"] = current_count + 1 return StepOutput( content=f"Retry count incremented to {run_context.session_state['retry_count']}", success=True, ) def reset_retry_count(step_input: StepInput, run_context: RunContext) -> StepOutput: """Reset retry count in session state.""" run_context.session_state["retry_count"] = 0 return StepOutput(content="Retry count reset to 0", success=True) retry_agent = Agent( name="Retry Handler", model=OpenAIChat(id="gpt-5.6-luna"), instructions="You are handling a retry attempt. Acknowledge this is a retry and try a different approach.", markdown=True, ) max_retries_agent = Agent( name="Max Retries Handler", model=OpenAIChat(id="gpt-5.6-luna"), instructions="Maximum retries reached. Provide a helpful fallback response and suggest alternatives.", markdown=True, ) workflow = Workflow( name="CEL Retry Logic", steps=[ Step(name="Increment Retry", executor=increment_retry_count), Condition( name="Retry Check", evaluator="session_state.retry_count <= 3", steps=[ Step(name="Attempt Retry", agent=retry_agent), ], else_steps=[ Step(name="Max Retries Reached", agent=max_retries_agent), Step(name="Reset Counter", executor=reset_retry_count), ], ), ], session_state={"retry_count": 0}, ) if __name__ == "__main__": for attempt in range(1, 6): print(f"--- Attempt {attempt} ---") workflow.print_response( input=f"Process request (attempt {attempt})", stream=True, ) print()

要点拆解：

• 工作流通过Workflow(..., session_state={"retry_count": 0})初始化会话状态，session_state变量在 CEL 中表现为字典，因此支持点号取字段与数值比较<=；
• 两个可调用步骤以executor=方式挂载：increment_retry_count接收(StepInput, RunContext)并读写run_context.session_state，每次运行先把计数器加一；这正是「状态 + 条件」闭环的写法——计数变化发生在条件判断之前；
• 主循环连续发起 5 次请求：第 1～3 次retry_count <= 3成立走 Attempt Retry；第 4 次起计数为 4，落入 else 分支触发 Max Retries Reached 并调用reset_retry_count把计数器归零，方便下一轮演示重复执行；
• stream=True表明该流程同样支持流式输出，Condition内部对应提供execute_stream的流式执行实现（condition.py）；
• 相比固定阈值，将重试阈值放入 CEL 表达式的价值在于可配置化与多维度扩展——例如session_state.retry_count < session_state.max_retries && session_state.backoff_seconds < 60，或叠加input内容做加权判断。

源码视角：Condition + CEL 的求值与执行链

把 5 个案例串起来，其背后的调用链完全一致（同步版见 condition.py）：

1. Condition.execute()首先调用_evaluate_condition(step_input, session_state, run_context)；
2. 当evaluator是字符串时，若CEL_AVAILABLE为 False 直接记错误日志并返回False；否则调用evaluate_cel_condition_evaluator(expression, step_input, session_state)（cel.py）；
3. evaluate_cel_condition_evaluator通过_build_step_input_context把input、previous_step_content、previous_step_outputs、additional_data、session_state组装成 CEL 上下文；
4. _evaluate_cel用celpy.Environment()编译并执行表达式，结果统一强转为布尔值（cel.py）；
5. 任何求值异常都会被捕获并按False处理（安全失败：宁可走 else，也不中断工作流）；
6. 依据布尔结果与else_steps的有无确定分支，逐个子步骤串行执行，并将每步输出通过_update_step_input_from_outputs回填到previous_step_content/previous_step_outputs（condition.py），实现跨步骤数据链。

分支内部执行的步骤类型是递归开放的：steps/else_steps里不仅能放普通Step，还能放Steps（顺序组）、Loop、Parallel、嵌套Condition、Router乃至内嵌Workflow（condition.py）。也就是说，5 个案例里呈现的「单层 if/else」可以平滑升级为多级条件树、条件循环与条件并行。

若需在保存工作流配置前预先校验表达式语法，框架还暴露了validate_cel_expression()（cel.py），它会用celpy.Environment()编译但不执行，供 UI / 配置层做准入校验——这是把 CEL 条件能力推向工程化时值得利用的辅助函数。

小结与延伸

本文覆盖的 5 个示例构成了一条由浅入深的学习路径：从「直接看输入」(input)、「读结构化附加数据」(additional_data)，到「消费前一步输出」(previous_step_content)、「按名回溯任意历史步骤输出」(previous_step_outputs)，再到「结合跨运行会话状态做计数/重试」(session_state)。所有表达式都以声明式字符串写在Condition.evaluator上，无需手写 Python 分支函数，且具备一致的安全失败语义。

若希望继续深入 CEL 在 Workflow 中的其余用法，同模块的 loop 子目录 展示了current_iteration、max_iterations、all_success、last_step_content等循环上下文变量的退出条件写法；router 子目录 则展示了返回步骤名（而非布尔值）的 CEL 选择器，以及step_choices、三元表达式等进阶语法。三块组合起来，即可用一套统一的 CEL 语言覆盖 Workflow 中「条件分支、循环退出、路由选择」三类控制流。

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