langchain LCEL原理分析

golangLeetcode

发布于 2026-03-18 17:33:39

1170

在使用langchain的时候，我们需要先定义输入提示词模板、大模型、输出模板解析器，然后invoke传入参数，获取LLM返回的结果：

from langchain_community.chat_modelsimportChatOllama
from langchain.schemaimportHumanMessage
from langchain_core.output_parsersimportStrOutputParser
chat = ChatOllama(model="llama3", temperature=0)
output_parser = StrOutputParser()
response = chat.invoke([HumanMessage(content="你好啊, AI小助手")])
output_parser.invoke(response)

看到这里会发现，定义了很多中间变量，写起来特别啰嗦，熟悉链式调用的大佬坐不住了，使用链式调用不是一句话的事么。于是， LCEL是 LangChain Expression Language诞生了，它可以使用类似链式调用的语法定义了LangChain自己的DSL。上面的代码就可以简化为下面一句话

from langchain_community.chat_models import ChatOllama
from langchain.schema import HumanMessage
from langchain_core.output_parsers import StrOutputParser
(ChatOllama(model="llama3", temperature=0) | StrOutputParser()).invoke([HumanMessage(content="你好啊, AI小助手")])

我们发现，它并不是传统的链式调用写法，而是使用了类似unix管道"|"，将整个调用过程串起来。但是Python语言不支持管道，这是如何实现的呢？

Python的魔法函数__or__,Python中的__or__魔法方法用于实现对象的按位或(|)运算符重载。当对象出现在|运算符左侧时自动调用__or__方法，必须返回一个新对象而不修改原操作数，典型应用在集合操作和字典合并场景。下来我们来自己实现python的管道调用

# -- coding: utf-8 --
from datetime import datetime
class MyPipeline:
    def __init__(self, func):
        self.func = func
    def __or__(self, otherfunc):
        def _func(*args, **kwargs):
            return otherfunc(self.func(*args, **kwargs))
        return MyPipeline(_func)
    def __call__(self, *args, **kwargs) :
        return self.func(*args, **kwargs)
    def invoke(self, *args, **kwargs):
        return self.__call__(*args, **kwargs)

        def hello(name) -> str:
    return f"hello，{name}。"
def welcome(greeting) -> str:
    return f"欢迎来到我的公众号, {greeting}"
def attach_date(mstr) -> str:
    return f"{mstr} 现在是{datetime.now().strftime('%Y-%m-%d')}。"

hello = MyPipeline(hello)
welcome = MyPipeline(welcome)
attach_date = MyPipeline(attach_date)
# my_chain = mychain = hello.__or__(welcome).__or__(attach_date)
my_chain = hello | welcome | attach_date
print(my_chain.invoke("golang算法架构leetcode技术php"))

可以看到，我们首先定义了invoke方法，在内部会调用call方法，把传入的参数往下传递进去

    def invoke(self, *args, **kwargs):
        return self.__call__(*args, **kwargs)

然后重载了__call__方法，调用自己的成员变量函数func

    def __call__(self, *args, **kwargs) :
        return self.func(*args, **kwargs)

func变量是怎么赋值的呢，这里就用到了前面提到的管道方法，我们知道管道方法类似于链式调用，每次调用完需要返回当前类的对象，所以，在类的构造函数里传入了func参数，保证invoke的时候从最新产生的对象里调用到func方法

class MyPipeline:
    def __init__(self, func):
        self.func = func

最后，我们终于到了，解开管道运算符的真实面纱的时刻了，要知道我们链式调用，对象是越往后调用对象越往后产生，但是我们invoke的时候，先定义的对象的函数要最先被调用，整体来说函数调用顺序是一个队列。

    def __or__(self, otherfunc):
        def _func(*args, **kwargs):
            return otherfunc(self.func(*args, **kwargs))
        return MyPipeline(_func)

所以，在重载“｜”作为管道符号的时候，定义了一个内部函数，把当前管道符右侧传递进来的函数的调用作为一个闭包返回给外侧，最终作为新对象的参数初始化了新对象的func函数。这样invoke的时候就会调用新产生的这个闭包。仔细看会发现，self.func被作为了闭包的参数进行了调用，这样就保证了，当前对象的func函数比逼管道产生的新函数更早调用，达到队列的效果，因为它需要为新对象调用进行参数初始化。这也是很多框架的中间件喜欢用的套路，洋葱模型。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2025-05-19，如有侵权请联系 cloudcommunity@tencent.com 删除

闭包