22. LangChain LCEL，用 | 串联AI的魔法语言

在 LangChain 生态中，LCEL（LangChain Expression Language）是一种新的编程范式。它用一个简单的|，改变了 AI 应用的构建方式。曾经需要编写大量胶水代码才能串联起来的提示词、模型和输出解析器，现在只需要一行代码就能完成：

chain = prompt | model | parser

1. LCEL 核心原理：| 运算符的秘密
LCEL 的核心设计哲学就是 "数据流优先"。所有组件都遵循统一的接口，数据从左向右沿着链条自动流动，无需开发者手动处理中间状态。

链式组合：LCEL 使用|（竖线）运算符串联 Prompt、模型与输出解析器形成 AI 流水线，数据严格按照从左到右的顺序传递。

底层实现：| 运算符本质上是 Python 中__or__方法的语法糖。

Runnable 基类：LangChain 中所有可执行组件都继承自Runnable基类，该基类重载了__or__方法，这就是为什么所有组件都能使用|进行链式组合的根本原因。

 2. 三步构建你的第一条 AI 链

下面，让我们从最简单的例子开始，用 LCEL 构建一个笑话生成器。这个例子将展示 LCEL 最核心的用法：串联提示词、模型和字符串输出解析器。

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser# 加载环境变量（包含API密钥、模型名称等）
load_dotenv()# 初始化模型（使用config_prefix支持多模型配置）
prefix = "QWEN"
model = init_chat_model(model_provider="openai",configurable_fields=["model", "api_key", "base_url"],config_prefix=prefix,temperature=0.5,max_tokens=500
)# 模型配置（从环境变量中读取）
config = {"configurable": {f"{prefix}_model": os.getenv(f"{prefix}_MODEL"),f"{prefix}_api_key": os.getenv(f"{prefix}_API_KEY"),f"{prefix}_base_url": os.getenv(f"{prefix}_BASE_URL")}
}# 创建提示词模板
prompt = ChatPromptTemplate.from_messages([{"role": "system", "content": "你是一个笑话大王"},{"role": "user", "content": "{new_input}"}]
)# 用LCEL构建完整链条
chain = prompt | model | StrOutputParser()# 执行链条，传入原始输入
result = chain.invoke({"new_input": "讲一个笑话"}, config=config)
print(result)

3. 进阶用法：什么可以入链？
LCEL 的强大之处在于它的包容性。几乎任何逻辑都可以被包装成 Runnable 组件，然后无缝接入流水线。但有一个核心规则必须牢记：

入链的内容必须是 Runnable 的子类，字典的值也必须符合这个规则。

下面是不同数据类型的处理方式对照表：

数据类型 处理方式 对应 Runnable子类 核心作用
静态值 包装为lambda _: 静态值 RunnableLambda 生成固定值
函数/lambda表达式 直接使用（自动转换） RunnableLambda 执行自定义逻辑
动态值 使用RunnablePassthrough() RunnablePassthrough 原样传递数据
字典 直接使用，值也要符合规范 RunnableParallel 并行执行多分支
让我们用一个例子来演示这些规则。我们将构建一个能生成结构化 JSON 数据的 AI 链，这在实际应用中非常常见。

RunnablePassthrough()：当你直接传入字符串"请描述一下张三..."时，RunnablePassthrough()会将这个字符串原样传递给new_input变量。

lambda _: parser.get_format_instructions()：这里的下划线_表示我们不关心上游输入，因为格式指令是固定的。这个 lambda 会被自动转换为RunnableLambda。

字典入链：整个字典会被转换为RunnableParallel对象，它会并行执行所有键对应的 Runnable，然后将结果合并为一个新的字典传递给下一个组件（这里是 prompt）。

import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
import asyncio
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.runnables import RunnablePassthroughload_dotenv()
prefix = "QWEN"
model = init_chat_model(model_provider="openai",configurable_fields=["model", "api_key", "base_url"],config_prefix=prefix,temperature=0.5,max_tokens=500
)
config = {"configurable": {f"{prefix}_model": os.getenv(f"{prefix}_MODEL"),f"{prefix}_api_key": os.getenv(f"{prefix}_API_KEY"),f"{prefix}_base_url": os.getenv(f"{prefix}_BASE_URL")}
}prompt = ChatPromptTemplate.from_messages([{"role": "system", "content": "你是一位有10年经验的资深软件工程师。{format_instructions}"},{"role": "user", "content": "{new_input}"}]
)parser = JsonOutputParser()
chain = ({"new_input": RunnablePassthrough(),"format_instructions": lambda _: parser.get_format_instructions()}| prompt| model| parser
)
result = chain.invoke("请描述一下张三这个人，包括姓名、年龄、职业和兴趣爱好。", config=config)
print(result)

4. 函数入链：无缝集成自定义逻辑

LCEL 最强大的特性之一就是可以轻松将任意 Python 函数接入流水线。函数会被自动转换为RunnableLambda，无需任何额外的包装代码。

让我们扩展上一个例子，添加一个函数将生成的 JSON 数据自动保存到本地文件。

import json
import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.runnables import RunnablePassthroughload_dotenv()prefix = "QWEN"
model = init_chat_model(model_provider="openai",configurable_fields=["model", "api_key", "base_url"],config_prefix=prefix
)config = {"configurable": {f"{prefix}_model": os.getenv(f"{prefix}_MODEL"),f"{prefix}_api_key": os.getenv(f"{prefix}_API_KEY"),f"{prefix}_base_url": os.getenv(f"{prefix}_BASE_URL")}
}# 自定义函数：保存JSON数据到文件
def save_json(data, filename="output.json"):try:with open(filename, "w", encoding="utf-8") as f:json.dump(data, f, ensure_ascii=False, indent=4)print(f"JSON数据已成功保存到 {filename}")except Exception as e:print(f"保存JSON文件时出错: {e}")return data  # 重要：返回数据以便继续传递给下一个组件
 
parser = JsonOutputParser()
prompt = ChatPromptTemplate.from_messages([{"role": "system", "content": "你是一位有10年经验的资深软件工程师。{format_instructions}"},{"role": "user", "content": "{new_input}"}]
)# 将自定义函数直接加入链条
chain = ({"new_input": RunnablePassthrough(),"format_instructions": lambda _: parser.get_format_instructions()}| prompt| model| parser| save_json  # 函数直接入链
         )result = chain.invoke("请描述一下张三这个人，包括姓名、年龄、职业和兴趣爱好。", config=config)
print(result)

运行这段代码后，你会在当前目录下看到一个output.json文件，内容如下：

{"姓名": "张三","年龄": 30,"职业": "软件工程师","兴趣爱好": ["阅读技术书籍","徒步旅行","围棋"]
}

重要提示：自定义函数必须返回数据，这样数据才能继续传递给链条中的下一个组件。如果函数没有返回值（返回None），那么链条的后续组件将收到None作为输入。