Langchain对管道操作符|的重构实现链式流程

目录 一、管道操作符二、LangChain 的链式流程三、源码剖析

一、管道操作符 Python 中的管道操作符 在 Python 3.10 及以上版本中，管道操作符 | 主要用于以下场景：

（1）集合操作 管道操作符可以用于集合的并集操作。

set1 = {1, 2, 3} set2 = {3, 4, 5} result = set1 | set2 # 并集: {1, 2, 3, 4, 5}

（2）类型联合 在类型注解中，| 可以表示“或”的关系。

def func(x: int | float) -> int | float: return x * 2

（3）模式匹配 在 match 语句中，| 可以用于匹配多个模式。

value = 3 match value: case 1 | 2 | 3: print("Small number") case _: print("Other number") 二、LangChain 的链式流程

在 LangChain 中，管道操作符 | 用于将多个步骤串联起来，形成一个链式流程。

chain = ( {"question": RunnablePassthrough()} | prompt_template | llm | output_parser )

每一步的输出会作为下一步的输入。

这种设计使得代码更加模块化和可读。

三、源码剖析

在 LangChain 的实际源码中，管道操作符的实现位于 langchain/schema/runnable.py 文件中。以下是一些关键代码片段：

class Runnable: def __or__(self, other: Union["Runnable", Callable]) -> "RunnableSequence": from langchain.schema.runnable import RunnableSequence return RunnableSequence([self, other])

or 方法，它是 Python 中的特殊方法（也称为魔术方法或双下方法）。or 方法用于重载 | 操作符的行为。具体来说，这段代码的作用是允许一个对象与另一个对象（或可调用对象）通过 | 操作符连接，并返回一个新的 RunnableSequence 对象。 or 方法定义了 | 操作符的行为。当你在代码中使用 | 操作符连接两个对象时，Python 会调用左侧对象的 or 方法，并将右侧对象作为参数传递。

step1 = Runnable() step2 = Runnable() # 使用 | 操作符连接两个对象 chain = step1 | step2

Union[“Runnable”, Callable] 的含义 Union 是 Python 类型注解中的一个工具，表示“或”的关系。Union[“Runnable”, Callable] 表示 other 可以是以下两种类型之一：

Runnable：一个实现了 Runnable 接口的对象。

Callable：一个可调用对象（如函数、lambda 表达式等）。

链式流程有递归的思想

from typing import Union, Callable class Runnable: def __or__(self, other: Union["Runnable", Callable]) -> "RunnableSequence": if callable(other): # 如果 other 是可调用对象，将其包装为 Runnable other = RunnableWrapper(other) return RunnableSequence(self, other) class RunnableSequence: def __init__(self, first: Runnable, second: Runnable): self.first = first self.second = second def run(self, input): # 先运行第一个步骤 intermediate = self.first.run(input) # 再运行第二个步骤 return self.second.run(intermediate) class RunnableWrapper(Runnable): def __init__(self, func: Callable): self.func = func def run(self, input): return self.func(input)