agno Workflow 条件分支实战用 CEL 表达式驱动 Condition 做智能路由【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本文围绕 agnoagnoWorkflow 的 CEL 条件分支能力展开。你将理解Condition步骤如何用 CELCommon 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模块按用途拆分为三个子目录子目录主题文档condition用 CEL 判断条件并走 if / else 分支本文主体README.mdloop07_cel_expressions/loop用 CEL 作为循环结束条件如cel_iteration_limit.py、cel_content_keyword.pyREADME.mdrouter07_cel_expressions/router用 CEL 作为 Router 选择器返回目标步骤名如cel_ternary.py、cel_using_step_choices.pyREADME.md本文档目录下共 5 个可运行示例逐一对应Condition求值环境中可用的一种上下文变量示例文件核心 CEL 表达式演示的上下文变量cel_basic.pyinput.contains(urgent)inputcel_additional_data.pyadditional_data.priority 5additional_datacel_previous_step.pyprevious_step_content.contains(TECHNICAL)previous_step_contentcel_previous_step_outputs.pyprevious_step_outputs.Research.contains(SAFETY_REVIEW_NEEDED)previous_step_outputscel_session_state.pysession_state.retry_count 3session_state对应本目录的 TEST_LOG.md 记录了这些示例的实测运行日志可作为行为验证参考。运行前置条件按文档说明运行这些示例需要准备三件事激活 demo 虚拟环境仓库约定使用.venvs/demo/bin/python作为示例解释器加载 API Key执行direnv allow前提是本地存在.envrc文件direnv 会在进入目录时自动注入环境变量安装 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.pyCondition是一个 dataclass核心字段如下steps条件为真时执行的步骤列表evaluator条件求值器支持三种形态——返回布尔值的可调用函数、布尔字面量True/False、或CEL 表达式字符串默认值为Trueelse_steps条件为假且非空时执行的分支可省略name/description步骤命名与描述human_review人工审核配置决定分支前的确认时机与拒绝策略。在三种 evaluator 形态中CEL 表达式最具表达力你无需写 Python 函数只需一行声明式字符串即可描述分支条件。官方实现里以注释形式列出的五个求值上下文变量condition.py正是本文 5 个示例逐一演示的对象CEL 变量含义官方示例表达式inputWorkflow 输入字符串input.contains(urgent)session_state会话状态字典session_state.retry_count 3additional_data传给 Workflow 的附加数据字典additional_data.priority 5previous_step_outputs此前各步骤「步骤名 → 内容」映射previous_step_outputs.research.contains(error)previous_step_content上一步骤的输出内容—值得注意的一点是框架对字符串 evaluator 的判定策略。在 cel.py 的is_cel_expression()中若字符串是纯 Python 标识符如my_evaluator会被当作注册表函数名处理只有包含.、(、比较运算符、逻辑运算符、字面量或引号等特征时才判定为 CEL 表达式。这决定了你在evaluator里写priority 5是表达式、写evaluate_priority是函数引用。分支语义与条件行为阅读任何示例前先明确Condition的执行语义condition.py求值结果为True→ 顺序执行stepsif 分支求值结果为False且提供了非空else_steps→ 执行else_stepselse 分支求值结果为False且未提供else_steps→ 记录「条件未满足跳过 N 个步骤」并继续后续流程跳过而非报错。case 4 正是第三种语义的典型没有 else 分支的Condition起到「门卫 / 关卡」作用不满足条件就直接放行到下一个步骤。案例一按用户输入内容路由inputcel_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( nameUrgent Handler, modelOpenAIChat(idgpt-5.6-luna), instructionsYou handle urgent requests with high priority. Be concise and action-oriented., markdownTrue, ) normal_handler Agent( nameNormal Handler, modelOpenAIChat(idgpt-5.6-luna), instructionsYou handle normal requests thoroughly and thoughtfully., markdownTrue, ) workflow Workflow( nameCEL Input Routing, steps[ Condition( nameUrgent Check, evaluatorinput.contains(urgent), steps[ Step(nameHandle Urgent, agenturgent_handler), ], else_steps[ Step(nameHandle Normal, agentnormal_handler), ], ), ], ) if __name__ __main__: print(--- Urgent request ---) workflow.print_response( inputThis is an urgent request - please help immediately! ) print() print(--- Normal request ---) workflow.print_response(inputI 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( nameHigh Priority Agent, modelOpenAIChat(idgpt-5.6-luna), instructionsYou handle high-priority tasks. Be thorough and detailed., markdownTrue, ) low_priority_agent Agent( nameLow Priority Agent, modelOpenAIChat(idgpt-5.6-luna), instructionsYou handle standard tasks. Be helpful and concise., markdownTrue, ) workflow Workflow( nameCEL Priority Routing, steps[ Condition( namePriority Gate, evaluatoradditional_data.priority 5, steps[ Step(nameHigh Priority, agenthigh_priority_agent), ], else_steps[ Step(nameLow Priority, agentlow_priority_agent), ], ), ], ) if __name__ __main__: print(--- High priority (8) ---) workflow.print_response( inputReview this critical security report., additional_data{priority: 8}, ) print() print(--- Low priority (2) ---) workflow.print_response( inputUpdate the FAQ page., additional_data{priority: 2}, )要点拆解附加数据通过print_response(..., additional_data{priority: 8})传入在 CEL 端表现为字典类型的additional_data因此支持additional_data.priority的点号取字段与 5数值比较同一段代码跑出两条对照路径priority8 触发高优 Agentpriority2 落到低优 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( nameClassifier, modelOpenAIChat(idgpt-5.6-luna), instructions( Classify the request as either TECHNICAL or GENERAL. Respond with exactly one word: TECHNICAL or GENERAL. ), markdownFalse, ) technical_agent Agent( nameTechnical Support, modelOpenAIChat(idgpt-5.6-luna), instructionsYou are a technical support specialist. Provide detailed technical help., markdownTrue, ) general_agent Agent( nameGeneral Support, modelOpenAIChat(idgpt-5.6-luna), instructionsYou handle general inquiries. Be friendly and helpful., markdownTrue, ) workflow Workflow( nameCEL Classify and Route, steps[ Step(nameClassify, agentclassifier), Condition( nameRoute by Classification, evaluatorprevious_step_content.contains(TECHNICAL), steps[ Step(nameTechnical Help, agenttechnical_agent), ], else_steps[ Step(nameGeneral Help, agentgeneral_agent), ], ), ], ) if __name__ __main__: print(--- Technical question ---) workflow.print_response( inputMy API returns 500 errors when I send POST requests with JSON payloads. ) print() print(--- General question ---) workflow.print_response(inputWhat are your business hours?)要点拆解这是真正的多步流水线Step(Classify)先执行其输出被框架自动写入后续步骤可见的上下文previous_step_content在求值期被绑定为「上一步输出内容」字符串condition.py 展示了 content 如何从单步输出或步骤列表中取出因此可直接调用.contains(TECHNICAL)为了让字符串匹配可靠Classifier 的 instructions 被刻意约束为“只输出一个词 TECHNICAL 或 GENERAL”这是 CEL 字符串匹配类条件能够稳定工作的关键工程实践注意markdownFalse用在分类器上、markdownTrue用于输出型 Agent避免格式噪音污染单词语义。案例四按名称引用历史步骤的输出previous_step_outputsprevious_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( nameResearcher, modelOpenAIChat(idgpt-5.6-luna), instructionsResearch the topic. If the topic involves safety risks, include SAFETY_REVIEW_NEEDED in your response., markdownTrue, ) safety_reviewer Agent( nameSafety Reviewer, modelOpenAIChat(idgpt-5.6-luna), instructionsReview the research for safety concerns and provide recommendations., markdownTrue, ) publisher Agent( namePublisher, modelOpenAIChat(idgpt-5.6-luna), instructionsPrepare the research for publication., markdownTrue, ) workflow Workflow( nameCEL Previous Step Outputs Condition, steps[ Step(nameResearch, agentresearcher), Condition( nameSafety Check, # Check the Research step output by name evaluatorprevious_step_outputs.Research.contains(SAFETY_REVIEW_NEEDED), steps[ Step(nameSafety Review, agentsafety_reviewer), ], ), Step(namePublish, agentpublisher), ], ) if __name__ __main__: print(--- Safe topic (skips safety review) ---) workflow.print_response(inputWrite about gardening tips for beginners.) print() print(--- Safety-sensitive topic (triggers safety review) ---) workflow.print_response( inputWrite 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( contentfRetry count incremented to {run_context.session_state[retry_count]}, successTrue, ) 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(contentRetry count reset to 0, successTrue) retry_agent Agent( nameRetry Handler, modelOpenAIChat(idgpt-5.6-luna), instructionsYou are handling a retry attempt. Acknowledge this is a retry and try a different approach., markdownTrue, ) max_retries_agent Agent( nameMax Retries Handler, modelOpenAIChat(idgpt-5.6-luna), instructionsMaximum retries reached. Provide a helpful fallback response and suggest alternatives., markdownTrue, ) workflow Workflow( nameCEL Retry Logic, steps[ Step(nameIncrement Retry, executorincrement_retry_count), Condition( nameRetry Check, evaluatorsession_state.retry_count 3, steps[ Step(nameAttempt Retry, agentretry_agent), ], else_steps[ Step(nameMax Retries Reached, agentmax_retries_agent), Step(nameReset Counter, executorreset_retry_count), ], ), ], session_state{retry_count: 0}, ) if __name__ __main__: for attempt in range(1, 6): print(f--- Attempt {attempt} ---) workflow.print_response( inputfProcess request (attempt {attempt}), streamTrue, ) print()要点拆解工作流通过Workflow(..., session_state{retry_count: 0})初始化会话状态session_state变量在 CEL 中表现为字典因此支持点号取字段与数值比较两个可调用步骤以executor方式挂载increment_retry_count接收(StepInput, RunContext)并读写run_context.session_state每次运行先把计数器加一这正是「状态 条件」闭环的写法——计数变化发生在条件判断之前主循环连续发起 5 次请求第 13 次retry_count 3成立走 Attempt Retry第 4 次起计数为 4落入 else 分支触发 Max Retries Reached 并调用reset_retry_count把计数器归零方便下一轮演示重复执行streamTrue表明该流程同样支持流式输出Condition内部对应提供execute_stream的流式执行实现condition.py相比固定阈值将重试阈值放入 CEL 表达式的价值在于可配置化与多维度扩展——例如session_state.retry_count session_state.max_retries session_state.backoff_seconds 60或叠加input内容做加权判断。源码视角Condition CEL 的求值与执行链把 5 个案例串起来其背后的调用链完全一致同步版见 condition.pyCondition.execute()首先调用_evaluate_condition(step_input, session_state, run_context)当evaluator是字符串时若CEL_AVAILABLE为 False 直接记错误日志并返回False否则调用evaluate_cel_condition_evaluator(expression, step_input, session_state)cel.pyevaluate_cel_condition_evaluator通过_build_step_input_context把input、previous_step_content、previous_step_outputs、additional_data、session_state组装成 CEL 上下文_evaluate_cel用celpy.Environment()编译并执行表达式结果统一强转为布尔值cel.py任何求值异常都会被捕获并按False处理安全失败宁可走 else也不中断工作流依据布尔结果与else_steps的有无确定分支逐个子步骤串行执行并将每步输出通过_update_step_input_from_outputs回填到previous_step_content/previous_step_outputscondition.py实现跨步骤数据链。分支内部执行的步骤类型是递归开放的steps/else_steps里不仅能放普通Step还能放Steps顺序组、Loop、Parallel、嵌套Condition、Router乃至内嵌Workflowcondition.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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考