当前位置：网站首页>Durable rules(持久规则引擎) 学习小记

Durable rules(持久规则引擎) 学习小记

2022-07-29 19:43:00 【翻滚的小@强】

1. 写在前面

这篇文章记录下学习durable rules的过程，持久规则引擎是一个多语言框架， 用于创建能够对大量事实陈述进行推理的基于规则的系统，白话的讲就是事先制定好一些规则，这些规则描述要匹配的事件以及采取的操作。这样，当有事实过来的时候，就可以去匹配事件然后采取相应的行为了。很类似我们代码中写的if...else if..else的逻辑。

那么为啥要用这东西呢？快，且适用于多语言，且如果判断太多，不像if else那么臃肿。写出来的判断语句更加的优雅。

当然，由于我也是刚学，且可参考的文档不多，我就拿github项目来通过3个demo整理下怎么用，并且最后自己想了个demo，整理下实际使用应该怎么用。在公司见同事用的这个操作，感觉还是很骚气的，于是就像学习下。

场景：

比如一个女生找对象的条件：长得帅并且有房子，并且收入很高，并且是职业是老师或者程序员的时候，才会和他约会，那么假设我们有很多个男生的前面这几个属性，最后判断这个女生对他的态度。

这时候，假设有个男生的数据：{"name": "zhangsan", "age": 25, "job": "teacher", "salary": 2000, "appearance": "good"}，我们判断女生最后的态度，喜欢，一般，不喜欢三种感觉吧。

demo是我自己编的，这个比较常规的做法，就是写if语句，if ... then ....的这种，但当条件判断很多的时候，这个写法就显得很繁琐臃肿，并且也不优雅，所以如果这个女生选对象的标准固定之后，我们就可以写一个规则，往后再有新数据判断，只需要调用这个规则，就可以。

2. 用法demo理解

2.1 最基本

这里整理上面链接中的demo理解。

定义规则，描述要匹配的事件或事实模式（先行）和要采取的操作（后续）

首先，安装包：

pip install durable_rules

我这里发现，如果系统里面装上pyenv，然后再装conda之后，发现直接用pip，并不会把包装到了conda的环境中，而是pyenv的环境中了，所以我这里必须进入具体的conda虚拟环境，然后调用那里面的pip才行。否则，在conda里面运行jupyter会报错找不到包，这是因为直接pip会把包装到pyenv中。这个得注意下。当然，没有pyenv的直接pip即可。

from durable.lang import *

# 规则名字
with ruleset('test'):
		# 这东西如果匹配上一个，就不往下走了
        # antecedent 先行条件 
        @when_all(m.name == 'wuzhongqiang')   # 这里面的字段就是传入的数据里面有的字段
        def say_hello(c):
            # consequent 后续
            print ('Hello {}'.format(c.m.name))
        
        @when_all(m.age > 18)
        def young(c):
            print("{} 成年了".format(c.m.name))
        
        @when_all(m.sex == "boy")
        def sex(c):
            print("{} 是个男孩".format(c.m.name))
        
        # 兜底的功能，也就是上面这些都不匹配的时候， 就走最下面这个
        @when_all(+m.sex)
        def output(c):
            print(c.m.name, c.m.age, c.m.sex)

这是最基本最常用的使用方法，看这个语法也比较简单，先定义一个规则名字，然后开始写先行条件，以及满足这个先行条件之后，触发的行为。

# 这个匹配第一个先行条件 
post('test', {
     'name': 'wuzhongqiang', 'age': 25, 'sex': 'boy'})
## Hello wuzhongqiang 也就是匹配到一个先行条件，执行之后，就不忘后面走了

# 匹配第二个先行条件
post('test', {
     'name': 'zhongqiang', 'age': 25, 'sex': 'boy'})
# zhongqiang 成年了

# 不匹配所有先行条件，就走最后一个兜底策略
post('test', {
     'name': 'zhongqiang', 'age': 16, 'sex': 'girl'})
# zhongqiang 16 girl

这个逻辑很简单，不用多说，但是有两点注意：

规则名字只能用一次，也就是上面规则定义这段代码，如果重复定义，会报错，显示规则名注册过，报异常
要有兜底的行为，否则，假设传入的数据都不匹配先行条件的话，会报错无法处理异常

2.2 assert_fact

这个写法就有点复杂了，讲真，没想出来啥时候会用到，但我大约看明白他的例子了。

这个东西用于正向推理，类似三段论：

如果P，那么Q
P出现
因此， Q

他给的那个例子，其实一开始没看明白，后面经过调试，差不多搞懂了。

from durable.lang import *
with ruleset('animal'):
        @when_all(c.first << (m.predicate == 'eats') & (m.object == 'flies'),
                  (m.predicate == 'lives') & (m.object == 'water') & (m.subject == c.first.subject))
        def frog(c):
            print(c.first.subject)
            c.assert_fact({
     'subject': c.first.subject, 'predicate': 'is', 'object': 'frog' })

        @when_all(c.first << (m.predicate == 'eats') & (m.object == 'flies'),
                  (m.predicate == 'lives') & (m.object == 'land') & (m.subject == c.first.subject))
        def chameleon(c):
            c.assert_fact({
     'subject': c.first.subject, 'predicate': 'is', 'object': 'chameleon' })

        @when_all(+m.subject)
        def output(c):
            print('Fact: {0} {1} {2}'.format(c.m.subject, c.m.predicate, c.m.object))

这里其实定义了先决条件，比如我运行下面这行代码：

assert_fact('animal', {
     'subject': 'Kermit', 'predicate': 'eats', 'object': 'flies' })

这个会输出Fact: Kermit eats flies，这是因为传入的这个东西，不满足上面任何一个先行条件（因为只有他自己），但是运行了这个之后，接下来，如果再运行下面这行代码：

assert_fact('animal', {
     'subject': 'Kermit', 'predicate': 'lives', 'object': 'water' })

这个的输出结果：

Kermit
Fact: Kermit is frog
Fact: Kermit lives water

这个是因为，有了一个先决条件(c.first)，也就是运行的第一行代码，然后再运行下面这个的话，正好匹配上第一条先决条件

        @when_all(c.first << (m.predicate == 'eats') & (m.object == 'flies'),
                  (m.predicate == 'lives') & (m.object == 'water') & (m.subject == c.first.subject))
        def frog(c):
            print(c.first.subject)
            c.assert_fact({
     'subject': c.first.subject, 'predicate': 'is', 'object': 'frog' })

然后都在output那里输出。就出来了上面的结果。

并且还发现一个现象，如果先运行

assert_fact('animal', {
     'subject': 'Kermit', 'predicate': 'lives', 'object': 'water' })

只会输出Fact: Kermit lives water，但在这个基础上，再运行

assert_fact('animal', {
     'subject': 'Kermit', 'predicate': 'eats', 'object': 'flies' })

这个同样也会输出：

Kermit
Fact: Kermit is frog
Fact: Kermit eats flies

竟然也会是上面的这个逻辑，就很纳闷，不知道为啥。

但这个使用场景，我理解会有一个先行条件先发生，发生了之后，在做匹配，如果匹配上先决条件，就会执行相应的语句。

2.3 模式匹配

这个就是上面最基本的用法

from durable.lang import *

with ruleset('pipei'):
    @when_all(m.subject.matches('3[47][0-9]{13}'))
    def amex(c):
        print ('Amex detected {0}'.format(c.m.subject))

    @when_all(m.subject.matches('4[0-9]{12}([0-9]{3})?'))
    def visa(c):
        print ('Visa detected {0}'.format(c.m.subject))

    @when_all(m.subject.matches('(5[1-5][0-9]{2}|222[1-9]|22[3-9][0-9]|2[3-6][0-9]{2}|2720)[0-9]{12}'))
    def mastercard(c):
        print ('Mastercard detected {0}'.format(c.m.subject))

所以用的时候，如果条件互斥的情况，我们就可以写到同一个规则下面。

3. demo

通过了一上午摸索，发现这个东西并不是很难，和if…else很像，下面通过开头的场景demo来看看如何用。

实际使用的时候，我们一般会把规则引擎携写成一个类

class RuleEngine:
    def __init__(self, rule_name: str = ''):
        if not rule_name:
            raise ValueError("rule_name is empty")
        self.rule_name = rule_name

    def get_result(self, data: dict = None):
        if not data:
            raise ValueError("data is empty")

        output = None
        for _ in range(3):
            output = post(self.rule_name, data)
            if not output is None:
                break

        if not output is None:
            return output.get('result', {
    })
        else:
            print("get_result, return None: {}".format(data))
            return {
    }

把规则名字传进去，建立对象

attitude_rule = RuleEngine('attitude')

建立一个新数据

person1 = {
    "name": "zhangsan", "age": 25, "job": "teacher", "salary": 2000, "appearance": "good"}

编写规则，这步是核心

with ruleset("attitude"):
    @when_all(
        (m.age < 30)
        & ((m.salary > 10000) | (m.appearance == "good")) & ((m.job == "teacher") | (m.job == "manong"))
    )
    def like(c):
        c.s.result = {
    
            'attitude_res': 'like',
            }
        
    @when_all(
        (m.age < 30)
        & ((m.salary <= 10000) | (m.appearance == "medium")) & ((m.job == "teacher") | (m.job == "manong"))
    )
    def yiban(c):
        c.s.result = {
    
            'attitude_res': 'yiban',
            }
        
    @when_all(+m._age)
    def other(c):
        c.s.result = {
    
            'attitude_res': 'dislike'
        }

然后调用get_result函数就能拿到结果

attitude_res = attitude_rule.get_result(person1)
# {'attitude_res': 'like'}

把这个更新到person

person1.update(attitude_res)

person1: 
{
    'name': 'zhangsan',
 'age': 25,
 'job': 'teacher',
 'salary': 2000,
 'appearance': 'good',
 'attitude_res': 'like'}