在学 Haskell 的过程中, 人产生了一种强烈的欲望: 我想把其中一些特征给搬运到 Python 里. 我知道里面有一些特征是比较好实现的, 比如 Monad. 但是, 另一些特征比如函数组合反而不好办. 当然 Monad 的 Py 实现也是通过一种间接路径来达成的——用一个附着在 Monad 类上的 bind 方法来实现 function apply. 这从一个侧面也反映了 Haskell 这类函数式语言和 Python 这种 OOP(姑且称为 OOP) 语言的区别: 前者更关心 transformation, 后者更关心 operation.

正是这种差别导致 python 上的 Monad 常规情况下只能 bind. 而 transformation 是更关心 operator 的: 数据的类别由其可以执行的操作 / 变换来定义. 然后这些操作 / 变换则是函数来实现. 此时一个函数就像一根线缆, 其头和尾有独特的标识, 数据则像是装有值的中继器. 当把一个函数应用到一个数据上时, 这个函数首先会检测中继器上的接口能否被自己插上去. 如果能, 则接入, 取出其中的值并做一些事情, 然后把值吐出来. 然而在 Python 里, 达到同样目的的思路却有点不同. 我们要么采用纯命令式的思维, 也就是说, 用极其有限的基础工具操作一块内存以达到处理的目的. 或者, 我们想象要处理的事情为一个群岛. 每个岛上存在一些船, 这些船通过把数据运进来处理然后返回到港口来实现数据操作. 后面这个看上去和 FP 的思路有点像, 但是在群岛上, 船虽然是通用的, 却更是属于岛的. 每个岛制造自己的船, 其对数据的处理也不是发生在船上, 而是在岛内.

由于这种从群岛到线缆系统的转换很有意思, 我就把其中一些操作给记录下来写在这里.

假 Monad

下面是一个非常粗糙, 并且不纯的 "Monad" 的实现. 在其中, 除了想达到数据绑定到函数的效果, 我还想做一点缓存. 所以加上了一些处理:

class DataMonad:
    def __init__(
        self,
        data: Dict[str, np.ndarray],
        fs: Dict[str, Func],
    ):
        self.values = data
        self.fs = fs

    def bind(self, f_name: str) -> DataMonad:
        if f_name in self.values.keys():
            return DataMonad(self.values, self.fs)
        try:
            factor_func = self.fs[f_name]
            assert callable(factor_func)
        except Exception:
            return DataMonad(self.values, self.fs)
        try:
            data_transformed = factor_func(self.values)
            nf = {f_name: data_transformed}
            if type(data_transformed) is tuple:
                nf = {f"{f_name}[{idx}]": val for idx, val in enumerate(data_transformed)}
            else:
                nf = {f_name: data_transformed}
            # disable changes
            for name in nf.keys():
                nf[name].flags.writeable = False

            new_data = im.Map({**self.values, **nf})
        except Exception as e:
            return DataMonad(self.values, self.fs)

这个类主要接受一个 numpy 数组作为值的字典, 不断通过 bind 对其进行特征工程. 可以看到, 我 apply 的并非函数, 而是一个名字. 所以严格意义上这不是 Monad. 另外, 为了让系统能快一点, 我加了一个缓存机制进去.

复合函数装饰器

复合函数就是要实现类似 f.g x 等价于 f(g(x)) 的效果. 本质来说, 这是一个 parser 的问题, 我们只是要找到平铺的表达式和嵌套表达式之间的转换方式. 最简单的解决方案就是写一个 composite(f,g) 函数, 让它可以生成复合以后的函数. 但是这么做太丑了. 而要在 Python 中 让 f(g(x)) 以某种方式等价于 f.g x 这种平铺描述的话, 最好的方式是把数据和操作一视同仁——也就是说我们需要某种东西把函数包起来, 如果这个东西碰到了一个函数, 就组合起来, 如果碰到了数据, 就开始求值. 这样一来就得到一个类似 f(g)(x) 的东西. 但是这还是不够好. 因为 f 此时看上去是一个高阶函数而非一个普通函数.

为了让所有的函数看起来平等, 也就是可以被 . 链接起来, 我们可以把函数都用一个盒子包起来, 然后对这个盒子的某个运算符, 比如 * 进行重载, 使其可以进行 f(g) 的操作. 不过这么做的话我们得牺牲一些东西, 比如可变参数, 关键词参数, 因为它们天然地就是非“线缆”的.

class Fun:
    def __init__(self, function, n_args: Optional[int] = None):
        self.function = function
        if n_args is None:
            fun_sig = inspect.signature(self.function)
            for name, param in fun_sig.parameters.items():
                if param.default is not inspect._empty:
                    raise Exception(f"Funs do not accept keyword arguments")
            self.n_args = len(fun_sig.parameters)
        else:
            assert type(n_args) is int and n_args > 0
            self.n_args = n_args

    def __mul__(self, other) -> Fun:
        n_args = other.n_args

        def new_func(*args):
            return self.__call__(other(*args))

        return Fun(new_func, n_args=n_args)

    def __call__(self, *args: Any, **kwargs: Any) -> Union[Fun, Any]:
        n_args_passed = len(args)
        if len(kwargs) > 0:
            raise Exception("Funs do not accept keyword arguments.")
        if n_args_passed > self.n_args:
            raise Exception("Too many arguments.")
        elif n_args_passed == self.n_args:
            return self.function(*args)
        else:
            return Fun(
                functools.partial(
                    self.function,
                    *args,
                ),
                self.n_args - n_args_passed,
            )

其使用方式如下:

@Fun
def _add(x, y, z):
    return x + y + z

@Fun
def _mul(x, y):
    return x * y

add23 = _add(2)(3)
add23(4)
_mul05 = _mul(0.5)
_mul05(2)
res = (add23 * _mul05 * _mul(-0.5))(2)

这个处理方式其实还是不够好. 理想情况下我们应该把数据也包起来. 而如果数据也包起来, 变换也包起来以后, 我们好像就模拟了某种 applicative functor. 不过这个目前我没做.

拟态中缀运算符

这个不是我想的, 是从 这篇文章 里看到的. 不得不说, 这个代码非常有创意, 看到的时候我愣了一下才反应过来其工作原理是一种“折叠”, 其中 | 充当了“折叠”工具人的角色:

from functools import partial

class Infix(object):
    def __init__(self, func):
        self.func = func
    def __or__(self, other):
        return self.func(other)
    def __ror__(self, other):
        return Infix(partial(self.func, other))
    def __call__(self, v1, v2):
        return self.func(v1, v2)

使用方式

>>> @Infix
... def add(x, y):
...     return x + y
...
>>> 5 |add| 6
11

做中缀运算符的目的是为了有更多的自定义算子可以被写成普通的数学表达式. 当然, 这个拟态中缀运算符是有局限的, 因为它并不能识别两边的参数的类型来定制操作.

一点想法

之所以强烈地想搬运一些 Haskell 的特性到 Python 上, 是因为我在写数据处理的代码的时候碰到了一些麻烦. 简而言之, 我的数据是一个类似字典的结构, 其要被一些未知的, 动态生成的函数进行流式处理, 我希望这些函数能很好地处理数据, 如果其中一些出现错误了, 不要影响其他函数的操作. 另外, 由于有一些函数的运算是建立在其他函数的结果上的, 所以如果这个 pipeline 能有缓存是最好的. 在用 Python 实现这个程序的过程中我碰到了很多麻烦. 比如函数通常“不纯”, 这就让很多动态生成的函数容易互相影响. 另外, 我希望这些函数可以是并行运行的, 然而要把动态生成的函数和 multiprocessing 之类的并行库结合起来是难上加难, 尤其是如果要用到 RPC 框架的话.

而此时我正好在看一些关于 Haskell 的书, 因为有一个自动微分库是用 Haskell 写的, 而我想将其结合到我自己的工作里. 在了解 Haskell 特性的过程中, 我被其一些概念所震撼, 尤其是一些和范畴论相关的概念. 在之前读一些关于 hippocampus 的论文过程中, 我接触了一些基本的范畴论概念, 感到其中对“关系”的抽象的魅力. 但我那时没有想过, 机器能以一种顺滑的方式实现这些概念. 同时, Haskell 的这些语言特性也让我产生了一些看待数据的新视角. 比如, 我以前认为, 列表就是列表而已, 一串连接起来的数字. 但是在 FP 的视角下, 列表是一个结构, 其被元素间的后继关系 morphism, 而非其他什么东西所定义. 这种关系可以被保持, 所以我们有了 functor. Monad 也可以用这种方式来看待. 一旦采用这种视角以后, 我们可以在数据的关系固定以后, 将注意力全部集中到 transformation 上. 由于数据的关系是固定的, 我们也可以更加方便地实现缓存. 此时整个系统就变成一个线缆 +WiFi 的组合. 线缆系统是函数, 而 WiFi 是一个缓存机制. 整个过程是安全而高效的.

但是当我想在 Python 里实现这个系统时, 我碰到了不少问题. 与其一个又一个地搞一些 ad-hoc 的方案, 我想不如干脆实现一套有限的模拟 FP 特性的系统, 让后面的工作能轻松一些. 可是在实现一些基础概念的过程中, 我发现这个事情和不同语言的语法结构有关. 拿那个中缀运算装饰器 Infix 来说, 它本质上就是干了一个栈的事情, 从而让把一个平铺的表达式给折叠起来了. 而复合函数那个, 则是将一个回调圣诞树通过尾递归改写成一个更自然的链式调用. 实际柯里化也是类似的, 很多代码里的技巧都是类似的, 如果把 bytecode 打印出来看一看, 这一点会更加明显.