导言

LangChain Expression Language（LCEL）是LangChain框架中一种强大和灵活的语言，用于描述和构建复杂的链式调用。它允许开发人员以声明式的方式定义数据的转换和流程，并与LangChain的其他组件无缝集成。本文将对其设计理念进行介绍，并做简单的代码实战。

LangChain Expression Language

设计来源及语法

前文提到，使用LangChain Expression Language主要有两个目的：统一的接口和组合原语。如果有过任务调度引擎，大数据，规则引擎等开发经验的同学应该马上可以抽象出来核心其实就是DAG(有向无环的模式)。LangChain Expression Language就是LangChain框架中一种强大而灵活的领域特定语言(DSL)，用于描述和构建复杂的链式调用。下面看其标准流程：

设计模型输入INPUT参数

将用户所需传递给模型的参数设计好，如：用户输入传递为 {“topic”: “赛博”}

结合Prompt模版进行参数渲染

通过RunnablePassthrough函数获取用户输入传递给prompt模版，然后在使用构造提示prompt后将其用于构造 PromptValue，以给大模型输入

发送大语言模型进行处理，输出结果

model组件采用生成的PromptValue提示，并传递到 OpenAI LLM 模型进行评估。模型生成的输出是一个ChatMessage对象。

通过格式化进行输出

output_parser组件接收 aChatMessage并将其转换为 Python 字符串的结果，从 invoke 方法返回。

示意图如下：

编程实战

本文基础环境为anacoda，使用notebook进行示例：

首先，pip安装相关的包：

%pip install --upgrade --quiet  langchain-core langchain-community langchain-openai

然后配置环境变量，这里以OpenAI的gpt3.5为例：

%env OPENAI_API_KEY="你的OPENAPI_KEY"

import库，进行相关包的导入：

from langchain_core.output_parsers 
import StrOutputParserfrom langchain_core.prompts 
import ChatPromptTemplatefrom langchain_openai import ChatOpenAI

核心逻辑如下：

prompt = ChatPromptTemplate.from_template("给我讲个简短的冷笑话 {topic}")
model = ChatOpenAI(model="gpt-3.5-turbo")
output_parser = StrOutputParser()
chain = prompt | model | output_parser
chain.invoke({"topic": "赛博"})

运行结果如下：

由于有读者国内无法访问OpenAI接口的问题，这里也给出了平替的解决方案，就是国产大模型GLM，智谱AI于24年1月16日发布的GLM-4，增强了工具(tool)调用能力，API接口方式与OpenAI几乎完全相同，推理能力在GPT3.5和GPT-4之间。如果有不会申请智谱AI的接口的读者，也可以后台私信詹特曼，后台进行解答。

同样，获取到了智谱AI开放平台的API后先在notebook设置环境变量：

%env ZHIPU_API_KEY="你的智谱平台API_KEY"

由于LangChain的community集成了大语言模型的接口其实并没有随模型及时更新，所以关于智谱平台的API_KEY客户端的鉴权我们需要自己手动实现，要先倒入依赖包：

import osimport time
import jwtfrom langchain_core.output_parsers 
import StrOutputParserfrom langchain_core.prompts 
import ChatPromptTemplatefrom langchain_openai import ChatOpenAI

这里要自己实现根据API_KEY生成客户端令牌(要不是翻了zhipu库的源码，也不会料想是这样，也是自己没详细看智谱的API文档)：

def generate_token(apikey: str, exp_seconds: int):  
    try:    
        id, secret = apikey.split(".")  
    except Exception as e:    
        raise Exception("invalid apikey", e)  
    payload = {     
        "api_key": id,     
        "exp": int(round(time.time() * 1000)) + exp_seconds * 1000,     
        "timestamp": int(round(time.time() * 1000)),   
    }  
    return jwt.encode(     
        payload,     
        secret,     
        algorithm="HS256",     
        headers={"alg": "HS256", "sign_type": "SIGN"},   
    )

prompt = ChatPromptTemplate.from_template("给我讲个简短的冷笑话 {topic}")
model = ChatOpenAI(model_name="glm-4",
openai_api_base="https://open.bigmodel.cn/api/paas/v4",
openai_api_key=generate_token(os.environ.get("ZHIPU_API_KEY"), 600), streaming=False)
output_parser = StrOutputParser()
chain = prompt | model | output_parser 
chain.invoke({"topic": "赛博"})