在 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 RunnablePassthrough load_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 RunnablePassthrough load_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作为输入。