LangChain LCEL链入门必看：零基础教程

LangChain LCEL 链式开发实战指南（新手必看）

本节深入解析 LangChain 中的核心机制——LCEL 链式编排。对于刚踏入大模型应用开发的工程师而言，掌握 LCEL 意味着从“功能实现”跃迁至“架构优雅”。本文从零开始，逐步拆解链式开发的设计哲学，每个环节均附可运行代码，跟随操作即可快速上手。

一、为什么必须学习 Chain / LCEL？

1.1 传统调用模式的痛点

在引入 Chain 之前，调用大模型的典型流程是：拼接提示词 → 调用模型 → 解析结果。每一步均需手动调用独立方法，代码碎片化，复用性极低。参考以下示例：

传统分步写法（冗余且脆弱）

# 1. 拼接提示词
prompt_msg = prompt.invoke({"city": "北京"})
# 2. 调用大模型
llm_result = llm.invoke(prompt_msg)
# 3. 解析结果
final_data = parser.invoke(llm_result)

功能虽能实现，但随着业务复杂度增加，代码可维护性急剧下降。

1.2 LCEL Chain 带来的变革

LCEL（LangChain Expression Language，LangChain 表达式语言）是官方为解决组件编排问题推出的声明式语法。其核心思想是利用管道符 | 将多个功能组件串联成一条流水线。

优化后，整个流程仅需一行关键代码：

chain = prompt | llm | parser
final_data = chain.invoke({"city": "北京"})

精简之外，更带来实质收益：

• 代码极简：多步逻辑合并为单一链路，可读性陡增。
• 接口统一：整条链原生支持 invoke、stream、batch 等方法，调用风格一致。
• 扩展灵活：支持串行、并行、条件分支、自定义逻辑，适配复杂业务场景。
• 高复用性：链可作为独立组件嵌入其他链路，实现模块化组装。

1.3 学完本文你将达成

• 透彻理解 Runnable、LCEL、Chain 等核心抽象。
• 熟练运用串行链路——覆盖 80% 的业务需求。
• 掌握并行、条件分支、数据透传、自定义函数等高阶组件。
• 实现流式输出、批量推理与结构化数据解析。
• 规避新手常见的陷阱，独立构建生产级链路。

二、环境准备：统一配置规范

2.1 安装依赖包

首先安装 LangChain 核心库及环境变量管理工具：

pip install langchain python-dotenv

2.2 配置环境变量

在项目根目录创建 .env 文件，填入 API 密钥和接口地址（以硅基流动为例）：

# .env 文件内容
SILICON_KEY=你的硅基API密钥
SILICON_BASE_URL=https://api.siliconflow.cn/v1

2.3 全局初始化 LLM（所有示例共享）

创建统一配置文件，集中初始化模型，后续示例无需重复加载环境：

# base_config.py 全局基础配置
import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

# 加载 .env 环境变量
load_dotenv()

# 将硅基接口映射为 OpenAI 兼容格式
os.environ["OPENAI_API_KEY"] = os.getenv("SILICON_KEY")
os.environ["OPENAI_BASE_URL"] = os.getenv("SILICON_BASE_URL")

# 初始化 DeepSeek-V3 模型（全局复用）
llm = init_chat_model("openai:deepseek-ai/DeepSeek-V3")

三、核心概念速览（新手必知）

先理清三个术语，后续学习事半功倍：

一句话：LCEL 就是用管道符将一个个“功能零件”组装成自动化流水线。

四、第一阶段：基础串行链（必掌握）

串行（RunnableSequence）是 LCEL 最基础、最常用的形态。结构为 组件1 | 组件2 | 组件3，按序执行。

标准链路结构：

提示词模板(ChatPromptTemplate) → 大模型(LLM) → 结果解析器(OutputParser)

4.1 最简串行链：文本问答

实现一个基础功能：传入问题，模型回答，输出纯文本。

完整代码：

from base_config import llm
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 1. 定义提示词模板（Runnable 组件）
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是小学老师，用两三句话通俗解释知识点"),
    ("human", "请讲解：{question}")
    # {question} 为占位符，调用时传入
])

# 2. 组装链路：提示词 → 模型 → 文本解析器
chain = prompt | llm | StrOutputParser()

# 3. 调用链：传入字典，键名必须匹配模板占位符
result = chain.invoke({"question": "什么是数独？"})

# 输出结果（纯字符串）
print(result)
print("结果类型：", type(result))

代码要点：

• ChatPromptTemplate 用于构建对话提示词，{question} 是动态占位符。
• StrOutputParser 将模型返回的消息对象转换为普通字符串。
• chain.invoke(字典) 的键必须与模板占位符一一对应，否则报错。
• 链路自动完成“填模板→调模型→转文本”全流程。

4.2 切换解析器：输出 JSON 格式

若需要结构化数据（如字典），只需替换末尾解析器，链路主体无需改动。

完整代码：

from base_config import llm
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser

# 提示词模板（要求模型输出标准 JSON）
prompt = ChatPromptTemplate.from_messages([
    ("system", "只输出JSON，不要多余文字，字段：name(城市名)、population(人口)"),
    ("human", "描述城市：{city}")
])

# 链路：提示词 → 模型 → JSON解析器
chain = prompt | llm | JsonOutputParser()

# 调用
result = chain.invoke({"city": "北京"})
print(result)
print("结果类型：", type(result))  # 

4.3 固定模板变量：partial 方法

当提示词中有固定不变的内容（如角色、规则），可用 partial 提前绑定，后续调用无需重复传参。

完整代码：

from base_config import llm
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是{role}，回答简洁明了"),
    ("human", "{question}")
])

# 提前固定 role=数学老师，后续只需传 question
fixed_prompt = prompt.partial(role="数学老师")
chain = fixed_prompt | llm | StrOutputParser()

# 多次调用，无需再传 role
print(chain.invoke({"question": "什么是余数？"}))
print(chain.invoke({"question": "什么是分数？"}))

4.4 生产级结构化输出（Pydantic）

正式项目中推荐使用 with_structured_output 替代传统解析器。结合 Pydantic 做类型校验，自动解析并验证格式，稳定性更高。

完整代码：

from base_config import llm
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field

# 1. 用 Pydantic 定义数据结构（约束字段类型与描述）
class Animal(BaseModel):
    name: str = Field(description="动物名称")
    emoji: str = Field(description="对应表情符号")
    age: int = Field(description="平均寿命")

# 2. 绑定结构化输出（内部自动解析与校验）
structured_llm = llm.with_structured_output(Animal)

# 3. 组装链路（无需额外解析器）
prompt = ChatPromptTemplate.from_messages([
    ("system", "按照要求描述动物"),
    ("human", "介绍一种{type}动物")
])
chain = prompt | structured_llm

# 调用
result = chain.invoke({"type": "家养宠物"})
print(result.model_dump())  # 转为字典输出

4.5 流式输出：实现打字机效果

在聊天界面或实时对话场景中，流式输出是刚需。使用 chain.stream() 可逐块返回内容，模拟逐字输出效果。

完整代码：

from base_config import llm
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是小学老师，通俗讲解知识点"),
    ("human", "讲解：{question}")
])
chain = prompt | llm | StrOutputParser()

# 流式遍历输出（逐块打印）
print("回答：", end="")
for chunk in chain.stream({"question": "天空为什么是蓝色的？"}):
    print(chunk, end="", flush=True)

4.6 批量处理：batch 高效执行

对于离线场景（批量翻译、批量摘要），使用 chain.batch() 可一次性处理多个独立请求，效率极高。

完整代码：

from base_config import llm
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "用一句话解释知识点"),
    ("human", "{question}")
])
chain = prompt | llm | StrOutputParser()

# 批量输入：多个独立请求列表
batch_inputs = [
    {"question": "什么是比喻？"},
    {"question": "什么是光合作用？"},
    {"question": "什么是加减法？"}
]

# 批量调用
batch_results = chain.batch(batch_inputs)

# 打印所有结果
for idx, res in enumerate(batch_results, 1):
    print(f"问题{idx}：{batch_inputs[idx-1]['question']}")
    print(f"答案：{res}\n")

4.7 链路复用：封装为通用函数

Chain 本身即可作为可复用组件，封装成函数后可在项目中反复调用。

from base_config import llm
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是{subject}老师，简短回答"),
    ("human", "{question}")
])
chain = prompt | llm | StrOutputParser()

# 封装函数
def ask_teacher(subject: str, question: str) -> str:
    return chain.invoke({"subject": subject, "question": question})

# 调用
print(ask_teacher("语文", "什么是成语？"))
print(ask_teacher("科学", "水为什么会结冰？"))

五、第二阶段：进阶组件（复杂业务必备）

掌握串行链后，学习四个高频进阶组件，分别解决并行执行、条件判断、数据透传、自定义逻辑四大场景。

5.1 并行执行 RunnableParallel

场景：同一份输入同时执行多条链路（例如同一主题同时生成笑话和诗歌）。多条链路并发运行，最终结果整合为一个字典。

两种等价写法：

• 显式写法：RunnableParallel(键=链路)，推荐新手使用。
• 隐式写法：直接传字典 {"键": 链路}，LCEL 自动识别为并行。

完整代码：

from base_config import llm
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel

# 定义两条独立子链路
joke_chain = ChatPromptTemplate.from_template("写一个{topic}相关的冷笑话") | llm | StrOutputParser()
poem_chain = ChatPromptTemplate.from_template("写两句{topic}相关的短诗") | llm | StrOutputParser()

# 并行组装：key 为结果标识
parallel_chain = RunnableParallel(
    joke=joke_chain,
    poem=poem_chain
)

# 调用：输入共享给所有子链路
result = parallel_chain.invoke({"topic": "晚霞"})

# 按 key 取结果
print("冷笑话：", result["joke"])
print("短诗：", result["poem"])

5.2 条件分支 RunnableBranch

场景：类似代码中的 if/elif/else，根据输入内容自动路由到不同链路。例如用户问笑话走幽默链路，问知识走科普链路。

语法规则：

RunnableBranch(
    (条件函数1, 分支链路1),
    (条件函数2, 分支链路2),
    默认链路  # 所有条件不满足时执行
)

完整代码：

from base_config import llm
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableBranch

# 1. 定义条件判断函数
def need_joke(input_dict: dict) -> bool:
    """判定用户是否在索要笑话"""
    question = input_dict["question"]
    return any(word in question for word in ["笑话", "搞笑", "段子"])

def need_poem(input_dict: dict) -> bool:
    """判定用户是否在索要诗歌"""
    question = input_dict["question"]
    return any(word in question for word in ["诗", "诗歌", "写一首"])

# 2. 定义不同分支链路
joke_chain = (ChatPromptTemplate.from_messages([
    ("system", "你是脱口秀演员，简短讲笑话"),
    ("human", "{question}")
]) | llm | StrOutputParser())

poem_chain = (ChatPromptTemplate.from_messages([
    ("system", "你是诗人，写4行以内短诗"),
    ("human", "{question}")
]) | llm | StrOutputParser())

default_chain = (ChatPromptTemplate.from_messages([
    ("system", "你是知识助手，通俗解答问题"),
    ("human", "{question}")
]) | llm | StrOutputParser())

# 3. 组装条件分支路由
router = RunnableBranch(
    (need_joke, joke_chain),
    (need_poem, poem_chain),
    default_chain  # 默认分支
)

# 测试不同问题
print(router.invoke({"question": "讲一个程序员的笑话"}))
print(router.invoke({"question": "写一首关于春天的诗"}))
print(router.invoke({"question": "地球是什么形状？"}))

5.3 数据透传 RunnablePassthrough

场景：保留原始输入数据的同时追加新字段。这在 RAG（检索增强生成）和多链路数据拼接中极为关键。

核心方法 RunnablePassthrough.assign(新字段=链路)：

• 保留原始输入字典。
• 自动执行指定链路，将结果作为新 key 追加到字典中。

模拟 RAG 场景（检索+问答）：

from base_config import llm
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda

# 模拟检索器：根据问题返回参考资料（真实项目替换为向量库检索）
mock_retriever = RunnableLambda(lambda x: f"【参考资料】关于{x['question']}的科普内容")

# 组装链路：透传原始问题 + 追加检索结果 → 传入提示词
rag_chain = (
    RunnablePassthrough.assign(docs=mock_retriever)  # 新增字段 docs=检索结果
    | ChatPromptTemplate.from_messages([
        ("system", "根据参考资料回答用户问题：{docs}"),
        ("human", "{question}")
    ])
    | llm
    | StrOutputParser()
)

# 调用：原始输入只有 question，链路自动追加 docs
result = rag_chain.invoke({"question": "数独是什么？"})
print(result)

5.4 自定义逻辑 RunnableLambda

场景：将普通 Python 函数嵌入链路，用于数据清洗、格式转换、外部接口调用等自定义操作。

完整代码：

from base_config import llm
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda

# 自定义处理函数：加工输入数据
def process_input(input_dict: dict) -> dict:
    new_dict = dict(input_dict)
    # 追加格式化字段
    new_dict["subject_label"] = f"【{new_dict['subject']}】"
    return new_dict

# 链路：自定义函数 → 提示词 → 模型 → 解析器
chain = (
    RunnableLambda(process_input)
    | ChatPromptTemplate.from_messages([
        ("system", "你是{subject_label}老师"),
        ("human", "讲解：{question}")
    ])
    | llm
    | StrOutputParser()
)

# 调用
result = chain.invoke({"subject": "数学", "question": "什么是几何图形？"})
print(result)

六、综合实战：多组件组合案例

真实业务中常需嵌套组合多个组件。以下两个案例覆盖主流组合模式。

案例1：并行生成 + 汇总总结（并行 + 透传 + 串行）

需求：输入一个主题，同时生成笑话和诗歌，再基于两者做总结。

from base_config import llm
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# 子链路
joke_chain = ChatPromptTemplate.from_template("{topic}的冷笑话") | llm | StrOutputParser()
poem_chain = ChatPromptTemplate.from_template("{topic}的两句短诗") | llm | StrOutputParser()

# 第一步：并行生成内容，并透传原始 topic
step1 = RunnablePassthrough.assign(
    joke=joke_chain,
    poem=poem_chain
)

# 第二步：汇总总结
summary_prompt = ChatPromptTemplate.from_messages([
    ("human", "主题：{topic}\n笑话：{joke}\n诗歌：{poem}\n简单总结一下内容")
])

# 完整链路
full_chain = step1 | summary_prompt | llm | StrOutputParser()

# 执行
print(full_chain.invoke({"topic": "大海"}))

案例2：智能路由问答（分支 + 串行）

基于用户输入自动分发到不同人设，前文分支部分已演示，可直接复用扩展。

七、新手高频踩坑与解决方案

整理开发中最常见的报错，按现象分类，便于快速排错：

八、知识总结与学习路线

8.1 核心语法速查表

8.2 解析器选型建议（生产环境）

• 纯文本输出：StrOutputParser()，适用于通用问答。
• JSON 字典输出：优先使用 llm.with_structured_output(Pydantic模型)（强校验），其次考虑 JsonOutputParser。
• 复杂结构化数据：统一使用 Pydantic + with_structured_output。

8.3 新手循序渐进学习路线

• 先吃透串行链（prompt | llm | 解析器），掌握 invoke、stream、batch、partial 等基础用法。
• 接着练习用 RunnableLambda 嵌入自定义函数。
• 然后学习 RunnableParallel 实现并行链路。
• 掌握 RunnablePassthrough.assign，为 RAG 打基础。
• 最后学习 RunnableBranch 条件分支。
• 尝试组合多个组件，开发完整的业务链路。

九、附录：最小可运行完整代码

新建 demo.py 文件，搭配前文创建的 base_config.py 和 .env 文件，直接运行验证环境：

# demo.py
from base_config import llm
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 基础问答链
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是小学老师，简短回答问题"),
    ("human", "{question}")
])
chain = prompt | llm | StrOutputParser()

# 调用
res = chain.invoke({"question": "什么是人工智能？"})
print(res)