Chain

之前在第一篇关于LangChain的文章中就说到了关于Chain的有趣之处，可以像Linux管道一样，把函数串联起来，执行很方便，步骤也很清晰，是一个比较优秀的功能：

1
2
# Linux管道
cat test.txt | awk '$2 > 10' | head -n 100 | sed 's/NA//g' | cut -f1,4
1
2
3
# LangChain的管道
chain = chat_prompt | llm | CommaSplitParse()
chain.invoke({"text": "colors"})

实际上这种管道功能我的印象是在R语言的dplyr中体现的淋漓尽致，后面我也查到在python中也有针对pandas的类似的管道包，后续会再单开一篇文章介绍。

符号重载

在C++里面会比较常听到这个操作，我记得之前是看到别人有个骚操作好像就是通过符号重载实现的，通过将不同的符号赋予不同的功能然后组合起来就成了外星文字，运行起来竟然能够得到有意义的结果。

符号重载就是给符号比如+、-、==、|、&等等赋予新的作用，比如我可以重新定义+表示为两个数字相减，也可以定义|为and，这样可以做code防御，在当今的时代还是有点意义的，可以迷惑人，但是自己写代码的时候最好还是做一些有意义的事情，比如这里把|重载为管道。

实现

下面是我和ChatGPT合作完成的代码，是一个普通版本，虽然标题是写的『拆解』，这里并不是拆解，是为了后续从Langchain中拆一些比如文本阅读器啊pdf阅读器之类，为了保持一致就这样起的标题，LangChain的高级的Chain的功能还没有实现，这里仅仅是提供一下思路：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class FunctionChain:
    
    def __init__(self, func=None):
        self.functions = []
        if func is not None:
            self.functions.append(func)

    def __or__(self, other):
        # 重载 | 运算符，添加函数到链中
        if isinstance(other, FunctionChain):
            self.functions.extend(other.functions)
        else:
            self.functions.append(other)
        return self
	
    # 如果想要以函数的形式调用就写成：
    # def __call__(self, value):
    def run(self, value):
        # 依次执行链中的函数
        result = value
        for func in self.functions:
            result = func(result)
        return result

# 定义一些示例函数
def func1(v):
    v.append(1)
    return v

def func2(v):
    v.append(2)
    return v

def func3(v):
    v.append(3)
    return v

def func4(v):
    v.append(4)
    return v

# 创建函数链并运行
chain = FunctionChain(func1) | func2 | func3 | func4
result = chain.run(["start"])
print(result)
# ['start', 1, 2, 3, 4]

原理其实就是把这些函数依次放到一个列表中，然后在调用run的时候依次执行，上一步的输出成为这一步的输入。但是怎么让函数听话的进入列表中，这里就是符号重载|，也就是 def __or__(self, other):将|的功能改变，self就是这个FunctionChain，other就是|右侧的函数，简单判断一下是不是已经是FunctionChain对象了，如果不是就把函数追加到列表中，同时返回self，也就是返回自己，这是常见的链式调用的做法，然后不断地把函数吞到自己肚子里（self.functions），又或者说这个链像是传染了一样。

利用返回自己也可以不使用符号重载，这样：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
class MyChain(object):
    def __init__(self):
        self.functions = list()
    
    def a(self, func):
        self.functions.append(func)
        return self
    
    def run(self, value):
        result = value
        for func in self.functions:
            result = func(result)
        return result
    
chain = MyChain()
chain.a(func1).a(func2).a(func3).a(func4)
result = chain.run(["start"])
print(result)

同样也可以。当然如果不使用这里的chain的话，想要得到结果就只能是一层套一层：

1
result = func4(func3(func2(func1(["start"]))))

上面的链其实已经可以了，但是我想能不能再做的漂亮一些，于是问ChatGPT弄一个装饰器的形式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def chainable(func):
    # 将普通函数装饰为支持链式调用的函数
    return FunctionChain(func)

# 将func1转变为FunctionChain对象
@chainable
def func1(v):
    v.append(1)
    return v

# 创建函数链并运行
# 这样写就美观很多了
chain = func1 | func2 | func3 | func4
result = chain(["start"])
print(result)

我猜测LangChain应该prompt就是一个这种Chain的对象，然后逐渐往后吞后面的函数。

最后玩一下+，-互换：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class MyNumber(object):
    
    def __init__(self, value):
        self.value = value

    def __add__(self, other):
        if isinstance(other, MyNumber):
            return MyNumber(self.value - other.value)
        return NotImplemented

    def __sub__(self, other):
        if isinstance(other, MyNumber):
            return MyNumber(self.value + other.value)
        return NotImplemented

    def __repr__(self):
        return str(self.value)

# 示例用法
a = MyNumber(10)
b = MyNumber(5)

print(a + b)  # 10 - 5 = 5
print(a - b)  # 10 + 5 = 15