在 LangChain 生态中LCELLangChain Expression Language是一种新的编程范式。它用一个简单的|改变了 AI 应用的构建方式。曾经需要编写大量胶水代码才能串联起来的提示词、模型和输出解析器现在只需要一行代码就能完成chain prompt | model | parser动画视频在《22. LangChain LCEL用 | 串联AI的魔法语言》。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_provideropenai, configurable_fields[model, api_key, base_url], config_prefixprefix, temperature0.5, max_tokens500 ) # 模型配置从环境变量中读取 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: 讲一个笑话}, configconfig) 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_provideropenai, configurable_fields[model, api_key, base_url], config_prefixprefix, temperature0.5, max_tokens500 ) 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(请描述一下张三这个人包括姓名、年龄、职业和兴趣爱好。, configconfig) 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_provideropenai, configurable_fields[model, api_key, base_url], config_prefixprefix ) 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, filenameoutput.json): try: with open(filename, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent4) print(fJSON数据已成功保存到 {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(请描述一下张三这个人包括姓名、年龄、职业和兴趣爱好。, configconfig) print(result)运行这段代码后你会在当前目录下看到一个output.json文件内容如下{ 姓名: 张三, 年龄: 30, 职业: 软件工程师, 兴趣爱好: [ 阅读技术书籍, 徒步旅行, 围棋 ] }重要提示自定义函数必须返回数据这样数据才能继续传递给链条中的下一个组件。如果函数没有返回值返回None那么链条的后续组件将收到None作为输入。