agno 工作流 CEL 表达式完全指南用 Condition / Loop / Router 实现声明式流程控制【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno本文以 cookbook/04_workflows/07_cel_expressions 目录下的 14 个可运行示例为骨架系统讲解 agno Workflow 如何利用 GoogleCELCommon Expression Language表达式为Condition条件分支、Loop循环终止与Router多路路由三种控制结构编写声明式判断逻辑。读完本文你将掌握 CEL 在 agno 工作流中的全部上下文变量、求值函数、错误处理与运行前置条件并能直接运行仓库内示例复现路由、重试、审核循环等经典场景。CEL 是什么为什么工作流需要它在 agno 的Workflow中Step步骤定义做什么而流程的控制下一步走哪条路、循环何时停止、多路选择中选哪个分支则属于流程逻辑。传统做法是使用 Python 函数作为 evaluator/selector而 CEL 允许你用纯字符串表达式来编写这些判断例如input.contains(urgent) additional_data.priority 5 all_success current_iteration 2由于表达式只是字符串它们可以被安全持久化、动态下发、在 UI 中校验甚至由非 Python 组件生成这正是 agno 在工作流控制点引入 CEL 的核心动机。从库实现看CEL 支持封装在 libs/agno/agno/workflow/cel.py 中依赖可选的cel-pythoncelpy包未安装时模块通过ImportError将CEL_AVAILABLE置为False见 cel.py提供了三个公开求值函数分别服务于三类控制结构evaluate_cel_condition_evaluator求值Condition结果强转为布尔cel.pyevaluate_cel_loop_end_condition求值Loop的结束条件cel.pyevaluate_cel_router_selector求值Router的选择器结果强转为字符串步骤名cel.py另有validate_cel_expression仅编译不执行可用于保存配置前的 UI 校验与is_cel_expression自动区分函数名与CEL 表达式两个工具函数。CEL 求值统一走celpy.Environment()的compile → program → evaluate管道Python 的bool/int/float/str/list/dict/None会被_to_cel逐层转换为对应 CEL 类型其余类型则回退为字符串cel.py。运行前置条件三个子目录的 README 均声明了相同的前置条件示例见 condition/README.md安装 CEL 运行时pip install cel-python。每个示例脚本开头都有防御性检查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)运行时若缺少依赖实际求值会抛出RuntimeError: cel-python is not installed见 cel.py。激活 demo 环境.venvs/demo/bin/python。加载 API 密钥通过direnv allow需要本地.envrc文件。示例默认使用 OpenAI 模型OpenAIChat(idgpt-5.6-luna)可替换为其他 agno 支持的模型。目录结构如下子目录控制结构覆盖知识点condition二元 if/else 分支input、additional_data、previous step、session_state 判定loop循环终止判定迭代次数、内容关键词、步骤输出、复合条件退出router多路选择路由三元运算符、step_choices 索引、多分支分类路由Condition二分支条件判断condition/Condition是工作流中的if/else。当evaluatorCEL 表达式求值为true时执行steps否则执行else_steps。构造位置在 libs/agno/agno/workflow/condition.py用法统一从agno.workflow导入from agno.workflow import CEL_AVAILABLE, Condition, Step, WorkflowCondition.evaluator可用的上下文变量源码注释见 cel.py实际由_build_step_input_context装配cel.py变量类型含义取值来源inputstring工作流本次输入的文本step_input.input非字符串会先转字符串previous_step_contentstring上一步骤输出的内容step_input.previous_step_contentprevious_step_outputsmap所有历史步骤名称 → 内容的映射遍历previous_step_outputs取各contentadditional_datamap随输入传入的附加结构化数据step_input.additional_datasession_statemap当前会话状态由工作流/运行层维护并单独传入1. 基于输入文本路由cel_basic.pycel_basic.py 演示最基础的场景用input.contains()判断用户请求是否含urgent关键词紧急请求走urgent_handler否则走normal_handlerworkflow Workflow( nameCEL Input Routing, steps[ Condition( nameUrgent Check, evaluatorinput.contains(urgent), steps[ Step(nameHandle Urgent, agenturgent_handler), ], else_steps[ Step(nameHandle Normal, agentnormal_handler), ], ), ], )分别以 This is an urgent request - please help immediately! 与普通问题运行即可观察两条分支。2. 基于附加数据路由cel_additional_data.pyprint_response()等入口支持additional_data参数传递结构化附加信息。 cel_additional_data.py 演示用additional_data.priority做数值比较优先级 5 时走专门的高优 AgentCondition( namePriority Gate, evaluatoradditional_data.priority 5, steps[Step(nameHigh Priority, agenthigh_priority_agent)], else_steps[Step(nameLow Priority, agentlow_priority_agent)], )运行时分别传入additional_data{priority: 8}与{priority: 2}验证两分支。适用前提表达式取additional_data中的字段前调用方必须传入相应数据否则字段不存在会导致求值失败。3. 基于上一步骤内容路由cel_previous_step.py多步流水线中先分类、再分流是最常见模式。cel_previous_step.py 先用一个只输出单词语种的分类 AgentTECHNICAL/GENERAL再用previous_step_content.contains()决定流向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)], ), ],注意分类 Agent 设置了markdownFalse以保证输出是干净的单关键词便于contains精确匹配。4. 基于具名步骤输出路由cel_previous_step_outputs.py当流水线存在多个前置步骤时previous_step_content仅指紧邻上一步就不够用了此时用previous_step_outputs.StepName按名称访问任意历史步骤输出。cel_previous_step_outputs.py 构造了调研 → 安全检查可选→ 发布的流水线steps[ Step(nameResearch, agentresearcher), Condition( nameSafety Check, # 按名称检查 Research 步骤的输出 evaluatorprevious_step_outputs.Research.contains(SAFETY_REVIEW_NEEDED), steps[Step(nameSafety Review, agentsafety_reviewer)], ), Step(namePublish, agentpublisher), ],Condition未提供else_steps时条件不满足则直接跳过该组steps继续向下执行——所以普通话题会跳过安全审核直接发布危险话题则插入 Safety Review。实现上previous_step_outputs键值来自每个StepOutput的step_name字段cel.py因此Step 的name必须与表达式中使用的名称完全一致。5. 基于会话状态实现重试逻辑cel_session_state.pycel_session_state.py 展示如何在Condition中读写session_state实现带上限的重试逻辑。流程是一个 Python executor 步骤自增计数器Condition判断session_state.retry_count 3def 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, ) 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}, )外层循环连续运行 5 次print_response前 3 次命中重试分支第 4、5 次命中最大重试分支并触发计数器归零。executor 函数签名(step_input: StepInput, run_context: RunContext) - StepOutput是本目录示例中自定义步骤的标准形态。Loop声明式循环终止条件loop/Loop将一组步骤反复执行直到满足end_conditionCEL 表达式或达到max_iterations上限。相关源码位于 libs/agno/agno/workflow/loop.py求值上下文变量源码注释见 cel.py装配逻辑见_build_loop_step_output_contextcel.py变量类型含义current_iterationint当前迭代号1 起本轮结束后计值max_iterationsintLoop 配置的最大迭代次数all_successbool本轮所有步骤是否全部成功由各StepOutput.success汇合last_step_contentstring本轮最后一个步骤的输出内容step_outputsmap本轮内步骤名 → 输出内容的映射all_success的推导逻辑遍历本轮所有结果若任一StepOutput.success为假则整体为假cel.py。1. 按迭代次数提前终止cel_iteration_limit.pycel_iteration_limit.py 中max_iterations10但end_condition让循环在第 2 次后即结束展示了结束条件优先于上限Loop( nameWriting Loop, max_iterations10, end_conditioncurrent_iteration 2, # 上限虽为 10跑 2 次即停 steps[Step(nameWrite, agentwriter)], )2. 按输出关键词终止cel_content_keyword.pycel_content_keyword.py 是Agent 自报完成的范式编辑 Agent 被要求文本打磨完成后在回复末尾包含 DONE循环以last_step_content.contains(DONE)作为退出信号Loop( nameEditing Loop, max_iterations5, end_conditionlast_step_content.contains(DONE), steps[Step(nameEdit, agenteditor)], )这种方式把完成度判断委托给模型自身代码侧只声明关键词协议适合文本迭代打磨类任务。3. 按具名步骤输出终止cel_step_outputs_check.py多步骤循环内可用step_outputs.StepName精确引用本轮某个步骤的输出。cel_step_outputs_check.py 让评审 Agent 在认可研究时输出APPROVED循环在step_outputs.Review含APPROVED时停止Loop( nameResearch Loop, max_iterations5, end_conditionstep_outputs.Review.contains(APPROVED), steps[ Step(nameResearch, agentresearcher), Step(nameReview, agentreviewer), ], )与previous_step_outputs的差异在于作用域step_outputs只覆盖当前迭代轮次内产生的步骤输出不会跨迭代累计。4. 复合退出条件cel_compound_exit.pyCEL 支持、||、比较运算符自由组合。cel_compound_exit.py 要求本轮全部成功且至少迭代 2 次才退出形成质量与工作量双重门槛Loop( nameResearch Loop, max_iterations5, end_conditionall_success current_iteration 2, steps[ Step(nameResearch, agentresearcher), Step(nameReview, agentreviewer), ], )Router多路选择路由router/Router与Condition的区别在于Condition是二元是否判断而Router在choices多个候选 Step中精确选中一个执行。其selector为 CEL 表达式求值结果被强转为字符串并当作目标步骤名因此要求表达式返回分支步骤的 name且必须与choices中某个Step.name一致。Router从agno.workflow.router导入实现见 libs/agno/agno/workflow/router.pyfrom agno.workflow import CEL_AVAILABLE, Step, Workflow from agno.workflow.router import Routerselector的上下文变量在 Condition 全部变量基础上增加step_choices各候选步骤名的列表由求值函数注入cel.py。1. 三元表达式二选一cel_ternary.pycel_ternary.py 用 CEL 三元运算符cond ? A : B返回步骤名请求含video选视频专家否则选图片专家Router( nameMedia Router, selectorinput.contains(video) ? Video Handler : Image Handler, choices[ Step(nameVideo Handler, agentvideo_agent), Step(nameImage Handler, agentimage_agent), ], )2. 用 step_choices 索引引用分支cel_using_step_choices.py直接在表达式中硬编码步骤名容易拼错、难以维护。cel_using_step_choices.py 改用step_choices[0]、step_choices[1]按位置索引引用候选步骤0 起Router( nameAnalysis Router, selectorinput.contains(quick) || input.contains(brief) ? step_choices[0] : step_choices[1], choices[ Step(nameQuick Analysis, agentquick_analyzer), Step(nameDetailed Analysis, agentdetailed_analyzer), ], )文件注释明确总结了这种写法的三个收益避免步骤名拼写错误、提升表达式可维护性、支持基于索引的动态引用。注意||也是 CEL 合法运算符可自然组合多个触发词。3. 前置分类 多路路由cel_previous_step_route.pycel_previous_step_route.py 把 Condition 示例中的分类路由升级为三路分类器输出BILLING/TECHNICAL/GENERAL之一Router 用嵌套三元匹配steps[ Step(nameClassify, agentclassifier), Router( nameSupport Router, 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), ], ), ],嵌套三元即 CEL 版的 if-elif-else逐级匹配最后一个字符串兜底。此处按名称引用的是紧邻上一步之外的具名前驱故用previous_step_outputs.Classify而不是previous_step_content。4. 其余路由变体同一路由机制的另外两个应用与本目录 Condition 用例一脉相承路径与作用如下其思路可分别对照前面已给出的 Condition 等价代码理解cel_additional_data_route.py以additional_data字段作为路由依据把结构化元数据如地域、级别、渠道映射到不同处理 Agentcel_session_state_route.py结合session_state与会话级变量路由可依据历史尝试次数、用户上下文等跨请求状态选择策略。运行与验证每个示例均可用 demo 环境直接运行。以 condition 目录为例# 使用 demo 虚拟环境预先 direnv allow 加载 .envrc 中的 API 密钥 .venvs/demo/bin/python cookbook/04_workflows/07_cel_expressions/condition/cel_basic.py .venvs/demo/bin/python cookbook/04_workflows/07_cel_expressions/loop/cel_content_keyword.py .venvs/demo/bin/python cookbook/04_workflows/07_cel_expressions/router/cel_ternary.py运行前应确认满足三项前置cel-python已安装否则CEL_AVAILABLE为False直接退出、已加载模型 API 密钥、模型 id 可用。目录内还包含 TEST_LOG.md记录了各示例的实测运行日志可作为预期输出参考。总结三种控制结构的选型对照控制结构导入路径求值结果表达方式典型场景Conditionagno.workflow布尔分支或跳过steps 可选else_steps紧急度分流、审核门禁、重试上限、会话状态判断Loopagno.workflow布尔是否退出end_conditionmax_iterations迭代写作、质量循环、关键词/评审批准退出Routeragno.workflow.router字符串步骤名selectorchoices多级客服分类、媒体类型路由、按元数据分发一个实用判断口径只需判断是否用 Condition需要在多个互斥分支中挑一个用 Router需要反复执行直到条件满足用 Loop。三类表达式共享input、previous_step_content、previous_step_outputs、additional_data、session_state五元上下文Router 额外有step_choicesLoop 独享current_iteration、max_iterations、all_success、last_step_content、step_outputs迭代上下文。掌握这组变量就能把绝大多数流程控制需求改写成可持久化、可校验、可动态生成的 CEL 声明。【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考