agno 工作流 CEL 表达式完全指南：用 Condition / Loop / Router 实现声明式流程控制

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 如何利用 GoogleCEL（Common 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-python（celpy）包，未安装时模块通过ImportError将CEL_AVAILABLE置为False（见 cel.py）；
• 提供了三个公开求值函数，分别服务于三类控制结构：
  • evaluate_cel_condition_evaluator：求值Condition，结果强转为布尔（cel.py）；
  • evaluate_cel_loop_end_condition：求值Loop的结束条件（cel.py）；
  • evaluate_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）：

1. 安装 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）。

2. 激活 demo 环境：.venvs/demo/bin/python。
3. 加载 API 密钥：通过direnv allow（需要本地.envrc文件）。示例默认使用 OpenAI 模型OpenAIChat(id="gpt-5.6-luna")，可替换为其他 agno 支持的模型。

目录结构如下：

Condition：二分支条件判断（condition/）

Condition是工作流中的"if/else"。当evaluator（CEL 表达式）求值为true时执行steps，否则执行else_steps。构造位置在 libs/agno/agno/workflow/condition.py，用法统一从agno.workflow导入：

from agno.workflow import CEL_AVAILABLE, Condition, Step, Workflow

Condition.evaluator可用的上下文变量（源码注释见 cel.py，实际由_build_step_input_context装配，cel.py）：

1. 基于输入文本路由（cel_basic.py）

cel_basic.py 演示最基础的场景：用input.contains()判断用户请求是否含urgent关键词，紧急请求走urgent_handler，否则走normal_handler：

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), ], ), ], )

分别以 "This is an urgent request - please help immediately!" 与普通问题运行，即可观察两条分支。

2. 基于附加数据路由（cel_additional_data.py）

print_response()等入口支持additional_data参数传递结构化附加信息。 cel_additional_data.py 演示用additional_data.priority做数值比较：优先级 > 5 时走专门的高优 Agent：

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)], )

运行时分别传入additional_data={"priority": 8}与{"priority": 2}，验证两分支。适用前提：表达式取additional_data中的字段前，调用方必须传入相应数据，否则字段不存在会导致求值失败。

3. 基于上一步骤内容路由（cel_previous_step.py）

多步流水线中，"先分类、再分流"是最常见模式。cel_previous_step.py 先用一个只输出单词语种的分类 Agent（TECHNICAL/GENERAL），再用previous_step_content.contains()决定流向：

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)], ), ],

注意分类 Agent 设置了markdown=False，以保证输出是干净的单关键词，便于contains精确匹配。

4. 基于具名步骤输出路由（cel_previous_step_outputs.py）

当流水线存在多个前置步骤时，previous_step_content（仅指"紧邻上一步"）就不够用了，此时用previous_step_outputs.<StepName>按名称访问任意历史步骤输出。cel_previous_step_outputs.py 构造了"调研 → 安全检查（可选）→ 发布"的流水线：

steps=[ Step(name="Research", agent=researcher), Condition( name="Safety Check", # 按名称检查 Research 步骤的输出 evaluator='previous_step_outputs.Research.contains("SAFETY_REVIEW_NEEDED")', steps=[Step(name="Safety Review", agent=safety_reviewer)], ), Step(name="Publish", agent=publisher), ],

Condition未提供else_steps时，条件不满足则直接跳过该组steps继续向下执行——所以普通话题会跳过安全审核直接发布，危险话题则插入 Safety Review。实现上previous_step_outputs键值来自每个StepOutput的step_name字段（cel.py），因此Step 的name必须与表达式中使用的名称完全一致。

5. 基于会话状态实现重试逻辑（cel_session_state.py）

cel_session_state.py 展示如何在Condition中读写session_state，实现带上限的重试逻辑。流程是：一个 Python executor 步骤自增计数器，Condition判断session_state.retry_count <= 3：

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, ) 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}, )

外层循环连续运行 5 次print_response，前 3 次命中重试分支，第 4、5 次命中最大重试分支并触发计数器归零。executor 函数签名(step_input: StepInput, run_context: RunContext) -> StepOutput是本目录示例中自定义步骤的标准形态。

Loop：声明式循环终止条件（loop/）

Loop将一组步骤反复执行，直到满足end_condition（CEL 表达式）或达到max_iterations上限。相关源码位于 libs/agno/agno/workflow/loop.py，求值上下文变量（源码注释见 cel.py，装配逻辑见_build_loop_step_output_context，cel.py）：

all_success的推导逻辑：遍历本轮所有结果，若任一StepOutput.success为假则整体为假（cel.py）。

1. 按迭代次数提前终止（cel_iteration_limit.py）

cel_iteration_limit.py 中max_iterations=10，但end_condition让循环在第 2 次后即结束，展示了"结束条件优先于上限"：

Loop( name="Writing Loop", max_iterations=10, end_condition="current_iteration >= 2", # 上限虽为 10，跑 2 次即停 steps=[Step(name="Write", agent=writer)], )

2. 按输出关键词终止（cel_content_keyword.py）

cel_content_keyword.py 是"Agent 自报完成"的范式：编辑 Agent 被要求"文本打磨完成后在回复末尾包含 DONE"，循环以last_step_content.contains("DONE")作为退出信号：

Loop( name="Editing Loop", max_iterations=5, end_condition='last_step_content.contains("DONE")', steps=[Step(name="Edit", agent=editor)], )

这种方式把"完成度判断"委托给模型自身，代码侧只声明关键词协议，适合文本迭代打磨类任务。

3. 按具名步骤输出终止（cel_step_outputs_check.py）

多步骤循环内可用step_outputs.<StepName>精确引用本轮某个步骤的输出。cel_step_outputs_check.py 让评审 Agent 在认可研究时输出APPROVED，循环在step_outputs.Review含APPROVED时停止：

Loop( name="Research Loop", max_iterations=5, end_condition='step_outputs.Review.contains("APPROVED")', steps=[ Step(name="Research", agent=researcher), Step(name="Review", agent=reviewer), ], )

与previous_step_outputs的差异在于作用域：step_outputs只覆盖当前迭代轮次内产生的步骤输出，不会跨迭代累计。

4. 复合退出条件（cel_compound_exit.py）

CEL 支持&&、||、比较运算符自由组合。cel_compound_exit.py 要求"本轮全部成功且至少迭代 2 次"才退出，形成质量与工作量双重门槛：

Loop( name="Research Loop", max_iterations=5, end_condition="all_success && current_iteration >= 2", steps=[ Step(name="Research", agent=researcher), Step(name="Review", agent=reviewer), ], )

Router：多路选择路由（router/）

Router与Condition的区别在于：Condition是二元（是否）判断，而Router在choices（多个候选 Step）中精确选中一个执行。其selector为 CEL 表达式，求值结果被强转为字符串并当作目标步骤名，因此要求表达式返回分支步骤的 name，且必须与choices中某个Step.name一致。Router从agno.workflow.router导入（实现见 libs/agno/agno/workflow/router.py）：

from agno.workflow import CEL_AVAILABLE, Step, Workflow from agno.workflow.router import Router

selector的上下文变量在 Condition 全部变量基础上增加step_choices（各候选步骤名的列表，由求值函数注入，cel.py）。

1. 三元表达式二选一（cel_ternary.py）

cel_ternary.py 用 CEL 三元运算符cond ? A : B返回步骤名：请求含video选视频专家，否则选图片专家：

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), ], )

2. 用 step_choices 索引引用分支（cel_using_step_choices.py）

直接在表达式中硬编码步骤名容易拼错、难以维护。cel_using_step_choices.py 改用step_choices[0]、step_choices[1]按位置索引引用候选步骤（0 起）：

Router( name="Analysis Router", 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), ], )

文件注释明确总结了这种写法的三个收益：避免步骤名拼写错误、提升表达式可维护性、支持基于索引的动态引用。注意||也是 CEL 合法运算符，可自然组合多个触发词。

3. 前置分类 + 多路路由（cel_previous_step_route.py）

cel_previous_step_route.py 把 Condition 示例中的"分类路由"升级为三路：分类器输出BILLING/TECHNICAL/GENERAL之一，Router 用嵌套三元匹配：

steps=[ Step(name="Classify", agent=classifier), Router( name="Support Router", 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), ], ), ],

嵌套三元即 CEL 版的 if-elif-else：逐级匹配，最后一个字符串兜底。此处按名称引用的是紧邻上一步之外的具名前驱，故用previous_step_outputs.Classify而不是previous_step_content。

4. 其余路由变体

同一路由机制的另外两个应用与本目录 Condition 用例一脉相承，路径与作用如下（其思路可分别对照前面已给出的 Condition 等价代码理解）：

• cel_additional_data_route.py：以additional_data字段作为路由依据，把结构化元数据（如地域、级别、渠道）映射到不同处理 Agent；
• cel_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，记录了各示例的实测运行日志，可作为预期输出参考。

总结：三种控制结构的选型对照

一个实用判断口径：只需判断"是否"用 Condition，需要在多个互斥分支中挑一个用 Router，需要反复执行直到条件满足用 Loop。三类表达式共享input、previous_step_content、previous_step_outputs、additional_data、session_state五元上下文（Router 额外有step_choices），Loop 独享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