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: 大额订单风控

每个字段都有它的用处，我列个表说明：

这个模型看起来简单，但足够覆盖大多数业务场景。我刻意没有加"规则有效期""规则分组""规则版本"这类字段，不是不需要，而是觉得可以放到外层配置里去解决，让核心模型保持清晰。

2.2 条件表达式的语义设计：op、value与$字段引用

when部分是整个规则模型里最容易踩坑的地方。最初我设计的条件只有固定几种比较操作，后来实际使用中慢慢扩展到了十多种。目前支持的操作符如下：

设计上有几个容易被忽略的点。第一，value默认是字面量，但如果value以$开头，就代表引用另一个字段的值。比如我想判断"订单金额大于客户信用额度"，可以写value: "$customer.credit_limit"，这种设计让规则能表达字段间比较，而不仅仅是字段跟常量的比较。

第二，字段路径用点分语法。order.amount会被解析成data["order"]["amount"]，规避了深层嵌套时反复写循环取值的问题。不过这也要求入参数据结构稳定，如果你的数据源字段经常变动，建议在规则加载时做一次字段路径校验，防止一条规则在运行时才发现字段不存在。

第三，all和any的嵌套。我最终实现的是支持任意层级的组合：when下面可以是all，all的子项又可以是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, encoding="utf-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( name=item["name"], when=item["when"], then=item["then"], priority=item.get("priority", 0), stop_on_match=item.get("stop_on_match", False), comment=item.get("comment", ""), )) rules.sort(key=lambda r: r.priority, reverse=True) 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 改完的收益：可读性、可测试性、上线速度

我把改造后的收益总结成一张表，这种对比最直观：

更让我意外的是可测试性的提升。改造前想测试某条分支，得准备一个满足所有前置条件的完整订单对象；改造后我可以直接把某条规则抽出来，单独喂一组字段数据，断言命中结果。这个转变让测试用例数量增加了一倍还不止，但每个用例都更小、更清晰。

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_DEBUG=1开启。开启后，引擎会为每条被评估的规则输出一个结构化日志，包含字段路径、实际取值、期望值、比较结果。这样发现问题后，我直接看日志就能定位：原来是customer.level取到了None，而不是规则的vip。这套审计日志后来成了解决线上问题最快的入口，强烈建议你在一开始就设计进去，别等出事了再补。

5.5 热加载的坑：原子替换与缓存失效

为了实现"改规则、马上生效"，我做了文件监听，结果有一次线上事故让我记忆深刻：同事保存规则文件时，编辑器先清空内容再写入新内容，监听器恰好捕获了中间状态的空文件，引擎加载出空规则列表，一瞬间所有规则全部失效。虽然只持续了几百毫秒，但恰好有订单漏过了风控。

正确做法是两件事。一是采用临时文件+原子替换的发布方式，先写入.tmp文件再os.replace，让监听器永远只看到完整文件；二是监听器捕获到变更事件后，先校验YAML合法性再换入内存，校验失败就保留旧版本并报警。热加载这件事，宁可新规则晚一秒生效，也不能让旧规则突然消失。

6. 什么时候不要用"恭介的法则"

最后这个部分可能比前面所有内容都重要。规则引擎不是银弹，它适合的场景和不该碰的场景，我得说清楚。

6.1 规则少于十条、半年不变：别折腾

如果你的系统里只有个位数的规则，而且业务上已经稳定了几个月，那规则引擎就是纯粹增加复杂度。你多了一层文件加载、一套解析逻辑、一个动作注册表，换来的是压根用不上的灵活度。这种情况下，老老实实用函数写判断，比你搞任何引擎都强。我做这个引擎，是因为当时的规则数量已经失控，而不是因为"规则引擎听起来高级"。

6.2 没有规则测试意识：YAML也会出事

很多人觉得规则用配置写，就不用测试了。这是最大的误解。配置是代码的一种形式，而且因为它是声明式的，更容易出现"漏写条件""优先级错误"这类隐性bug。我的习惯是，每个规则文件都必须配套一组测试用例，用构造好的数据验证命中与不命中两种情况。这个习惯能帮你过滤掉大部分"规则写出来的瞬间就是错的"场景。

6.3 性能敏感路径：字符串匹配的开销

规则引擎的匹配本质是动态解析字段路径、动态调用比较函数，这个开销远大于直接写if判断。如果你的系统在单毫秒内要处理数十万次判断，比如网关层面的风控，规则引擎就不是好选择。对于这类场景，把规则预编译成可执行的字节码或原生函数才是正道。我做这个引擎的模块，流量一天也就几十万次判断，完全不用担心性能，但你要是拿它去撑高并发核心路径，一定会被压垮。

最后再分享一个实际操作中的小技巧：我给这套引擎写规则的时候，每条规则都会顺手加一行comment，写清楚"业务方要什么、为什么加这条"。半年后再回来看，这些注释的价值甚至超过了引擎本身的代码。你如果也要做类似的东西，建议从一开始就养成这个习惯。