首页 > 编程语言> > 在python里写Monad

在python里写Monad

2019-03-14 14:49:18 作者：互联网

这段时间写微信小程序看见callback hell非常炸毛，而且确实我不熟悉js的各种衍生工具(async/await, promise)，想着python的coroutine是yield模改出来的，准备自己也造一个轮子。由于生成器的原理就是把一段代码挂起然后可以让调用者控制是否继续下去，这个行为和Monad非常像。Monad本身的原理极其简单(bind, ret)，关键在于Haskell、gluon或者Idris之类的语言里面可以有语法糖，防止一个embedding hell，于是我也想到要用yield来模拟这个过程。大致思路是a = yield m_a让调用者知道这个是一个monad，然后通过generator的send来把bind的参数赋值给a，把a看作是后面的函数的参数即可，以避免不断写嵌套的匿名函数。因此我把js里的callback hell也变成了一个CPS来写(Cont Monad)，当然毕竟我比较懒不想学js/ts/es567等等新特性，所以只好自己造轮子了，大家要是在大的团队里就别这样了，省的被打死。

我们知道Haskell里的do-notation：

main = do
  x <- someIOMonad
  y <- anotherIOMonad
  otherIOMonads

其实只是如下结构

main :: IO ()
main = someIOMonad >>= (\x -> (anotherIOMonad >>= (\y -> otherIOMonads)))

就只是把do-notation变成一堆monad的bind，由于monad里的内容必须通过一个函数才能拿到，所以现在所有实现monad的方法里都是不停地嵌套才行，那使用monad的意义并不明确，monad本质上在于用一个语法可以重载任意流程控制(也算是一种CPS？)来节省代码，节约代码就是节约生命，写括号相当于慢性自杀，所以得想办法避开这个矛盾。

我们先简单实现一个Maybe，方便后面的说明：

class Maybe:
    def __init__(self, data, nothing=False):
        self.data = data
        self._nothing = nothing

    @staticmethod
    def nothing():
        return Maybe(None, nothing=True)

    @property
    def just(self):
        return not self._nothing

    @staticmethod
    def ret(value):
        return Maybe(value, nothing=False)

    def bind(self, f):
        if self._nothing:
            return self
        return f(self.data)

    def __repr__(self):
        if self.just:
            return f"Just {self.data}"
        return "Nothing"

以及一个简单的使用：

def my_input(prompt="input a number: ") -> Maybe:
    v = input(prompt)
    for x in v:
        if x not in "0123456789":
            return Maybe.nothing()
    return Maybe.ret(int(v))

考虑如下python代码：

def do_process():
    a1 = yield my_input('number1: ')  # anchor 1
    a2 = yield my_input('number2: ')  # anchor 2
    return Maybe(a1 + a2)

假设有gen = do_process()是取得的generator
那么next(gen)将得到anchor 1位置的my_input所得到的结果，对于剩下的那一部分来说，我们只要gen.send就可以把值发送给a1，刚好对应了bind行为

因此考虑函数reduce，它的作用就是把一个不断yield monad以及return monad的generator计算成最终return的结果

def reduce(generator):
    try:
        m_a = next(generator)
    except StopIteration as err:
        return err.value

    return m_a.bind(lambda a: reduce(输入a之后所形成的新的generator))

中间不难发现，我们需要一个输入a，然后生成新的generator的函数来保持继续reduce，我把这个过程称为partial_apply:

def partial_apply(generator, x):
    while True:
        try:
            x = yield generator.send(x)
        except StopIteration as err:
            return err.value

实现也非常简单，但是我发现，这样做相当于每次都生成了一个新的generator，并且旧的generator还在那里，这是没必要的，所以设计一个自己的Generator来辅助：

class _Generator:
    class _Null:
        pass

    def __init__(self, generator):
        self._generator = generator
        self._x = _Generator._Null

    @staticmethod
    def new(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            return _Generator(f(*args, **kwargs))
        return wrapper

    def __next__(self):
        if self._x is not _Generator._Null:
            x = self._x
            self._x = _Generator._Null
            return self._generator.send(x)
        return next(self._generator)

    def __iter__(self):
        return self

    def send(self, x):
        return self._generator.send(x)

    def partial_apply(self, x):
        self._x = x
        return self

    def __repr__(self):
        return f"Generator {self._generator.__repr__()}"

以及对应的reduce函数

def reduce(generator: _Generator):
    try:
        m_a = generator.__next__()
    except StopIteration as err:
        return err.value

    return m_a.bind(lambda a: reduce(generator.partial_apply(a)))

这里的递归我要说明一下，考虑Cont r a，由于continuation是需要bound f的运算结果来继续运算的，所以没办法把它没完全reduce的generator当成这个f来用，而且这个generator只能用reduce来计算（否则没办法正确bind），所以这个递归我暂时想不到好方法来避免，以及我不知道怎么手写栈来避开反复调用的开销，当然这个项目我放到GitHub上了，如果有谁有好想法可以发pull request

进一步的，为了方便使用，我还加了一个decorator来辅助，思路都很简单，就不多做赘述了

from functools import wraps


def do(generator_func):
    @wraps(generator_func)
    def wrapper(*args, **kwargs):
        return reduce(generator_func(*args, **kwargs))
    return wrapper

最终的使用效果是这样的(来自我那个仓里的实现):

from math import sqrt

from functionalpy.monad import do, Either


def safe_div(a, b) -> Either:
    if b == 0:
        return Either.left("ZeroDivision")
    return Either.ret(a/b)


def safe_sqrt(a) -> Either:
    if a < 0:
        return Either.left("Sqrt of a negative number")
    return Either.ret(sqrt(a))


@do
def cal(a, b):
    root = yield safe_sqrt(a)
    quotient = yield safe_div(root, b)
    return Either.ret(quotient)


if __name__ == '__main__':
    print(cal(1, 2))
    print(cal(1, 0))
    print(cal(-1, 1))
    print(cal(-1, 0))

输出

Right 0.5
Left ZeroDivision
Left Sqrt of a negative number
Left Sqrt of a negative number

虽然这个做法性能堪忧，但是对于js而言速度还是能接受的而且如果yield的次数不多的话，调用次数也不会太多，相比IO的等待以及callback hell，这些还是更容易接受的

然后是一个我在js里的实现（当然我js写的很烂了，而且很多的类型似乎因为缺少higher order type，也就是Haskell的kind或者idris的Type n，我没办法表达出来，所以到处都是any了，有谁能审计一下是最好的）

interface F<a, b> {
  (input: a): b
}

type Callback<a, r=void> = F<a, r>
type Cont<r, a> = F<F<a, r>, r>


class Pro<a> {
  cont: Cont<void, a>
  constructor(cont: Cont<void, a>) {
    this.cont = cont
  }
  run(callback: Callback<a, void>=console.log): void {
    this.cont(callback)
  }
  static ret<x>(value: x): Pro<x> {
    return new Pro(f => f(value))
  }
  bind<b>(f: F<a, Pro<b>>): Pro<b> {
    return new Pro(callback => {
      // this.callback: (a -> r) -> r
      // callback: (b -> r) -> r
      // f: a -> m b
      // return: r
      this.run(function(x: a) {
        f(x).run(callback)
      })
    })
  }
}

interface Monad<a> {
  ret: (value: a) => Monad<a>
  bind: <b>(f: (input: a) => Monad<b>) => Monad<b>
}

interface Result<a=any> {
  done: boolean
  value: a
}

interface Gen<a=any> {
  next: (x?: any) => Result<a>
}

class _Null {}

class _Gen<a=any> {
  gen: Gen
  x: any
  constructor(gen: Gen) {
    this.gen = gen
    this.x = new _Null()
  }
  partialApply(x: any) {
    this.x = x
    return this
  }
  next(x?: any): Result<a> {
    if(this.x instanceof _Null) {
      return this.gen.next(x)
    }
    let v = this.x
    this.x = new _Null()
    return this.gen.next(v)
  }
}

function reduce(gen: Gen): any {
  let _gen = new _Gen(gen)
  var result = _gen.next()
  if(result.done) {
    return result.value
  }
  var mA: Monad<any> = result.value
  return mA.bind(a => reduce(_gen.partialApply(a)))
}

function monad(generatorFunc: any): any {
  return function(...args: any[]): any {
    return reduce(generatorFunc(...args))
  }
}

function wrap(f: (...args: any[]) => void): any {
  return function(...args: any[]): Pro<any> {
    return new Pro(callback => {
      f(...args, callback)
    })
  }
}

function coroutine<a>(generatorFunc: any): Pro<a> {
  return monad(generatorFunc)
}

function p<a>(generatorFunc: F<void, any>): Pro<a> {
  return monad(generatorFunc)()
}

function pret<a>(v: a): Pro<a> {
  return Pro.ret(v)
}

使用例子是这样的，当然普通callback的调用也就不言而喻了

function callAsync(a: any, f: any) {return f(a)}
let call = wrap(callAsync)
p(function* () {
  let a = yield call(1)
  let b = yield call(2)
  return pret(a + b)
}).run(console.log)

其实这个东西是为了做这就事的：
考虑如下结构：

switch(status) {
    case null:
        weixinLogin
        webLogin
        getUserInfo
        break
    case weixin:
        webLogin
        getUserInfo
        break
    case login:
        getUserInfo
        break
}

然而注意的switch可以省去break，于是有

switch(status) {
    case null:
        weixinLogin
    case weixin:
        webLogin
    case login:
        getUserInfo
}

可以节约不少代码的

但是如果js只能异步调用

switch(status) {
    case null:
        weixinLogin({success: () => {
            webLogin({success: () => {
                getUserInfo({
                    success: callback
                })
            }})
        }})
        break
    case weixin:
        webLogin({success: ()=>{
            getUserInfo({
                success: callback
            })
        }})

        break
    case login:
        getUserInfo({success: callback})
        break
}

这样代码量就相当大了，所以说模改这个假装是同步的代码就很必要了

标签：__,return,generator,python,self,Monad,any,def
来源： https://blog.csdn.net/w17688977481/article/details/88552364