1. 恭介的法则是从哪来的一个被if-else逼出来的项目我最初接触的大部分业务系统其实都不是败在架构上而是死在无止境的if-else里。去年我花了两周时间把一套订单处理逻辑里将近三百行的条件判断全部抽了出来换成了一种声明式的规则文件这个方案的内部代号就叫恭介的法则(rule)。先别被这个名字劝退它不是什么论文级别的复杂框架就是一个我自己维护的轻量规则引擎规则用YAML写Python负责读和执行总共两百多行核心代码。有人问为什么起这么个名字。说白了就是团队起代号时图省事当时的临时目录名叫rule_engine_v3后来又觉得太无趣顺手换成了某位同事随口念叨过的一个名字。法则这两个字倒是歪打正着——它确实把散落在代码各个角落的判断逻辑变成了可以一条条列出来、一条条遵守的规则清单。下面要聊的内容全是落地角度的问题不搞学术讨论。如果你是后端开发、数据工程或者任何跟业务规则打过交道的人可以参考我是怎么设计规则模型、怎么实现匹配执行以及路上踩过的那些坑。1.1 业务规则疯狂增长的三个月我接手的那套系统最初逻辑非常简单订单提交之后做一次基础校验金额超过一千直接转人工审核其余自动通过。这个逻辑用一个if就能写完没有任何问题。可业务方不会一直让你闲着。第一个月需求变成新客户首次下单且金额超过五千需要部门主管审批第二个月追加了华东地区订单超过三千要额外做风控校验第三个月又来了黑名单客户任何金额都直接拒绝但要发站内信通知。每一个需求单看都合理合起来就是灾难。三个月后核心流程函数里塞了十几层条件判断。最要命的是这些规则之间还有隐含顺序——比如风控校验必须发生在金额判断之后黑名单判断必须发生在所有优惠计算之前。新同事改一个分支经常会把另一个分支的逻辑带崩。我试着整理注释发现写注释的速度根本赶不上规则叠加的速度。那段时间我意识到一件事业务规则最大的特点不是复杂而是变化频繁且相互影响。代码适合描述流程但不适合维护一堆随时会增删改的规则条目。规则和流程混在一起等于把两个变数绑在了一块任何一个变动都得重新发版。1.2 为什么不做规则配置中心而是自己写引擎当时的第一反应是找现成方案。市面上确实有不少商业规则引擎有的功能强大到自带可视化编排界面还有开源产品可以做复杂的决策表。但评估一圈之后发现对我要解决的场景来说它们都太重了。首先多数规则引擎以Java生态为主而这个模块是Python写的为了规则引入一套JVM服务部署和运维成本直接翻倍。其次商业产品的学习曲线很陡团队里的新同事不可能花两周去学一套规则语言而规则如果只有我一个人能写那这套系统早晚会变成新的瓶颈。最关键的还是适度问题——我的需求只是一堆字段比较动作分发用不到推理机、正向链、决策树这些高级能力。所以我决定自己写一个轻量引擎规则用YAML描述引擎负责三件事加载并校验规则、根据入参匹配命中的规则、调用对应的动作处理函数。控制在两百行左右的代码量不引入外部依赖团队成员看一遍就能上手规则文件直接放在代码仓库里走评审和版本管理。事实证明这个选择是对的低成本、低门槛而且完全可控。2. 规则长什么样YAML驱动的规则模型与执行流程在写引擎之前先要把规则这个东西的数据结构定下来。规则的本质很简单在什么条件下做什么事情。对应到数据模型就是when和then两个部分再加上一些元信息。2.1 一条规则的完整字段拆解我用YAML做规则载体因为可读性比JSON好能写注释还不用在每行后面纠结逗号。下面这条规则是我项目里真实在用的作用是非VIP客户订单金额超过五万触发总监级人工审核- name: non_vip_high_amount_review comment: 非VIP客户订单超过50000元需要总监审批 priority: 100 stop_on_match: false when: all: - field: order.amount op: gt value: 50000 - field: customer.level op: ne value: vip then: action: create_review_task params: level: director reason: 大额订单风控每个字段都有它的用处我列个表说明字段作用说明name规则唯一标识日志、审计、测试都要用最好全局唯一comment规则说明写给业务方看的不参与执行priority优先级数字越大越先执行解决规则间先后问题stop_on_match短路开关命中后是否继续评估后续规则when条件支持all/any/not组合的表达式树then动作命中后执行的动作名称和参数这个模型看起来简单但足够覆盖大多数业务场景。我刻意没有加规则有效期规则分组规则版本这类字段不是不需要而是觉得可以放到外层配置里去解决让核心模型保持清晰。2.2 条件表达式的语义设计op、value与$字段引用when部分是整个规则模型里最容易踩坑的地方。最初我设计的条件只有固定几种比较操作后来实际使用中慢慢扩展到了十多种。目前支持的操作符如下操作符含义示例eq / ne等于 / 不等于customer.level eq vipgt / gte / lt / lte大小比较order.amount gt 50000in / not_in是否在集合内order.status in [pending, paid]contains字符串包含customer.name contains 测试starts_with前缀匹配order.order_no starts_with DHregex正则匹配customer.email regex ^..\..$exists字段是否存在product.spec exists设计上有几个容易被忽略的点。第一value默认是字面量但如果value以$开头就代表引用另一个字段的值。比如我想判断订单金额大于客户信用额度可以写value: $customer.credit_limit这种设计让规则能表达字段间比较而不仅仅是字段跟常量的比较。第二字段路径用点分语法。order.amount会被解析成data[order][amount]规避了深层嵌套时反复写循环取值的问题。不过这也要求入参数据结构稳定如果你的数据源字段经常变动建议在规则加载时做一次字段路径校验防止一条规则在运行时才发现字段不存在。第三all和any的嵌套。我最终实现的是支持任意层级的组合when下面可以是allall的子项又可以是any。实现上就是递归求值逻辑并不复杂但一定要把语义写清楚否则业务方很容易把AND和OR混在一起写出一堆永远命中的规则。2.3 引擎执行流程匹配、排序、命中与短路引擎的执行流程我分成四步顺序非常关键。第一步加载规则文件按priority降序排序。排序要在匹配之前做而不是在匹配之后做。为什么因为短路的语义是——命中一条最高优先级的规则就停止后续如果先匹配再排序短路时你根本不知道当前这条是不是最高优先级。第二步遍历排序后的规则逐一评估条件。评估结果是布尔值命中则进入动作收集列表。第三步根据stop_on_match决定是否提前终止。默认false表示所有命中的规则都要执行某些强互斥场景比如命中黑名单直接拒绝的规则就应该设成true避免后面还跑一堆优惠计算。第四步按顺序调用动作分发器把规则里的then变成真实业务调用。动作执行失败时引擎要决定是整体回滚还是跳过继续。我采用的做法是可重试的规则动作加入重试队列不可重试的记录下来并触发告警。这套流程听着简单实际写的时候还是会碰到边界情况。比如规则命中但动作执行抛异常此时日志里要能同时看到命中了哪条规则和动作挂了哪个参数否则排查问题会非常痛苦。3. 核心代码实现两百行把引擎跑起来我不喜欢那种原理讲一半、代码不给全的文章。下面把核心实现拆开讲都是可以直接抄走改一改就能用的代码。3.1 规则加载与优先级排序加载这一步最重要的不是读取文件而是校验。YAML解析出来如果结构不对越早报错越好别等到运行时才暴露。import yaml from dataclasses import dataclass dataclass class Rule: name: str when: dict then: dict priority: int 0 stop_on_match: bool False comment: str def load_rules(path: str) - list[Rule]: with open(path, encodingutf-8) as f: raw_items yaml.safe_load(f) rules [] for item in raw_items: if not item.get(name): raise ValueError(f规则缺少name字段: {item}) if when not in item or then not in item: raise ValueError(f规则{item.get(name, unknown)}缺少when或then) rules.append(Rule( nameitem[name], whenitem[when], thenitem[then], priorityitem.get(priority, 0), stop_on_matchitem.get(stop_on_match, False), commentitem.get(comment, ), )) rules.sort(keylambda r: r.priority, reverseTrue) return rules这段代码里有个细节排序放在加载之后、使用之前且启动时只排序一次。如果规则量不大每次匹配时排序也行但没必要。另外我会在加载后打印一份规则清单标明每条规则的优先级这样规则文件里写得再乱日志里也能看出引擎实际拿到的顺序。3.2 条件求值器的实现要点条件求值器是整个引擎的核心。我把它拆成三个函数取值、比较、递归组合。import re def get_field(data: dict, path: str): cur data for key in path.split(.): if isinstance(cur, dict) and key in cur: cur cur[key] else: return None return cur def evaluate_condition(cond: dict, data: dict) - bool: field_value get_field(data, cond[field]) op cond.get(op) raw_value cond.get(value) if isinstance(raw_value, str) and raw_value.startswith($): compare_value get_field(data, raw_value[1:]) else: compare_value raw_value return compare(field_value, op, compare_value) def compare(actual, op: str, expect) - bool: if op eq: return actual expect if op ne: return actual ! expect if op gt: return actual is not None and expect is not None and float(actual) float(expect) if op in: return actual in expect if op regex: return actual is not None and re.search(str(expect), str(actual)) is not None raise ValueError(f未知操作符: {op})有几个点我必须提醒。一是浮点比较的问题业务数据里金额经常是Decimal或者字符串直接float转换可能丢失精度所以我在实际项目里对gt这类操作符专门做了Decimal优先的处理。二是空值的语义字段不存在时get_field返回None此时eq、ne、in这些操作符都有不同的预期行为我统一约定None只能和None或exists操作符比较其他比较一律返回False并记一条警告日志避免静默出错。三是大小比较和字符串比较得分开否则不同版本的Python对字符串和数字比较的处理会让你得到莫名其妙的结果。3.3 动作分发then字段怎么变成真实业务调用规则命中之后引擎要做的事情是调用真实业务函数。最土但最实用的方式是维护一个全局动作注册表用装饰器注册用字符串索引ACTIONS {} def register_action(name): def decorator(func): ACTIONS[name] func return func return decorator register_action(create_review_task) def create_review_task(params, context): # 实际创建一条人工审核任务 review_service.create(params[level], params[reason]) return True def dispatch(rule: Rule, context: dict) - bool: action_name rule.then[action] handler ACTIONS.get(action_name) if handler is None: raise KeyError( f规则{rule.name}引用的动作{action_name}未注册 请检查是否遗漏import ) return handler(rule.then.get(params, {}), context)这个设计看起来简单但它解决了一个真实的大麻烦动作和规则解耦。业务方可以新增规则文件只要动作名已注册引擎一行不用改反过来新写一个动作函数也不用碰规则文件。另外动作注册表要在引擎启动时做一次全量校验——遍历所有已加载规则确认每个then.action都能在ACTIONS里找到找不到就直接启动失败。宁可启动时报错也不要运行到一半才炸。4. 实际改造案例把订单审批从300行if-else换成规则文件理论讲完拿真实案例串一遍。我当时做这个引擎的初衷就是改造订单审批流程所以这个案例最能说明问题。4.1 改造前的代码长什么样改造前的判断逻辑集中在decide_order_flow函数里大约三百行。核心片段长这样def decide_order_flow(order): customer get_customer(order.customer_id) if customer.level vip: if order.amount 100000: return ApprovalFlow.director(VIP超大额) return ApprovalFlow.auto_pass() if order.amount 50000: if order.region east: return ApprovalFlow.risk_check(华东大额) return ApprovalFlow.director(大额) if order.amount 3000 and order.region east: return ApprovalFlow.manager(华东中额) if customer.is_blacklisted: return ApprovalFlow.reject(黑名单) return ApprovalFlow.auto_pass()这段代码最大的问题不是长而是逻辑顺序隐藏得太深。比如黑名单判断放在最后可实际业务需求是黑名单应该最先拦截只是之前实现得早了没人敢动。再比如金额大于50000走总监审批和金额大于50000且在华东走风控这两个规则同时命中时到底执行哪个代码里根本没有明确表达完全靠分支先后顺序猜。4.2 迁移后的规则文件迁移之后同样的逻辑变成了一份可以阅读的YAML规则文件- name: blacklist_reject comment: 黑名单客户直接拒绝 priority: 1000 stop_on_match: true when: all: - field: customer.is_blacklisted op: eq value: true then: action: reject_order params: reason: 黑名单客户 - name: vip_super_high_review comment: VIP大额订单总监审批 priority: 200 when: all: - field: customer.level op: eq value: vip - field: order.amount op: gt value: 100000 then: action: create_review_task params: level: director reason: VIP超大额 - name: east_risk_check comment: 华东区大额订单额外风控 priority: 150 when: all: - field: order.region op: eq value: east - field: order.amount op: gt value: 5000 then: action: risk_check params: channel: east优先级一目了然黑名单1000最先短路VIP超大额200其次华东风控150再次。业务方看这个文件能直接说出如果客户是黑名单任何订单都会被拒绝这个结论而在旧代码里同样的结论需要逐行读函数。4.3 改完的收益可读性、可测试性、上线速度我把改造后的收益总结成一张表这种对比最直观维度改造前改造后新增一条规则改代码、走发版改YAML、走评审业务方参与评审基本看不懂代码可以直接读规则文件规则间优先级靠分支顺序隐含显式priority字段每条规则的测试很难单独测可以构造数据单独验证上线耗时小时到天分钟级配合热加载更让我意外的是可测试性的提升。改造前想测试某条分支得准备一个满足所有前置条件的完整订单对象改造后我可以直接把某条规则抽出来单独喂一组字段数据断言命中结果。这个转变让测试用例数量增加了一倍还不止但每个用例都更小、更清晰。5. 踩坑实录规则引擎落地时的五个深坑规则引擎看着简单真正用起来坑一个接一个。我踩过的这些坑大概率你也会遇到。5.1 优先级和文件顺序你以为的、实际执行的第一次上线我就栽在顺序上。写规则文件时我习惯把黑名单拒绝写在最后面觉得最后兜底逻辑上顺。结果引擎按priority排序黑名单算出来是1000反而最先执行了。当时业务方反馈黑名单客户的优惠券居然也被拦截了我排查半天才反应过来。这个坑的本质是文件顺序、书写顺序、执行顺序是三个完全不同的概念。我的建议是规则文件里按业务逻辑顺序写同时每个规则都写清楚priority并且引擎在启动日志里打印一份按优先级排序的清单。这样即使有人乱序写文件最终执行顺序也永远可控。5.2 类型陷阱YAML解析后050000变成了什么YAML对类型的推断非常智能但也非常坑。value: 50000会被解析成整数value: 50000是字符串而value: 050000这种带前导零的写法在部分解析器里会被当成八进制算出来是20480跟你想要的50000完全不是一回事。而业务系统传过来的数据又往往是字符串比如订单金额字段可能就是个50000.00。我遇到的问题是规则写op: gt, value: 50000实际数据是字符串50000结果比较直接返回False。引擎不报错规则不命中但业务方看到的结果就是该拦截的没拦截。后来我在compare函数里做了一层安全转换并且加载规则时对每条规则做一次静态检查如果value和字段类型明显不匹配就打警告日志。这个小改动后来又救了我好几次。5.3 规则命中但动作没触发action handler注册遗漏还有一次规则日志显示已经命中但业务动作没有任何反应。查了半天发现是动作函数所在模块忘记import了装饰器根本没执行ACTIONS注册表里压根没有这个动作。引擎匹配成功dispatch时抛KeyError但是异常被上层捕获后只记了一行WARN没有告警所以肉眼看不到。修复方案分两层。第一层是启动时做全量校验规则文件里每个引用到的action都必须存在于ACTIONS注册表否则直接启动失败。第二层是运行时兜底dispatch找不到action时不只是记日志还要往监控系统发一条告警。从此这类问题基本绝迹。5.4 调试困难给引擎加一套规则审计日志规则引擎最大的隐性成本是调试。传统代码里你可以下断点看变量规则文件就像一个黑盒——条件不满足但你不知道是哪个条件不满足。我后来加了一个审计模式通过环境变量RULE_DEBUG1开启。开启后引擎会为每条被评估的规则输出一个结构化日志包含字段路径、实际取值、期望值、比较结果。这样发现问题后我直接看日志就能定位原来是customer.level取到了None而不是规则的vip。这套审计日志后来成了解决线上问题最快的入口强烈建议你在一开始就设计进去别等出事了再补。5.5 热加载的坑原子替换与缓存失效为了实现改规则、马上生效我做了文件监听结果有一次线上事故让我记忆深刻同事保存规则文件时编辑器先清空内容再写入新内容监听器恰好捕获了中间状态的空文件引擎加载出空规则列表一瞬间所有规则全部失效。虽然只持续了几百毫秒但恰好有订单漏过了风控。正确做法是两件事。一是采用临时文件原子替换的发布方式先写入.tmp文件再os.replace让监听器永远只看到完整文件二是监听器捕获到变更事件后先校验YAML合法性再换入内存校验失败就保留旧版本并报警。热加载这件事宁可新规则晚一秒生效也不能让旧规则突然消失。6. 什么时候不要用恭介的法则最后这个部分可能比前面所有内容都重要。规则引擎不是银弹它适合的场景和不该碰的场景我得说清楚。6.1 规则少于十条、半年不变别折腾如果你的系统里只有个位数的规则而且业务上已经稳定了几个月那规则引擎就是纯粹增加复杂度。你多了一层文件加载、一套解析逻辑、一个动作注册表换来的是压根用不上的灵活度。这种情况下老老实实用函数写判断比你搞任何引擎都强。我做这个引擎是因为当时的规则数量已经失控而不是因为规则引擎听起来高级。6.2 没有规则测试意识YAML也会出事很多人觉得规则用配置写就不用测试了。这是最大的误解。配置是代码的一种形式而且因为它是声明式的更容易出现漏写条件优先级错误这类隐性bug。我的习惯是每个规则文件都必须配套一组测试用例用构造好的数据验证命中与不命中两种情况。这个习惯能帮你过滤掉大部分规则写出来的瞬间就是错的场景。6.3 性能敏感路径字符串匹配的开销规则引擎的匹配本质是动态解析字段路径、动态调用比较函数这个开销远大于直接写if判断。如果你的系统在单毫秒内要处理数十万次判断比如网关层面的风控规则引擎就不是好选择。对于这类场景把规则预编译成可执行的字节码或原生函数才是正道。我做这个引擎的模块流量一天也就几十万次判断完全不用担心性能但你要是拿它去撑高并发核心路径一定会被压垮。最后再分享一个实际操作中的小技巧我给这套引擎写规则的时候每条规则都会顺手加一行comment写清楚业务方要什么、为什么加这条。半年后再回来看这些注释的价值甚至超过了引擎本身的代码。你如果也要做类似的东西建议从一开始就养成这个习惯。
