openJiuwen agent-core 工作流表达式条件判断 ExpressionCondition 完全指南-平芜编程栈

人工智能
AI Agent
Agent 框架
大模型
工具调用
RAG
提示工程
强化学习

【免费下载链接】agent-core

openJiuwen agent-core可提供AI Agent开发、运行、调优与演进相关的全套SDK能力

项目地址：https://gitcode.com/openJiuwen/agent-core

本篇技术指南围绕 openJiuwen agent-core 工作流组件中的ExpressionCondition表达式条件判断展开，系统讲解其表达式语法、占位符引用、逻辑组合规则、源码级求值原理与安全防护机制，并结合仓库源码与测试用例给出可复用的实战示例。读者将掌握如何在工作流分支路由与循环控制中编写健壮的布尔表达式，并理解其底层安全约束与异常处理方式。

一、模块定位与适用场景

ExpressionCondition是 openjiuwen.core.workflow 模块对外导出的核心条件判断组件之一，完整实现位于 openjiuwen/core/workflow/components/condition/expression.py。它是工作流中"基于字符串表达式进行复杂条件判断"的通用实现：

适合场景：需要将多个条件组合、或进行复杂逻辑判断（多级嵌套、复合表达式）的分支路由与循环控制；
核心特征：支持丰富的表达式语法，可通过${...}占位符动态引用输入字段与变量，适合复杂条件组合。

从类继承关系看，ExpressionCondition继承自Condition（condition.py），而Condition又是AtomicNode（原子节点）的子类，因此它天然具备工作流节点的执行、追踪能力。同一目录下还存在 number.py 与 array.py 等兄弟条件组件，分别面向数值与数组场景，ExpressionCondition则以通用表达式语法覆盖更多场景。

二、快速上手：构造与调用

2.1 导入与构造

from openjiuwen.core.workflow import ExpressionCondition # 构造时传入条件表达式字符串 condition = ExpressionCondition(expression="is_not_empty(${start.query})")

构造参数expression: str即条件表达式字符串，其长度必须满足[0, 5000]。构造阶段会对长度做校验：若超过MAX_EXPRESSION_LENGTH（5000），会抛出EXPRESSION_EVAL_ERROR错误，错误信息为expression length exceeds maximum allowed length of 5000（见 constant.py 与 expression.py）。

>>> # 1. 导入必要类 >>> from openjiuwen.core.workflow import ExpressionCondition >>> >>> # 2. 创建 ExpressionCondition 对象 >>> # 示例1：字符串为空或非空判断 >>> condition1_1 = ExpressionCondition(expression="is_not_empty(${start.query})") >>> condition1_2 = ExpressionCondition(expression="is_empty(${start.query})") >>> >>> # 示例2：字符串相等或不相等 >>> condition2_1 = ExpressionCondition(expression="${start.query} != '1'") >>> condition2_2 = ExpressionCondition(expression="${start.query} == '1'") >>> >>> # 示例3：字符串长度判断 >>> condition3_1 = ExpressionCondition(expression="length(${custom_component.p_str}) > 1") >>> condition3_2 = ExpressionCondition(expression="length(${custom_component.p_str}) == 1") >>> condition3_3 = ExpressionCondition(expression="length(${custom_component.p_str}) != 1") >>> condition3_4 = ExpressionCondition(expression="length(${custom_component.p_str}) < 1") >>> >>> # 示例4：数值比较 >>> condition4_1 = ExpressionCondition(expression="${custom_component.p_int} == 1") >>> condition4_2 = ExpressionCondition(expression="${custom_component.p_int} > 1") >>> condition4_3 = ExpressionCondition(expression="${custom_component.p_int} != 1") >>> >>> # 示例5：布尔值 true/false 判断 >>> condition5_1 = ExpressionCondition(expression="${custom_component.p_true} == true") >>> condition5_2 = ExpressionCondition(expression="${custom_component.p_true} == false") >>> condition5_3 = ExpressionCondition(expression="${custom_component.p_true} != true") >>> condition5_4 = ExpressionCondition(expression="${custom_component.p_true} != false") >>> >>> # 示例6：object 包含或不包含 >>> # 注意：对 object，使用 ${custom_component.p_obj['key']} 访问内部值 >>> condition6_1 = ExpressionCondition(expression="'a' in ${custom_component.p_obj}") >>> condition6_2 = ExpressionCondition(expression="'a' not in ${custom_component.p_obj}") >>> >>> # 示例7：array 包含或不包含 >>> # 注意：对 array，使用 ${custom_component.p_obj[index]} 访问内部元素 >>> condition7_1 = ExpressionCondition(expression="'1' in ${custom_component.p_array}") >>> >>> # 示例8：多级嵌套检查 >>> condition8_1 = ExpressionCondition(expression="${start.p_multi_nested}['level1']['level2']['key3'] > 5") >>> condition8_2 = ExpressionCondition(expression="(len(${start.p_complex}) > 1) && (${start.p_complex}[0]['subkey'] == 'subvalue')") >>> condition8_3 = ExpressionCondition(expression="(${start.p_nested}['level1']['key1'] == 'value1') && ('key2' in ${start.p_nested}['level1'])") >>> condition8_4 = ExpressionCondition(expression="(${start.p_int} > 1) && (${start.p_str} == '2') && (${start.p_array} not in ['3', '4']) && (length(${start.p_obj}) > 0)")

2.2 调用方式

ExpressionCondition支持两种调用形态（见 expression.py）：

invoke(inputs, session)：标准节点调用入口，返回布尔结果；
__call__(session)：直接以可调用对象形式使用，便于嵌入其他组件。

两种入口的逻辑一致：当表达式为空字符串时，直接返回True（视为无条件通过）；否则从会话中解析占位符对应变量后执行求值。

2.3 变量来源与追踪

表达式中${start.query}这类占位符代表"需要引用的输入字段或变量"，其值通过session.state().get_global(...)从会话全局状态中取出（_get_inputs方法，expression.py）。组件还实现了trace_info方法，返回{"bool_expression": 表达式原文, "inputs": 解析出的变量映射}，便于工作流运行时追踪分支判断依据。

三、表达式语法详解

3.1 支持的判断类型

判断类型	说明
大于 / 大于等于 / 小于 / 小于等于	数值与字符串大小比较，如`>`,`>=`,`<`,`<=`
等于 / 不等于	值是否相同，其中"等于"还包含数组长度判断（与`length()`/`len()`配合）
包含 / 不包含	元素是否在集合（数组、对象）中，如`in`,`not in`
为空 / 不为空	字符串长度是否为 0 或变量是否存在，如`is_empty`,`is_not_empty`

3.2 支持的数据类型

表达式可以操作的数据类型包括：string、number、integer、boolean、object、array，覆盖工作流节点间传递的常见数据结构。

3.3 占位符语法：动态引用

在需要动态引用的场景中，使用${...}占位符构建动态布尔表达式：

${与}分别是占位符的起始与结束符号；
start.query是占位符内容，代表需要引用的输入字段或变量，如${start.query}、${custom_component.p_int}、${start.p_multi_nested}。

3.4 逻辑判断符号表

类型	描述	示例
`&&, and, \\|\\|, or`	AND、OR 连接，连接多个布尔表达式（复合表达式，每个子表达式需要括号）	`(${start.query} == "test_query") && ("query" not in ${start.var})`
`==, !=`	字符串比较，判断两个字符串相同或不同	`${start.query} == "test_query"`
`in, not_in, not in`	字符串比较，判断一个字符串在或不在另一个字符串/集合中	`"query" in ${start.var}`
`is_empty`	判断字符串为空、长度为 0 或不存在	`is_empty(${start.query})`
`is_not_empty`	判断字符串不为空	`is_not_empty(${start.query})`
`length, len, >=, >, <=, <`	字符串长度比较	`length(${start.query}) >= 1`

3.5 表达式预处理器：语法归一化

在求值之前，框架会通过convert_condition对表达式做一层"语法归一化"（expression.py），其转换规则RULES如下：

原始写法	转换后
`&&`	`and`
`\\|\\|`	`or`
`true`（独立单词）	`True`
`false`（独立单词）	`False`
`length(`	`len(`
`not_in`	`not in`
`is_empty(`	`_safe_is_empty(`
`is_not_empty(`	`_safe_is_not_empty(`

特别地，true/false的替换带有词边界（\b），避免误伤类似is_false_alarm这样的标识符。整个替换过程会先提取并保护字符串字面量（含转义引号处理），完成规则替换后再还原，从而保证字符串内部内容不会被误替换。

四、源码级原理：表达式如何被求值

ExpressionCondition._evaluate_expression的完整求值链路如下（expression.py）：

预处理：调用convert_condition完成语法归一化；
占位符绑定：通过正则\$\{([^{}]*)\}提取所有占位符，按长度从长到短替换为安全变量名var_0, var_1, ...（长占位符优先替换，避免嵌套变量被提前截断），并将占位符原文到值的映射放入运行时环境；
AST 解析：使用 Python 标准库ast.parse(expr, mode='eval')将表达式解析为抽象语法树；
深度检查：调用_check_ast_depth递归检查 AST 嵌套深度，超过MAX_AST_DEPTH（50）即抛出求值错误（见 constant.py）；
安全求值：调用_evaluate_ast递归解释执行 AST，而不是使用 Python 的eval()内建函数，因此只暴露白名单内的运行时能力。

4.1 白名单运行时

求值仅向表达式暴露以下能力（runtime 环境）：len、bool、not、and、or、in、sum、inputs，以及is_empty/is_not_empty/_safe_is_empty/_safe_is_not_empty安全函数。未定义的名称会通过_evaluate_name被解析为字面量（如数字、True、False、None），否则抛出EXPRESSION_EVAL_ERROR。

4.2 多级嵌套与下标访问

_evaluate_ast支持BoolOp（and/or）、Compare（比较）、Subscript（下标）、Attribute（属性）、Call（函数调用）、List/Tuple/Dict（集合字面量）、UnaryOp（一元运算）、BinOp（二元运算）等节点类型，因此可以支持文章示例 8 中的多级嵌套访问（如${start.p_multi_nested}['level1']['level2']['key3'] > 5）。同时_evaluate_attribute在处理对象属性访问失败时会回退到字典键访问（value[node.attr]），兼容 object 与 dict 混用的场景。

五、安全防护机制（源码佐证）

ExpressionCondition并不是简单地把表达式交给eval()，而是实现了多层安全防护（这是它与直接 Python 求值方案的关键区别）：

敏感属性拦截：禁止访问__class__、__bases__、__subclasses__、__module__、__dict__等敏感属性，以及任何以__开头和结尾的 dunder 属性（expression.py）；
集合大小上限：MAX_COLLECTION_SIZE = 100000（constant.py），求值前对列表字面量、字典字面量、下标访问的集合做大小校验，防止构造超大集合；
危险运算拦截：乘法运算会校验[0] * N这类可能产生超大集合的运算（len * right > MAX_COLLECTION_SIZE时拒绝）；幂运算的指数超过 100 会被拒绝，防止天文数字级别的计算耗尽资源；
AST 深度限制：MAX_AST_DEPTH = 50，防止通过极深嵌套构造攻击性表达式；
切片越界保护：slice(lower, upper)的区间差超过MAX_COLLECTION_SIZE时拒绝执行。

这些防护在仓库的测试用例 tests/unit_tests/core/condition/test_condition.py 中有明确覆盖，例如is_not_empty([0] * (10 ** 10))这类会导致超大集合的表达式会被安全拒绝。

六、异常与错误码

ExpressionCondition求值失败时统一抛出BaseError，具体的错误码定义于 openjiuwen/core/common/exception/codes.py：

错误码	含义
`101024`	expression syntax error（表达式语法错误，如括号不匹配、非法符号）
`101025`	expression evaluation error（表达式求值错误，错误信息中携带具体原因）

这两类错误码均属于StatusCode枚举的一部分。更完整的错误码说明与排查建议参见 StatusCode 文档（中文版）。求值阶段抛出的异常类型（SyntaxError、NameError、其他Exception）都会被捕获并统一包装为携带原始原因的BaseError，方便上层统一处理。

七、在流程组件中的实际使用

ExpressionCondition不仅可独立使用，也是工作流内置组件的重要底层能力：

分支路由器 BranchRouter（branch_router.py）：add_branch接受Union[str, Callable, Condition]类型的条件，当传入字符串时，会自动包装为ExpressionCondition(condition)。这意味着用户可以在分支配置中直接写表达式字符串，例如"is_not_empty(${start.query})"，而无需手动实例化条件对象；
循环组件 LoopComp（loop_comp.py）：循环终止/继续的条件同样可以使用表达式字符串构造ExpressionCondition，从而支持基于运行上下文动态控制循环次数与退出条件。

八、测试验证与最佳实践

仓库测试文件 tests/unit_tests/core/condition/test_condition.py 中TestExpressionCondition覆盖了初始化、空表达式（返回True）、trace_info、复合表达式、多级嵌套、超大集合防护等多种场景，可作为表达式编写正确性的参照。

基于源码与文档总结的实战建议：

复合表达式务必加括号：&&/||连接多个子表达式时，每个子表达式需要用括号包裹，例如(${a} == '1') && (${b} != '2')；
优先使用语义化函数：空值判断优先用is_empty/is_not_empty，它们对None安全（_safe_is_empty将None视为空），但注意：对int、float、bool这类非集合类型调用is_empty会抛出求值错误，应避免对数值使用空值判断；
长度判断用length()或len()：二者等价（length(会被预处理器改写为len(），且len对None返回 0，不会抛异常；
对象/数组内部取值用下标：对象用['key']，数组用[index]，且支持多级嵌套；
注意表达式长度与复杂度上限：表达式长度不超过 5000 字符，AST 嵌套深度不超过 50 层，集合字面量不超过 100000 个元素，编写超复杂表达式时需预留校验空间；
优先使用分支路由器的字符串快捷方式：在BranchRouter.add_branch中直接写表达式字符串，代码更简洁且自动获得与ExpressionCondition一致的安全求值能力。