LangChain Expression Language

LangChain Expression Language#

LangChain表达语言，简写为LCEL，是一种声明式的方式来链接LangChain组件。

优点#

以官网为准：https://python.langchain.com/v0.2/docs/concepts/#langchain-expression-language-lcel 我觉得最大的优点是 具象化特别好，很清晰的表达了chain的概念，而不是按照之前的构建对象来做。

Runnable接口#

Runnable接口可以简化chain的过程，LangChain的很多组件都实现了Runnable的协议，通过它实现了Pipline的概念，它重写了 |表达式，可以将他的输出作为写一个组件的输入。就像shell编程的管道一样,如：

cat /tmp/test.log | grep task_id

LangChain中实现的他组件有，chat models, LLMs， output parsers, retrievers, prompt templates等。

Runnable接口方法和功能如下

  • stream: 响应值流式返回

  • invoke: 调用

  • batch: 批量调用 同时他也支持异步方法

  • astream: 异步流式返回

  • ainvoke: 异步调用

  • abatch: 异步批量调用

  • astream_log: 异步流式返回，在最终的响应中，会增加中间步骤

  • astream_events: 链中发生的beta流事件（在 langchain-core 0.1.14 中引入）

LangChain中实现Runnable的component的输入输出#

Runnable接口的实现#

Runnable接口是LangChain Expression Language (LCEL)的核心。通过它指定了规范，实现了管道符运算 实现管道符运算的重点在于 重写了python中的__or__和__ror__方法

基本的思路

  1. 定义Runnable接口，重写__or__和__ror__方法。

  2. 在上面的方法中返回一个新的对象(RunnableSequence)，新的对象封装了此对象和传递进来的写一个阶段。

  3. 在封装的新对象中，也重写上面两个方法。

  4. 之后的操作都是在RunnableSequence和Runnable对象的__or__和__ror__中，将传递进来的对象封装为RunnableSequence,在RunnableSequence中记录了 中间步骤

  5. 通过层层包装，最后返回的是RunnableSequence对象，在这个对象中封装了所有的步骤。

Runnable#

class Runnable(Generic[Input, Output], ABC):
    name: Optional[str] = None
    """The name of the runnable. Used for debugging and tracing."""
    ## 封装为RunnableSequence
    def __or__(
        self,
        other: Union[
            Runnable[Any, Other],
            Callable[[Any], Other],
            Callable[[Iterator[Any]], Iterator[Other]],
            Mapping[str, Union[Runnable[Any, Other], Callable[[Any], Other], Any]],
        ],
    ) -> RunnableSerializable[Input, Other]:
        """Compose this runnable with another object to create a RunnableSequence."""
        return RunnableSequence(self, coerce_to_runnable(other))
    ## 封装为RunnableSequence，将other和self调换
    def __ror__(
        self,
        other: Union[
            Runnable[Other, Any],
            Callable[[Other], Any],
            Callable[[Iterator[Other]], Iterator[Any]],
            Mapping[str, Union[Runnable[Other, Any], Callable[[Other], Any], Any]],
        ],
    ) -> RunnableSerializable[Other, Output]:
        """Compose this runnable with another object to create a RunnableSequence."""
        return RunnableSequence(coerce_to_runnable(other), self)

RunnableSequence#

class RunnableSequence(RunnableSerializable[Input, Output]):
    first: Runnable[Input, Any]
    """The first runnable in the sequence."""
    middle: List[Runnable[Any, Any]] = Field(default_factory=list)
    """The middle runnables in the sequence."""
    last: Runnable[Any, Output]
    """The last runnable in the sequence."""
     
    ## 封装为 RunnableSequence
    def __or__(
        self,
        other: Union[
            Runnable[Any, Other],
            Callable[[Any], Other],
            Callable[[Iterator[Any]], Iterator[Other]],
            Mapping[str, Union[Runnable[Any, Other], Callable[[Any], Other], Any]],
        ],
    ) -> RunnableSerializable[Input, Other]:
        if isinstance(other, RunnableSequence):
            return RunnableSequence(
                self.first,
                *self.middle,
                self.last,
                other.first,
                *other.middle,
                other.last,
                name=self.name or other.name,
            )
        else:
            return RunnableSequence(
                self.first,
                *self.middle,
                self.last,
                coerce_to_runnable(other),
                name=self.name,
            )
    ## 这也是封装为RunnableSequence，只不过，将self和other的顺序变了一下
    def __ror__(
        self,
        other: Union[
            Runnable[Other, Any],
            Callable[[Other], Any],
            Callable[[Iterator[Other]], Iterator[Any]],
            Mapping[str, Union[Runnable[Other, Any], Callable[[Other], Any], Any]],
        ],
    ) -> RunnableSerializable[Other, Output]:
        if isinstance(other, RunnableSequence):
            return RunnableSequence(
                other.first,
                *other.middle,
                other.last,
                self.first,
                *self.middle,
                self.last,
                name=other.name or self.name,
            )
        else:
            return RunnableSequence(
                coerce_to_runnable(other),
                self.first,
                *self.middle,
                self.last,
                name=self.name,
            )

按照思路自己写一个#

from typing import List


class CusotmerPiplineComponent:
    ## 通过
    steps: List["CustomerRunnable"]

    def __init__(self, *steps):
        self.steps = steps

    def __or__(self, other: "CusotmerPiplineComponent"):
        return CusotmerPiplineComponent(
            *self.steps, other
        )

    def __ror__(self, other: "CusotmerPiplineComponent"):
        print(other)

    def __str__(self):
        return ",".join(map(lambda item: item.name, self.steps))

    def invoke(self,**kwargs):
        print("kwargs:", kwargs)
        for item in self.steps:
            print(item.name)


class CustomerRunnable:
    name: str

    def __init__(self, name: str):
        self.name = name

    def __or__(self, other: "CustomerRunnable"):
        # print(self.name,",",other.name)
        return CusotmerPiplineComponent(
            self, other
        )

    def __ror__(self, other: "CustomerRunnable"):
        if not other:
            return
        # print(self.name,"+",other.name)

    def __str__(self):
        return self.name


if __name__ == '__main__':
    res = CustomerRunnable("a") | CustomerRunnable("b") | CustomerRunnable("c")
    print(res.steps)
    print(type(res))
    res.invoke(**{"input":12})

在上面的例子中，会包装为CusotmerPiplineComponent对象，最终的到的res是CusotmerPiplineComponent,最终通过invoke调用。

到此，这一章节就结束了。

Contents