LangChain模块之Runnable源码分析

源码环境：

• langchain==0.3.27

• langchain-core ==0.3.75

说明：本文中的代码为了简洁，移除了大部分类型注释

Runnable抽象类

类所在位置：langchain_core.runnables.base.Runnable

Runnable对象是 langchian中的核心对象，在链式调用中每个元素都是Runnable 对象。其实现了以下功能：

• 实现了LCEL，使得可以通过 |来链接不同的元素

• 定义了统一的invoke、batch、stream及对应的异步方法

• 定义了统一的重试策略、生命周期监听器等方法

• 定义了统一的调试与追踪功能

LCEL中的连接符

LCEL中通过“｜”来实现不同对象的管道式操作。其原本为 python中的 “或”语法。

当python解释器尝试计算 x | y 时，它会遵循一个特定的查找顺序：

1. 首先尝试调用左操作数 x 的 __or__ 方法：x.__or__(y)

2. 如果 x 没有实现 __or__ 方法，或者 x.__or__(y) 返回了 NotImplemented（表示它不知道如何处理 y 这种类型），那么解释器会转而尝试调用右操作数 y 的__ror__ 方法：y.__ror__(x)

因此，在Runnable 对象中重写了 __or__ 及__ror__ 方法。使得可以使用 ｜ 作为不同对象的运算符。在 langchain中不同对象通过|运算符后最终变成了一个RunnableSequence 对象。

# 重写的相关源码在：566 ～590 行
def __or__(self,other):
    """Compose this ``Runnable`` with another object to create a ``RunnableSequence``."""  # noqa: E501
    return RunnableSequence(self, coerce_to_runnable(other))
​
def __ror__(self,other):
    """Compose this ``Runnable`` with another object to create a ``RunnableSequence``."""  # noqa: E501
    return RunnableSequence(coerce_to_runnable(other), self)

其中的coerce_to_runnable主要是将非runnable对象转为runnable对象。

# 相关源码在：5978～5995 行
def coerce_to_runnable(thing: RunnableLike) -> Runnable[Input, Output]:
    """Coerce a ``Runnable``-like object into a ``Runnable``.
​
    Args:
        thing: A ``Runnable``-like object.
​
    Returns:
        A ``Runnable``.
​
    Raises:
        TypeError: If the object is not ``Runnable``-like.
    """
    # 如果为Runnable则直接返回
    if isinstance(thing, Runnable):
        return thing
    # 如果为异步生成器则转为RunnableGenerator
    if is_async_generator(thing) or inspect.isgeneratorfunction(thing):
        return RunnableGenerator(thing)
    # 是一个可被调用的对象，则直接使用RunnableLambda转为Runnable对象
    if callable(thing):
        return RunnableLambda(cast("Callable[[Input], Output]", thing))
    # 如果是字典则转为RunnableParallel对象（可并行的节点）。
    if isinstance(thing, dict):
        return cast("Runnable[Input, Output]", RunnableParallel(thing))
    msg = (
        f"Expected a Runnable, callable or dict."
        f"Instead got an unsupported type: {type(thing)}"
    )
    raise TypeError(msg)

RunnableSequence对象