在实际业务系统中条件工作流并不是一个陌生概念。无论是审批流转、订单处理还是任务编排系统都需要根据某个字段值或运算结果决定下一步执行哪个节点。常见实现有两种思路一种是在代码里用 if else 硬编码流程另一种是把流程定义做成可配置的数据结构。后一种方式里比较有争议的问题就是流程定义到底要不要引入类型系统。这里讨论的“条件工作流的无类型写法”就是使用 JSON、YAML 这类不携带类型声明的数据格式来定义流程节点、分支条件和跳转关系让流程引擎在运行时动态解释并执行。这种方式在轻量级工作流、规则引擎前端配置、运营后台配置化场景里非常常见。这篇文章会围绕一个最小可运行的订单审批条件工作流讲解无类型写法的定义方式、Python 执行引擎的实现思路、条件表达式的解析方法、常见排错链路以及从学习环境过渡到生产环境时需要补齐的能力。读完以后可以直接照着代码实现一个不依赖大型工作流框架、用 JSON 描述分支逻辑的简单执行引擎。1. 条件工作流为什么需要“无类型写法”1.1 从硬编码到流程定义早期系统的流程控制通常直接写在业务代码里。一个订单审批逻辑可能是这样的if (order.getAmount() 10000) { managerService.approve(order); } else { autoService.approve(order); }这种写法的优点是简单直接编译器能检查类型逻辑改动在代码评审里可以看到。但缺点也很明显业务流程一旦频繁调整比如金额阈值从 10000 改成 5000或者增加一个风险订单拦截节点就需要改代码、重新部署、再走一遍发布流程。对于运营、客服、风控团队来说这种节奏太慢。流程定义化则是把“顺序、分支、条件、动作”从代码中抽离出来放到可配置的数据文件或数据库表里。系统启动时加载这些定义运行时根据定义执行。这样流程变更就不需要改代码只需要更新配置并刷新缓存。1.2 无类型写法的含义与边界“无类型”这个词在不同语境下有不同含义。在编程语言层面无类型通常指不强制声明变量类型Python、JavaScript 都属于动态类型语言变量可以随时指向不同类型的数据。在工作流配置层面无类型写法更多指流程定义文件本身不引入 XML Schema、Protobuf、Thrift 这类强类型描述而是直接用 JSON、YAML 或 Map 结构描述节点和条件。例如一个条件节点在无类型写法里可能是{ id: check_amount, type: condition, condition: order.amount 10000, trueNext: manager_approve, falseNext: auto_approve }这里没有提前声明order.amount是整数还是字符串。它的实际类型由运行时传入的变量上下文决定。如果调用方传入的是10000那么条件求值时就会按数字比较如果传入的是10000那么引擎要么做类型转换要么按字符串比较具体行为取决于引擎实现。这种写法的核心价值是降低配置门槛让非开发人员也能看懂大部分分支条件。代价是类型安全被削弱很多错误要等到运行时才暴露。1.3 无类型与有类型选型对比实际项目中究竟该用无类型配置还是引入有类型 DSL需要结合团队情况和变更频率来判断。对比维度无类型 JSON/YAML 配置有类型 DSL 或代码定义表示形式JSON、YAML、Map无字段类型声明自定义 DSL、Protobuf、Java 类、TypeScript 类型修改成本改配置即可发布流程简单改代码或改 DSL 编译产物流程较重类型安全性弱运行时才暴露类型错误强编译期或加载期可校验学习成本低运营和产品可读性强高需要维护 DSL 语法和运行时性能需在运行时解析表达式略有损耗可预编译执行效率更高适用场景中小团队、轻量编排、运营配置大型平台、需要长期维护的复杂编排无类型写法并不是为了替代有类型方案而是为了解决“流程经常变、又不希望每次变更都发版本”这一类问题。等流程数量和复杂度增长到一定规模后再考虑引入 schema 校验和编译期检查也是一种平滑演进路径。2. 设计一份最小无类型条件工作流定义2.1 JSON 结构节点、连接和条件为了让执行引擎足够简单又具备代表性这里把工作流定义收敛成四种节点action执行一个动作通常是设置变量或调用外部服务然后跳转到下一个节点。condition对变量上下文求值一个条件表达式结果为真走trueNext为假走falseNext。end流程结束返回一个结果标记。start其实可以用普通action节点表示但显式命名会让结构更清晰。一个节点至少包含id和type两个字段。condition节点必须包含condition、trueNext、falseNext。action节点可以包含set用于在上下文中写入变量。{ name: 订单审批条件工作流, maxSteps: 30, nodes: [ { id: start, type: action, description: 接收订单, set: { order.status: pending }, next: check_amount }, { id: check_amount, type: condition, condition: order.amount 10000, trueNext: manager_approve, falseNext: auto_approve }, { id: manager_approve, type: action, description: 经理审批, set: { order.status: approved_by_manager }, next: end }, { id: auto_approve, type: action, description: 系统自动通过, set: { order.status: auto_approved }, next: end }, { id: end, type: end, result: order_processed } ] }这份定义里没有任何类型声明。order.amount是数字还是字符串要到引擎执行时才确定。set里的值也没有声明类型配置里写什么类型上下文里就是什么类型。这种设计在小型项目中非常灵活但也会带来一个隐患结构没有任何约束一旦某个节点漏写next字段引擎可能把流程断了。后面会专门讨论如何通过校验和检查清单防住这类问题。2.2 条件表达式语法设计条件表达式是整个工作流里最关键的部分。无类型写法不要求使用者声明类型但表达式语法必须足够简单才能避免直接使用eval带来的安全风险。最小引擎里支持的条件表达式语法如下比较运算、!、、、、逻辑运算and、or单值判断直接写变量路径或布尔值字面量数字、字符串、布尔值示例order.amount 10000 order.amount 5000 and order.status pending order.risk true or order.level high这里刻意不实现括号嵌套。原因是解析器会显著变复杂而对大多数审批、流转场景来说and和or的从左到右组合已经足够。如果确实需要复杂逻辑推荐的做法是拆成多个条件节点而不是把表达式写得越来越长。表达式里的空格不是必须的但强烈建议保留。解析器依赖空格来区分变量名、操作符和字面量。手写配置时统一使用半角空格可以避免很多解析问题。2.3 变量上下文与点路径执行引擎在运行过程中维护一个variables上下文对象。这个对象可以是任意嵌套的字典条件表达式通过点路径访问属性order.amount表示从根上下文取order再取amount。user.id表示从根上下文取user的id。config.enabled表示取config的enabled。在 Python 里点路径可以这样解析class Context: def __init__(self, variables): self.variables variables def get(self, path): parts path.split(.) current self.variables for part in parts: if not isinstance(current, dict) or part not in current: raise WorkflowRuntimeError(f变量路径不存在: {path}) current current[part] return current这种点路径设计的好处是配置可读性好运营人员可以直接看懂条件是在比较哪个字段。坏处是路径写错时错误信息需要足够清晰否则排错成本很高。3. 用 Python 实现一个最小执行引擎3.1 项目结构与依赖本文的示例只需要 Python 3.8 以上环境不依赖第三方库。项目结构如下condition-workflow/ ├── workflow.json ├── engine.py └── main.py三个文件的职责workflow.json工作流定义用无类型 JSON 描述节点和条件。engine.py执行引擎负责加载定义、解析节点、求值条件、维护节点跳转。main.py入口脚本构造变量上下文运行工作流并输出结果。3.2 加载与解析工作流定义在engine.py中先定义异常类型和上下文类import json class WorkflowRuntimeError(Exception): pass class Context: def __init__(self, variables): self.variables variables def get(self, path): parts path.split(.) current self.variables for part in parts: if not isinstance(current, dict) or part not in current: raise WorkflowRuntimeError(f变量路径不存在: {path}) current current[part] return current def set(self, path, value): parts path.split(.) current self.variables for part in parts[:-1]: if part not in current: current[part] {} if not isinstance(current[part], dict): raise WorkflowRuntimeError(f无法在非对象路径上设置变量: {path}) current current[part] current[parts[-1]] value上下文类只做两件事读变量和写变量。读变量时路径不存在会抛出异常这样配置错误可以在第一时间暴露。写变量时支持点路径例如order.status会写入到variables[order][status]。3.3 变量上下文与条件求值条件求值是引擎的核心也是最需要谨慎的部分。这里不使用eval而是实现一个小型解析器。先处理逻辑运算再处理比较运算def _split_top_level(expr, sep): parts [] start 0 depth 0 quote None i 0 while i len(expr): ch expr[i] if quote: if ch quote: quote None i 1 continue if ch in (, ): quote ch i 1 continue if ch (: depth 1 elif ch ): depth - 1 if depth 0 and expr.startswith(sep, i): parts.append(expr[start:i].strip()) start i len(sep) i len(sep) continue i 1 parts.append(expr[start:].strip()) return parts def _resolve_value(token, ctx): token token.strip() if token true: return True if token false: return False if len(token) 2 and token[0] in (, ) and token[-1] token[0]: return token[1:-1] if token.lstrip(-).isdigit(): return int(token) return ctx.get(token) def evaluate_condition(expr, ctx): expr expr.strip() if not expr: raise WorkflowRuntimeError(条件表达式不能为空) for op in ( or , and ): if op in expr: parts _split_top_level(expr, op.strip()) results [evaluate_condition(p, ctx) for p in parts] if op or : return any(results) return all(results) for operator in (, , , !, , ): # 从表达式里查找第一个比较运算符 idx _find_operator(expr, operator) if idx 0: left expr[:idx].strip() right expr[idx len(operator):].strip() left_val _resolve_value(left, ctx) right_val _resolve_value(right, ctx) return _compare(left_val, operator, right_val) # 没有比较运算符时按布尔值处理 return bool(_resolve_value(expr, ctx)) def _find_operator(expr, operator): quote None i 0 while i len(expr): ch expr[i] if quote: if ch quote: quote None i 1 continue if ch in (, ): quote ch i 1 continue if expr.startswith(operator, i): return i i 1 return -1 def _compare(left_val, operator, right_val): if operator : return left_val right_val if operator !: return left_val ! right_val if operator : return left_val right_val if operator : return left_val right_val if operator : return left_val right_val if operator : return left_val right_val raise WorkflowRuntimeError(f不支持的操作符: {operator})这段代码有几个关键点_split_top_level会跳过引号内的分隔符避免字符串里的and被误切。_resolve_value根据 token 形态决定它是字面量还是变量路径。比较运算符的查找也会跳过引号避免字符串值内部包含这类符号时被误判。数字解析目前只支持整数。如果需要小数可以在_resolve_value里补充float判断。这里展示的是学习用的最小实现。生产环境还应考虑负数、科学计数法、null、数组、括号嵌套等问题。最小实现的目的是把核心思路讲清楚而不是直接照搬到线上。3.4 执行主循环与节点跳转有了节点定义和条件求值之后执行引擎的主循环就变得很直接class WorkflowEngine: def __init__(self, definition, variables): self.definition definition self.context Context(variables) self.nodes {} self.start_node definition.get(start, start) self.max_steps definition.get(maxSteps, 100) for node in definition.get(nodes, []): self.nodes[node[id]] node if self.start_node not in self.nodes: raise WorkflowRuntimeError(f起始节点不存在: {self.start_node}) def run(self): current_id self.start_node steps 0 while current_id: if steps self.max_steps: raise WorkflowRuntimeError(超过最大执行步数疑似出现死循环) node self.nodes.get(current_id) if not node: raise WorkflowRuntimeError(f节点不存在: {current_id}) print(f[step {steps}] 进入节点: {current_id}, 类型: {node[type]}) node_type node[type] if node_type action: self._execute_action(node) current_id node.get(next) elif node_type condition: expr node[condition] result evaluate_condition(expr, self.context) print(f 条件: {expr} {result}) current_id node.get(trueNext if result else falseNext) elif node_type end: print( 流程结束) return node.get(result, success) else: raise WorkflowRuntimeError(f不支持的节点类型: {node_type}) steps 1 raise WorkflowRuntimeError(流程被中断没有到达结束节点) def _execute_action(self, node): if set in node: for path, value in node[set].items(): # 这里先简化处理set 的值直接作为字面量写入 self.context.set(path, value) print(f 设置变量: {path} {value}) if node.get(description): print(f 动作说明: {node[description]})执行循环的终止条件有两种到达end节点或者经过的节点数超过maxSteps。maxSteps是防止死循环的关键机制。action节点的set操作目前只支持写入字面量。实际项目中可能有三种动作写入固定值例如把状态改成pending。写入变量引用例如把order.userId赋给currentOperator。调用外部服务并保存返回值。这些都可以在_execute_action里通过增加动作类型来扩展本小节不展开。3.5 运行示例与预期输出在main.py中创建入口import json from engine import WorkflowEngine def main(): with open(workflow.json, r, encodingutf-8) as f: definition json.load(f) variables { order: { amount: 15000, status: submitted } } engine WorkflowEngine(definition, variables) result engine.run() print(最终结果:, result) if __name__ __main__: main()运行命令python main.py预期输出[step 0] 进入节点: start, 类型: action 动作说明: 接收订单 设置变量: order.status pending [step 1] 进入节点: check_amount, 类型: condition 条件: order.amount 10000 True [step 2] 进入节点: manager_approve, 类型: action 动作说明: 经理审批 设置变量: order.status approved_by_manager [step 3] 进入节点: end, 类型: end 流程结束 最终结果: order_processed如果把order.amount改为5000预期输出会跳过manager_approve进入auto_approve。这说明分支逻辑已经按配置生效。4. 关键实现细节与参数说明4.1 条件求值的安全边界实现条件求值最常见也最危险的错误写法是直接调用 Python 的evalresult eval(order.amount 10000, context) # 不推荐eval虽然能让配置变得极其灵活但也会带来任意代码执行风险。如果工作流配置来源不可信恶意配置可以直接调用系统函数或读取敏感文件。所以最小引擎里使用白名单解析器只允许变量路径、数字、字符串、布尔值和有限的比较、逻辑操作符。这是安全底线不能妥协。即使内部系统的配置来源可信也不建议用eval。原因有两点eval会让错误堆栈难以定位条件表达式里的语法错误和业务变量错误混在一起。eval会引入隐性依赖例如配置里使用了os、sys这类内置模块引擎将无法控制执行边界。4.2 节点表达式的数据流无类型写法看起来自由但执行引擎内部要建立清晰的“数据流”约束。一个节点能访问什么数据应该由引擎统一管理而不是配置里任意读取全局对象。最小引擎里所有数据都保存在Context中。action节点通过set写入condition节点通过表达式读取。表面上这是最简单的实现但它已经定义了一个重要边界表达式不能直接操作文件、网络或系统进程只能访问上下文里的数据。扩展方向上有几种选择在action节点中增加type: http或type: sql通过引擎内置的适配器执行外部请求并把返回值写回上下文。在condition中增加函数调用语法例如riskCheck(order.userId) approve由引擎注册可调用的函数白名单。增加输出参数让节点只暴露声明过的字段给后续节点形成显式数据流。这些能力都可以在最小引擎的基础上逐步补全但引入一项就要考虑一项的安全边界和错误处理。4.3 执行引擎的状态机约束工作流本质上是有限状态机。节点是状态next、trueNext、falseNext是状态转移。无类型写法不会自动带来正确性所以引擎需要显式保证几个约束每个节点必须有唯一的id。除end节点外每个节点必须有后续节点否则流程会中断。condition节点必须有trueNext和falseNext两个转移目标。节点之间不能存在循环依赖或者必须通过maxSteps限制步数。起始节点必须存在于nodes列表中。节点类型必须属于引擎支持的类型白名单。这些约束可以在加载定义时校验一次而不是等到运行时才发现。最小引擎目前只做了部分校验生产环境应该把校验逻辑集中到WorkflowLoader中与执行引擎分离。5. 从现象排查条件工作流的常见问题5.1 条件总是走到默认分支现象配置了order.amount 10000但无论传入金额是多少流程都走falseNext。排查路径先看变量是否真的传入了order.amount。可以在引擎启动时打印variables的完整内容。再看变量的类型。如果amount是字符串15000而表达式右侧是数字10000Python 的在字符串和数字之间会抛出TypeError但在某些实现里会被转成字符串比较导致逻辑错误。检查条件表达式是否有空格。最小引擎依赖空格切分 token如果写成order.amount10000解析器可能找不到操作符。常见原因和解决建议问题现象常见原因检查方式处理建议条件恒为假变量路径写错打印上下文并检查路径修正变量路径条件恒为真值比较的是字符串和数字打印变量类型在引擎里做显式类型转换条件解析失败表达式缺少空格使用workflow.json校验脚本统一配置格式异常被吞掉外层 catch 了所有异常查看日志是否出现WorkflowRuntimeError让异常显式抛出并记录5.2 变量取值为空或类型不匹配现象条件执行时报变量路径不存在: order.amount。原因通常是调用方传入的变量结构里没有order.amount或者变量名大小写不一致。例如传入的是order.Amount而配置写的是order.amount。处理做法在引擎入口打印变量完整结构。在Context.get的异常信息里带上实际可用路径。生产环境可以在加载工作流时对条件中出现的所有变量路径做一次静态扫描和样例变量结构对比提前发现拼写错误。5.3 工作流陷入死循环现象进程迟迟不结束日志里不断进入同一个节点。原因两个节点互指或者条件恒真导致trueNext一直回到自己。最小引擎里maxSteps已经能阻止无限循环但应该把它作为兜底而不是依赖它。更好的做法是加载工作流时做有向图环检测。如果maxSteps设为 100而流程正常情况下需要执行 120 个节点就会误报死循环。参数调大需要谨慎同时考虑内存和日志量。5.4 JSON 配置结构与字段命名不统一现象有的节点写next有的节点写then有的条件节点写trueNext有的写nextIfTrue。这些不一致会导致引擎报KeyError或走错分支。解决方式定义 JSON Schema 并在加载时校验。虽然无类型写法不声明值的类型但节点结构、必填字段、允许枚举仍然可以通过 Schema 约束。无类型不等于无结构。示例校验要点type必须是action、condition、end之一。condition节点必须包含condition、trueNext、falseNext。action节点必须包含next除非它是end节点。所有next、trueNext、falseNext引用的节点 id 必须存在于节点列表中。6. 从学习环境到生产环境的落地建议6.1 学习环境的快速验证路径如果想快速验证本文的引擎建议按以下顺序操作创建workflow.json直接使用第 2 节的订单审批示例。创建engine.py把第 3 节的代码复制进去。创建main.py先传入amount 15000确认输出走进经理审批分支。修改为amount 5000确认输出走进自动审批分支。故意写错条件表达式观察异常信息理解解析器的输出。这条路径能完整覆盖“定义 - 加载 - 求值 - 跳转 - 结束”的全过程。学习阶段不需要引入数据库和可视化界面。6.2 生产环境必须补充的能力从学习引擎到生产可用至少还需要补齐以下几块能力。第一配置存储和版本管理。工作流定义不应只放在本机 JSON 文件里建议放入配置中心或数据库并记录版本号。发布前做 diff回滚时切回上一版本。第二日志和链路追踪。每个节点执行都要记录工作流实例 ID、节点 ID、执行时间、条件结果、变量快照。排错时能完整还原一次执行过程。最小引擎只使用print输出生产环境必须换成结构化日志。第三监控和告警。对执行失败次数、死循环触发次数、平均执行时长、节点执行耗时做监控。超过阈值时告警。第四并发和幂等。同一份工作流定义会被多个实例并发执行。引擎不能持有共享的可变状态每一个执行实例都应该有独立的Context。第五权限和审核。工作流配置是业务规则的一部分修改后必须有审批流。不能允许任何人直接改生产配置。第六测试机制。可以录制历史执行样本喂给新版本工作流定义对比新旧执行结果降低配置变更风险。6.3 可复用的发布前检查清单在把一份条件工作流配置从测试环境发布到生产环境前可以按下面的清单逐项检查[ ] 所有节点 id 是否唯一。[ ] 起始节点是否存在。[ ]action节点是否都有next。[ ]condition节点是否都有trueNext和falseNext。[ ] 所有跳转目标是否存在于节点列表。[ ] 条件表达式是否能在样例变量上正确求值。[ ] 变量路径是否与调用方传入的数据结构一致。[ ] 数字比较是否会出现字符串和数字混比的情况。[ ] 是否配置了合理的maxSteps。[ ] 是否对工作流定义做了版本记录。[ ] 是否录制了至少一组正常场景和一组异常场景的测试样本。[ ] 是否确认配置修改不会影响正在执行中的工作流实例。这里最后一项尤其容易被忽略。生产中通常会有正在执行的流程实例它们使用的是旧版定义。如果直接覆盖配置会导致实例在中途突然使用新规则语义上很危险。常见做法是区分“实例启动时加载版本”和“新实例使用最新版本”让运行中的实例保持旧版本直到结束。6.4 扩展方向最小引擎验证了无类型条件工作流的可行性但距离一个通用工作流平台还有不少路要走。可以按以下方向逐步扩展条件表达式增强支持括号嵌套、函数调用、集合运算、null 判断。节点类型扩展支持延时节点、并行分支、子流程调用。可视化编辑根据 JSON 生成节点图支持拖拽配置。类型校验虽然配置是无类型写法但可以引入可选的schema字段对关键变量做运行时校验。多语言 SDK一套定义Java、Python、Node.js 各端执行结果一致。扩展时要坚持一个原则任何新增能力都要同时考虑安全边界和错误可观测性。无类型写法的自由度高执行引擎的约束和防护就要更明确。回到最开始的问题条件工作流到底应不应该用无类型写法对于规则频繁变化、调用方数据结构灵活、希望降低配置门槛的场景无类型写法非常有价值。它牺牲了编译期的类型安全换来了运营和配置层面的灵活性。但无类型不等于无约束节点结构、条件语法、变量路径、循环限制这些仍然需要引擎层严格校验。把“定义灵活”和“执行可控”分开处理才是无类型写法落地到生产环境时最该关注的设计思路。 
长河君 · 2026/8/30 5:46:01 · 阅读 1.2k 条件工作流无类型写法：JSON配置与Python执行引擎实战
在实际业务系统中条件工作流并不是一个陌生概念。无论是审批流转、订单处理还是任务编排系统都需要根据某个字段值或运算结果决定下一步执行哪个节点。常见实现有两种思路一种是在代码里用 if else 硬编码流程另一种是把流程定义做成可配置的数据结构。后一种方式里比较有争议的问题就是流程定义到底要不要引入类型系统。这里讨论的“条件工作流的无类型写法”就是使用 JSON、YAML 这类不携带类型声明的数据格式来定义流程节点、分支条件和跳转关系让流程引擎在运行时动态解释并执行。这种方式在轻量级工作流、规则引擎前端配置、运营后台配置化场景里非常常见。这篇文章会围绕一个最小可运行的订单审批条件工作流讲解无类型写法的定义方式、Python 执行引擎的实现思路、条件表达式的解析方法、常见排错链路以及从学习环境过渡到生产环境时需要补齐的能力。读完以后可以直接照着代码实现一个不依赖大型工作流框架、用 JSON 描述分支逻辑的简单执行引擎。1. 条件工作流为什么需要“无类型写法”1.1 从硬编码到流程定义早期系统的流程控制通常直接写在业务代码里。一个订单审批逻辑可能是这样的if (order.getAmount() 10000) { managerService.approve(order); } else { autoService.approve(order); }这种写法的优点是简单直接编译器能检查类型逻辑改动在代码评审里可以看到。但缺点也很明显业务流程一旦频繁调整比如金额阈值从 10000 改成 5000或者增加一个风险订单拦截节点就需要改代码、重新部署、再走一遍发布流程。对于运营、客服、风控团队来说这种节奏太慢。流程定义化则是把“顺序、分支、条件、动作”从代码中抽离出来放到可配置的数据文件或数据库表里。系统启动时加载这些定义运行时根据定义执行。这样流程变更就不需要改代码只需要更新配置并刷新缓存。1.2 无类型写法的含义与边界“无类型”这个词在不同语境下有不同含义。在编程语言层面无类型通常指不强制声明变量类型Python、JavaScript 都属于动态类型语言变量可以随时指向不同类型的数据。在工作流配置层面无类型写法更多指流程定义文件本身不引入 XML Schema、Protobuf、Thrift 这类强类型描述而是直接用 JSON、YAML 或 Map 结构描述节点和条件。例如一个条件节点在无类型写法里可能是{ id: check_amount, type: condition, condition: order.amount 10000, trueNext: manager_approve, falseNext: auto_approve }这里没有提前声明order.amount是整数还是字符串。它的实际类型由运行时传入的变量上下文决定。如果调用方传入的是10000那么条件求值时就会按数字比较如果传入的是10000那么引擎要么做类型转换要么按字符串比较具体行为取决于引擎实现。这种写法的核心价值是降低配置门槛让非开发人员也能看懂大部分分支条件。代价是类型安全被削弱很多错误要等到运行时才暴露。1.3 无类型与有类型选型对比实际项目中究竟该用无类型配置还是引入有类型 DSL需要结合团队情况和变更频率来判断。对比维度无类型 JSON/YAML 配置有类型 DSL 或代码定义表示形式JSON、YAML、Map无字段类型声明自定义 DSL、Protobuf、Java 类、TypeScript 类型修改成本改配置即可发布流程简单改代码或改 DSL 编译产物流程较重类型安全性弱运行时才暴露类型错误强编译期或加载期可校验学习成本低运营和产品可读性强高需要维护 DSL 语法和运行时性能需在运行时解析表达式略有损耗可预编译执行效率更高适用场景中小团队、轻量编排、运营配置大型平台、需要长期维护的复杂编排无类型写法并不是为了替代有类型方案而是为了解决“流程经常变、又不希望每次变更都发版本”这一类问题。等流程数量和复杂度增长到一定规模后再考虑引入 schema 校验和编译期检查也是一种平滑演进路径。2. 设计一份最小无类型条件工作流定义2.1 JSON 结构节点、连接和条件为了让执行引擎足够简单又具备代表性这里把工作流定义收敛成四种节点action执行一个动作通常是设置变量或调用外部服务然后跳转到下一个节点。condition对变量上下文求值一个条件表达式结果为真走trueNext为假走falseNext。end流程结束返回一个结果标记。start其实可以用普通action节点表示但显式命名会让结构更清晰。一个节点至少包含id和type两个字段。condition节点必须包含condition、trueNext、falseNext。action节点可以包含set用于在上下文中写入变量。{ name: 订单审批条件工作流, maxSteps: 30, nodes: [ { id: start, type: action, description: 接收订单, set: { order.status: pending }, next: check_amount }, { id: check_amount, type: condition, condition: order.amount 10000, trueNext: manager_approve, falseNext: auto_approve }, { id: manager_approve, type: action, description: 经理审批, set: { order.status: approved_by_manager }, next: end }, { id: auto_approve, type: action, description: 系统自动通过, set: { order.status: auto_approved }, next: end }, { id: end, type: end, result: order_processed } ] }这份定义里没有任何类型声明。order.amount是数字还是字符串要到引擎执行时才确定。set里的值也没有声明类型配置里写什么类型上下文里就是什么类型。这种设计在小型项目中非常灵活但也会带来一个隐患结构没有任何约束一旦某个节点漏写next字段引擎可能把流程断了。后面会专门讨论如何通过校验和检查清单防住这类问题。2.2 条件表达式语法设计条件表达式是整个工作流里最关键的部分。无类型写法不要求使用者声明类型但表达式语法必须足够简单才能避免直接使用eval带来的安全风险。最小引擎里支持的条件表达式语法如下比较运算、!、、、、逻辑运算and、or单值判断直接写变量路径或布尔值字面量数字、字符串、布尔值示例order.amount 10000 order.amount 5000 and order.status pending order.risk true or order.level high这里刻意不实现括号嵌套。原因是解析器会显著变复杂而对大多数审批、流转场景来说and和or的从左到右组合已经足够。如果确实需要复杂逻辑推荐的做法是拆成多个条件节点而不是把表达式写得越来越长。表达式里的空格不是必须的但强烈建议保留。解析器依赖空格来区分变量名、操作符和字面量。手写配置时统一使用半角空格可以避免很多解析问题。2.3 变量上下文与点路径执行引擎在运行过程中维护一个variables上下文对象。这个对象可以是任意嵌套的字典条件表达式通过点路径访问属性order.amount表示从根上下文取order再取amount。user.id表示从根上下文取user的id。config.enabled表示取config的enabled。在 Python 里点路径可以这样解析class Context: def __init__(self, variables): self.variables variables def get(self, path): parts path.split(.) current self.variables for part in parts: if not isinstance(current, dict) or part not in current: raise WorkflowRuntimeError(f变量路径不存在: {path}) current current[part] return current这种点路径设计的好处是配置可读性好运营人员可以直接看懂条件是在比较哪个字段。坏处是路径写错时错误信息需要足够清晰否则排错成本很高。3. 用 Python 实现一个最小执行引擎3.1 项目结构与依赖本文的示例只需要 Python 3.8 以上环境不依赖第三方库。项目结构如下condition-workflow/ ├── workflow.json ├── engine.py └── main.py三个文件的职责workflow.json工作流定义用无类型 JSON 描述节点和条件。engine.py执行引擎负责加载定义、解析节点、求值条件、维护节点跳转。main.py入口脚本构造变量上下文运行工作流并输出结果。3.2 加载与解析工作流定义在engine.py中先定义异常类型和上下文类import json class WorkflowRuntimeError(Exception): pass class Context: def __init__(self, variables): self.variables variables def get(self, path): parts path.split(.) current self.variables for part in parts: if not isinstance(current, dict) or part not in current: raise WorkflowRuntimeError(f变量路径不存在: {path}) current current[part] return current def set(self, path, value): parts path.split(.) current self.variables for part in parts[:-1]: if part not in current: current[part] {} if not isinstance(current[part], dict): raise WorkflowRuntimeError(f无法在非对象路径上设置变量: {path}) current current[part] current[parts[-1]] value上下文类只做两件事读变量和写变量。读变量时路径不存在会抛出异常这样配置错误可以在第一时间暴露。写变量时支持点路径例如order.status会写入到variables[order][status]。3.3 变量上下文与条件求值条件求值是引擎的核心也是最需要谨慎的部分。这里不使用eval而是实现一个小型解析器。先处理逻辑运算再处理比较运算def _split_top_level(expr, sep): parts [] start 0 depth 0 quote None i 0 while i len(expr): ch expr[i] if quote: if ch quote: quote None i 1 continue if ch in (, ): quote ch i 1 continue if ch (: depth 1 elif ch ): depth - 1 if depth 0 and expr.startswith(sep, i): parts.append(expr[start:i].strip()) start i len(sep) i len(sep) continue i 1 parts.append(expr[start:].strip()) return parts def _resolve_value(token, ctx): token token.strip() if token true: return True if token false: return False if len(token) 2 and token[0] in (, ) and token[-1] token[0]: return token[1:-1] if token.lstrip(-).isdigit(): return int(token) return ctx.get(token) def evaluate_condition(expr, ctx): expr expr.strip() if not expr: raise WorkflowRuntimeError(条件表达式不能为空) for op in ( or , and ): if op in expr: parts _split_top_level(expr, op.strip()) results [evaluate_condition(p, ctx) for p in parts] if op or : return any(results) return all(results) for operator in (, , , !, , ): # 从表达式里查找第一个比较运算符 idx _find_operator(expr, operator) if idx 0: left expr[:idx].strip() right expr[idx len(operator):].strip() left_val _resolve_value(left, ctx) right_val _resolve_value(right, ctx) return _compare(left_val, operator, right_val) # 没有比较运算符时按布尔值处理 return bool(_resolve_value(expr, ctx)) def _find_operator(expr, operator): quote None i 0 while i len(expr): ch expr[i] if quote: if ch quote: quote None i 1 continue if ch in (, ): quote ch i 1 continue if expr.startswith(operator, i): return i i 1 return -1 def _compare(left_val, operator, right_val): if operator : return left_val right_val if operator !: return left_val ! right_val if operator : return left_val right_val if operator : return left_val right_val if operator : return left_val right_val if operator : return left_val right_val raise WorkflowRuntimeError(f不支持的操作符: {operator})这段代码有几个关键点_split_top_level会跳过引号内的分隔符避免字符串里的and被误切。_resolve_value根据 token 形态决定它是字面量还是变量路径。比较运算符的查找也会跳过引号避免字符串值内部包含这类符号时被误判。数字解析目前只支持整数。如果需要小数可以在_resolve_value里补充float判断。这里展示的是学习用的最小实现。生产环境还应考虑负数、科学计数法、null、数组、括号嵌套等问题。最小实现的目的是把核心思路讲清楚而不是直接照搬到线上。3.4 执行主循环与节点跳转有了节点定义和条件求值之后执行引擎的主循环就变得很直接class WorkflowEngine: def __init__(self, definition, variables): self.definition definition self.context Context(variables) self.nodes {} self.start_node definition.get(start, start) self.max_steps definition.get(maxSteps, 100) for node in definition.get(nodes, []): self.nodes[node[id]] node if self.start_node not in self.nodes: raise WorkflowRuntimeError(f起始节点不存在: {self.start_node}) def run(self): current_id self.start_node steps 0 while current_id: if steps self.max_steps: raise WorkflowRuntimeError(超过最大执行步数疑似出现死循环) node self.nodes.get(current_id) if not node: raise WorkflowRuntimeError(f节点不存在: {current_id}) print(f[step {steps}] 进入节点: {current_id}, 类型: {node[type]}) node_type node[type] if node_type action: self._execute_action(node) current_id node.get(next) elif node_type condition: expr node[condition] result evaluate_condition(expr, self.context) print(f 条件: {expr} {result}) current_id node.get(trueNext if result else falseNext) elif node_type end: print( 流程结束) return node.get(result, success) else: raise WorkflowRuntimeError(f不支持的节点类型: {node_type}) steps 1 raise WorkflowRuntimeError(流程被中断没有到达结束节点) def _execute_action(self, node): if set in node: for path, value in node[set].items(): # 这里先简化处理set 的值直接作为字面量写入 self.context.set(path, value) print(f 设置变量: {path} {value}) if node.get(description): print(f 动作说明: {node[description]})执行循环的终止条件有两种到达end节点或者经过的节点数超过maxSteps。maxSteps是防止死循环的关键机制。action节点的set操作目前只支持写入字面量。实际项目中可能有三种动作写入固定值例如把状态改成pending。写入变量引用例如把order.userId赋给currentOperator。调用外部服务并保存返回值。这些都可以在_execute_action里通过增加动作类型来扩展本小节不展开。3.5 运行示例与预期输出在main.py中创建入口import json from engine import WorkflowEngine def main(): with open(workflow.json, r, encodingutf-8) as f: definition json.load(f) variables { order: { amount: 15000, status: submitted } } engine WorkflowEngine(definition, variables) result engine.run() print(最终结果:, result) if __name__ __main__: main()运行命令python main.py预期输出[step 0] 进入节点: start, 类型: action 动作说明: 接收订单 设置变量: order.status pending [step 1] 进入节点: check_amount, 类型: condition 条件: order.amount 10000 True [step 2] 进入节点: manager_approve, 类型: action 动作说明: 经理审批 设置变量: order.status approved_by_manager [step 3] 进入节点: end, 类型: end 流程结束 最终结果: order_processed如果把order.amount改为5000预期输出会跳过manager_approve进入auto_approve。这说明分支逻辑已经按配置生效。4. 关键实现细节与参数说明4.1 条件求值的安全边界实现条件求值最常见也最危险的错误写法是直接调用 Python 的evalresult eval(order.amount 10000, context) # 不推荐eval虽然能让配置变得极其灵活但也会带来任意代码执行风险。如果工作流配置来源不可信恶意配置可以直接调用系统函数或读取敏感文件。所以最小引擎里使用白名单解析器只允许变量路径、数字、字符串、布尔值和有限的比较、逻辑操作符。这是安全底线不能妥协。即使内部系统的配置来源可信也不建议用eval。原因有两点eval会让错误堆栈难以定位条件表达式里的语法错误和业务变量错误混在一起。eval会引入隐性依赖例如配置里使用了os、sys这类内置模块引擎将无法控制执行边界。4.2 节点表达式的数据流无类型写法看起来自由但执行引擎内部要建立清晰的“数据流”约束。一个节点能访问什么数据应该由引擎统一管理而不是配置里任意读取全局对象。最小引擎里所有数据都保存在Context中。action节点通过set写入condition节点通过表达式读取。表面上这是最简单的实现但它已经定义了一个重要边界表达式不能直接操作文件、网络或系统进程只能访问上下文里的数据。扩展方向上有几种选择在action节点中增加type: http或type: sql通过引擎内置的适配器执行外部请求并把返回值写回上下文。在condition中增加函数调用语法例如riskCheck(order.userId) approve由引擎注册可调用的函数白名单。增加输出参数让节点只暴露声明过的字段给后续节点形成显式数据流。这些能力都可以在最小引擎的基础上逐步补全但引入一项就要考虑一项的安全边界和错误处理。4.3 执行引擎的状态机约束工作流本质上是有限状态机。节点是状态next、trueNext、falseNext是状态转移。无类型写法不会自动带来正确性所以引擎需要显式保证几个约束每个节点必须有唯一的id。除end节点外每个节点必须有后续节点否则流程会中断。condition节点必须有trueNext和falseNext两个转移目标。节点之间不能存在循环依赖或者必须通过maxSteps限制步数。起始节点必须存在于nodes列表中。节点类型必须属于引擎支持的类型白名单。这些约束可以在加载定义时校验一次而不是等到运行时才发现。最小引擎目前只做了部分校验生产环境应该把校验逻辑集中到WorkflowLoader中与执行引擎分离。5. 从现象排查条件工作流的常见问题5.1 条件总是走到默认分支现象配置了order.amount 10000但无论传入金额是多少流程都走falseNext。排查路径先看变量是否真的传入了order.amount。可以在引擎启动时打印variables的完整内容。再看变量的类型。如果amount是字符串15000而表达式右侧是数字10000Python 的在字符串和数字之间会抛出TypeError但在某些实现里会被转成字符串比较导致逻辑错误。检查条件表达式是否有空格。最小引擎依赖空格切分 token如果写成order.amount10000解析器可能找不到操作符。常见原因和解决建议问题现象常见原因检查方式处理建议条件恒为假变量路径写错打印上下文并检查路径修正变量路径条件恒为真值比较的是字符串和数字打印变量类型在引擎里做显式类型转换条件解析失败表达式缺少空格使用workflow.json校验脚本统一配置格式异常被吞掉外层 catch 了所有异常查看日志是否出现WorkflowRuntimeError让异常显式抛出并记录5.2 变量取值为空或类型不匹配现象条件执行时报变量路径不存在: order.amount。原因通常是调用方传入的变量结构里没有order.amount或者变量名大小写不一致。例如传入的是order.Amount而配置写的是order.amount。处理做法在引擎入口打印变量完整结构。在Context.get的异常信息里带上实际可用路径。生产环境可以在加载工作流时对条件中出现的所有变量路径做一次静态扫描和样例变量结构对比提前发现拼写错误。5.3 工作流陷入死循环现象进程迟迟不结束日志里不断进入同一个节点。原因两个节点互指或者条件恒真导致trueNext一直回到自己。最小引擎里maxSteps已经能阻止无限循环但应该把它作为兜底而不是依赖它。更好的做法是加载工作流时做有向图环检测。如果maxSteps设为 100而流程正常情况下需要执行 120 个节点就会误报死循环。参数调大需要谨慎同时考虑内存和日志量。5.4 JSON 配置结构与字段命名不统一现象有的节点写next有的节点写then有的条件节点写trueNext有的写nextIfTrue。这些不一致会导致引擎报KeyError或走错分支。解决方式定义 JSON Schema 并在加载时校验。虽然无类型写法不声明值的类型但节点结构、必填字段、允许枚举仍然可以通过 Schema 约束。无类型不等于无结构。示例校验要点type必须是action、condition、end之一。condition节点必须包含condition、trueNext、falseNext。action节点必须包含next除非它是end节点。所有next、trueNext、falseNext引用的节点 id 必须存在于节点列表中。6. 从学习环境到生产环境的落地建议6.1 学习环境的快速验证路径如果想快速验证本文的引擎建议按以下顺序操作创建workflow.json直接使用第 2 节的订单审批示例。创建engine.py把第 3 节的代码复制进去。创建main.py先传入amount 15000确认输出走进经理审批分支。修改为amount 5000确认输出走进自动审批分支。故意写错条件表达式观察异常信息理解解析器的输出。这条路径能完整覆盖“定义 - 加载 - 求值 - 跳转 - 结束”的全过程。学习阶段不需要引入数据库和可视化界面。6.2 生产环境必须补充的能力从学习引擎到生产可用至少还需要补齐以下几块能力。第一配置存储和版本管理。工作流定义不应只放在本机 JSON 文件里建议放入配置中心或数据库并记录版本号。发布前做 diff回滚时切回上一版本。第二日志和链路追踪。每个节点执行都要记录工作流实例 ID、节点 ID、执行时间、条件结果、变量快照。排错时能完整还原一次执行过程。最小引擎只使用print输出生产环境必须换成结构化日志。第三监控和告警。对执行失败次数、死循环触发次数、平均执行时长、节点执行耗时做监控。超过阈值时告警。第四并发和幂等。同一份工作流定义会被多个实例并发执行。引擎不能持有共享的可变状态每一个执行实例都应该有独立的Context。第五权限和审核。工作流配置是业务规则的一部分修改后必须有审批流。不能允许任何人直接改生产配置。第六测试机制。可以录制历史执行样本喂给新版本工作流定义对比新旧执行结果降低配置变更风险。6.3 可复用的发布前检查清单在把一份条件工作流配置从测试环境发布到生产环境前可以按下面的清单逐项检查[ ] 所有节点 id 是否唯一。[ ] 起始节点是否存在。[ ]action节点是否都有next。[ ]condition节点是否都有trueNext和falseNext。[ ] 所有跳转目标是否存在于节点列表。[ ] 条件表达式是否能在样例变量上正确求值。[ ] 变量路径是否与调用方传入的数据结构一致。[ ] 数字比较是否会出现字符串和数字混比的情况。[ ] 是否配置了合理的maxSteps。[ ] 是否对工作流定义做了版本记录。[ ] 是否录制了至少一组正常场景和一组异常场景的测试样本。[ ] 是否确认配置修改不会影响正在执行中的工作流实例。这里最后一项尤其容易被忽略。生产中通常会有正在执行的流程实例它们使用的是旧版定义。如果直接覆盖配置会导致实例在中途突然使用新规则语义上很危险。常见做法是区分“实例启动时加载版本”和“新实例使用最新版本”让运行中的实例保持旧版本直到结束。6.4 扩展方向最小引擎验证了无类型条件工作流的可行性但距离一个通用工作流平台还有不少路要走。可以按以下方向逐步扩展条件表达式增强支持括号嵌套、函数调用、集合运算、null 判断。节点类型扩展支持延时节点、并行分支、子流程调用。可视化编辑根据 JSON 生成节点图支持拖拽配置。类型校验虽然配置是无类型写法但可以引入可选的schema字段对关键变量做运行时校验。多语言 SDK一套定义Java、Python、Node.js 各端执行结果一致。扩展时要坚持一个原则任何新增能力都要同时考虑安全边界和错误可观测性。无类型写法的自由度高执行引擎的约束和防护就要更明确。回到最开始的问题条件工作流到底应不应该用无类型写法对于规则频繁变化、调用方数据结构灵活、希望降低配置门槛的场景无类型写法非常有价值。它牺牲了编译期的类型安全换来了运营和配置层面的灵活性。但无类型不等于无约束节点结构、条件语法、变量路径、循环限制这些仍然需要引擎层严格校验。把“定义灵活”和“执行可控”分开处理才是无类型写法落地到生产环境时最该关注的设计思路。