2026-04-11发表2026-04-12更新AI / LangChain1 小时读完 (大约6888个字)

LangChain：LCEL与管道符(|)

LCEL（LangChain Expression Language—表达式语言）是 LangChain 框架中构建 Agent 智能体的核心与灵魂。它提供了一套声明式的语法，能用管道符（|）像拼接积木一样，将各种功能组件（模型、提示、解析器、工具等）组合成一个清晰的数据处理流水线。

LCEL概念

LCEL是什么

LCEL（LangChain Expression Language ）是LangChain的声明式组合语法，使用**管道符(|)**将各种组件（如模型、提示、解析器、工具等）连接成复杂的执行链。

设计理念是：**声明式编排 **和 **一切皆可运行(Everything is a Runnable)**。

声明式编排：只需要清晰地描述“我想做什么”（例如：用这个提示词，调用那个模型，然后解析结果），而不是一步步地指示“我该如何做”。这让代码逻辑变得极其直观。
一切皆 Runnable：在 LCEL 中，无论是提示词模板、大模型、输出解析器，还是自定义组件，都被抽象为 Runnable 对象，拥有 invoke、stream、batch 等统一接口。它们就像标准化的积木块，可以通过统一的接口被轻松组合和调用。

通过统一接口（Runnable）+ 管道组合（|），将复杂的工作流表示为简单的管道操作。

管道符(`|`)是什么

管道符 | 的灵感来自 Unix 管道，它将左边组件的输出作为右边组件的输入。

管道符 |是LCEL的连接运算符，它将多个 Runnable 组件按顺序连接起来，形成一个数据流。前一个组件的输出会自动作为下一个组件的输入。整个链条本身也是一个 Runnable，可以被调用。

例如，A | B意味着“将A的输出作为B的输入”。这种语法极其直观，让代码读起来就像描述一个数据处理流水线。

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 基础组件
prompt = ChatPromptTemplate.from_template("告诉我关于{topic}的趣事")
model = ChatOpenAI(model="gpt-4")
output_parser = StrOutputParser()

# 使用 | 组合成链
chain = prompt | model | output_parser

# 执行
result = chain.invoke({"topic": "人工智能"})
print(result)

执行流程图解

用户输入: {"topic": "人工智能"}
    ↓
prompt | model | output_parser
    ↓      ↓         ↓
  格式化 > 调用LLM > 解析输出
  提示词   生成回复   转为字符串
                     ↓
最终结果: "人工智能是..."

组件与管道符(|)

组件：任何实现了 Runnable 接口的对象（如 ChatPromptTemplate、ChatOpenAI、StrOutputParser 等）。
**管道符 (|)**：将一个 Runnable 的输出传递给下一个 Runnable 作为输入，从而构建工作流。
工作流：用户输入 → Prompt 模板（格式化）→ LLM 模型（生成回答）→ 输出解析器（提取纯文本）→ 最终输出。

可以把管道符 | 想象成工厂里的传送带，而每个组件则是传送带上的一个工作站。原料（用户输入）从传送带一端进入，依次经过每个工作站（提示模板、语言模型、输出解析器等）的加工，最终在另一端产出成品（最终答案）。

LCEL组件

关键组件详解

组件	作用	示例
`RunnablePassthrough`	原样传递输入	`{"query": RunnablePassthrough()}`
`RunnableParallel`	并行执行多个分支	`RunnableParallel(a=chain1, b=chain2)`
`RunnableLambda`	包装自定义函数	`RunnableLambda(lambda x: x.upper())`
`RunnableBranch`	条件分支	`RunnableBranch((condition, chain1), default)`
`itemgetter`	提取字典字段	`itemgetter("messages")`

常用组件的输入输出类型

PromptTemplate：input: dict → output: PromptValue
ChatModel：input: PromptValue / str / list[BaseMessage] → output: BaseMessage
StrOutputParser：input: BaseMessage → output: str
RunnableLambda：任意可调用对象，需要你手动保证类型匹配。

LCEL优势

代码更简洁：从嵌套调用变为线性的管道式组合，自动处理组件间的数据传递和格式转换，大幅提升可读性，让代码更优雅。
原生支持高级级特性：LCEL 链本身具备流式输出（stream）、并行处理（batch）、异步调用（ainvoke）和自动重试、回退等机制，也是生产环境所需的能力，可以轻松附加到链上，无需额外代码。
强大的组合性与灵活性：通过 RunnableParallel、RunnableBranch 等工具，能轻松构建并行、分支、循环等复杂的数据流。
Agent的本质：一个复杂的Agent本身就是由多个LCEL链（工具调用链、思考链、响应生成链）组合而成的。
统一的接口与生态：Runnable 协议统一了所有组件的行为，所有链都有 .invoke()， .ainvoke()， .stream()， .batch()方法，使得复用和集成变得异常简单。

LCEL写法

等价写法

可以用更显式的 RunnableLambda 写出同样的功能（不推荐，但便于理解）：

from langchain_core.runnables import RunnableLambda

def to_summary_dict(msg):
    return {"summary": msg.content}   # 如果 msg 是 AIMessage

full_chain = summarize_chain | RunnableLambda(to_summary_dict) | translate_chain

或者如果 summarize_chain 已经输出字符串：

1	full_chain = summarize_chain \| (lambda s: {"summary": s}) \| translate_chain

优雅写法

如果需要保留原始输入的同时添加新字段，或者不想写 lambda，可以使用 RunnablePassthrough.assign，如下：

from langchain_core.runnables import RunnablePassthrough

# 假设 summarize_chain 输出字符串
full_chain = (
    {"text": RunnablePassthrough()}          # 保留原始输入（可选）
    | summarize_chain                        # 输出字符串
    | RunnablePassthrough.assign(summary=lambda s: s)  # 将字符串包装成 {"summary": s}
    | translate_chain
)

但 lambda 方式已经足够清晰，是 LCEL 中非常常见的模式。

LCE使用示例

示例1：基础使用

最经典的用法： prompt | llm | output_parser

语种翻译链：

# 1. 导入所需组件
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# 2. 初始化组件
# 提示词模板：定义任务，并声明一个变量 {language} 和 {input_text}
prompt = ChatPromptTemplate.from_template(
    "请将以下内容翻译成{language}：{input_text}"
)

# 语言模型：此处使用 OpenAI 兼容接口
llm = ChatOpenAI(
    model="deepseek-chat",  # 替换为你的模型
    base_url="你的 Base URL",
    api_key="你的 API Key",
)

# 输出解析器：将模型的 AIMessage 输出提取为纯文本字符串
output_parser = StrOutputParser()

# 3. 使用管道符 (|) 构建 LCEL 链
chain = prompt | llm | output_parser

# 4. 调用链(四种方式)
## 4.1 同步调用
# .invoke() 方法传入一个字典，其中的键需要匹配 prompt 中定义的变量
response = chain.invoke({
    "language": "英文",
    "input_text": "LangChain 是一个用于构建大语言模型应用的框架。"
})

# 4.2 异步调用 (适合Web服务)
# async_result = await chain.ainvoke(...)

# 4.3 流式调用 (适合实时输出)
# for chunk in chain.stream(...):
#     print(chunk, end="", flush=True)

# 4.4 批量调用
# results = chain.batch([{"language": "法语", "input_text": "你好"}, {"language": "日语", "input_text": "谢谢"}])

print(response)
# 预期输出: "LangChain is a framework for building large language model applications."

一个简单的问答链：

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 1. 定义各个组件
# 提示词模板：一个 Runnable，接收 'topic' 变量，输出格式化后的提示词
prompt = ChatPromptTemplate.from_template("你是一位技术专家。请用通俗易懂的语言解释一下：{topic}。")

# 模型：一个 Runnable，接收提示词，输出 AI 响应
# 请确保已设置好环境变量 OPENAI_API_KEY
model = ChatOpenAI(model="gpt-4o")

# 输出解析器：一个 Runnable，将 AI 的原始响应（AIMessage）解析为字符串
output_parser = StrOutputParser()

# 2. 使用管道符 '|' 将它们组合成一个链
# 数据流向：{"topic": "..."} -> prompt -> model -> output_parser -> str
chain = prompt | model | output_parser

# 3. 调用这个链
# 同步调用
result = chain.invoke({"topic": "什么是量子纠缠？"})
print(result)

# 流式调用，提升用户体验
print("流式输出：", end="", flush=True)
for chunk in chain.stream({"topic": "什么是递归？"}):
    print(chunk, end="", flush=True)
print() # 换行

# 批量调用，提高效率
results = chain.batch([{"topic": "API"}, {"topic": "微服务"}])
for i, res in enumerate(results):
    print(f"结果 {i}: {res}")

简单的问答链，提示词指定角色：

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 1. 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位专业的{role}。"),
    ("human", "{question}")
])

# 2. 初始化模型
model = ChatOpenAI(model="gpt-4", temperature=0.7)

# 3. 输出解析器
parser = StrOutputParser()

# 4. 使用管道符组合
chain = prompt | model | parser

# 5. 执行
response = chain.invoke({
    "role": "Python专家",
    "question": "解释Python中的装饰器"
})
print(response)

关键解读：

prompt | llm | output_parser创建了一个清晰的数据流：输入字典-> 提示模板-> LLM模型-> 输出解析器-> 最终字符串。
每个组件都实现了 Runnable接口，知道如何接收上游输入，并产生下游能理解的输出。

示例2：复杂逻辑

LCEL 的强大之处在于它能轻松处理分支和并行等复杂逻辑，而无需编写复杂的 if/else 或循环。

1. 并行处理 (RunnableParallel)

当需要同时处理多个独立任务时（例如，从多个知识库检索信息），RunnableParallel 并发执行可以显著提升效率。

假设想同时获取一个城市的简介和最佳旅游时间：

from langchain_core.runnables import RunnableParallel

# 假设我们想同时获取一个城市的简介和最佳旅游时间
chain = RunnableParallel(
    intro=prompt_intro | llm | StrOutputParser(),   # 任务1：生成简介
    best_time=prompt_time | llm | StrOutputParser() # 任务2：推荐时间
) | final_prompt | llm | StrOutputParser() # 将两个结果合并后，再交给LLM处理

假设想同时生成摘要和提取关键词：

from langchain_core.runnables import RunnableParallel

# 定义两个独立的子任务
summary_chain = ... # 生成摘要的链
keywords_chain = ... # 提取关键词的链

# 使用 RunnableParallel 并行执行
parallel_chain = RunnableParallel(
    summary=summary_chain,
    keywords=keywords_chain,
)

# 调用后，会同时得到两个结果
result = parallel_chain.invoke({"input": "一篇关于AI的长文章"})
print(result['summary'])
print(result['keywords'])

并行处理三个任务：

from langchain_core.runnables import RunnableParallel

# 创建并行处理链
parallel_chain = RunnableParallel(
    summary=(
        ChatPromptTemplate.from_template("总结以下内容：{text}")
        | ChatOpenAI()
        | StrOutputParser()
    ),
    keywords=(
        ChatPromptTemplate.from_template("从以下内容提取关键词：{text}")
        | ChatOpenAI()
        | StrOutputParser()
    ),
    sentiment=(
        ChatPromptTemplate.from_template("分析以下内容的情感：{text}")
        | ChatOpenAI()
        | StrOutputParser()
    )
)

# 执行 - 三个任务并行运行
result = parallel_chain.invoke({
    "text": "OpenAI发布了GPT-5，性能大幅提升，令人兴奋！"
})
# 返回: {"summary": "...", "keywords": "...", "sentiment": "..."}

2. 条件分支 (RunnableBranch)

RunnableBranch 可以根据输入的条件，动态选择不同的处理路径，实现“if-else”逻辑。

根据问的是天气还是新闻路由：

from langchain_core.runnables import RunnableBranch

# 根据问题的主题，选择不同的提示模板
branch_chain = RunnableBranch(
    (lambda x: "天气" in x["question"], weather_prompt | llm),
    (lambda x: "新闻" in x["question"], news_prompt | llm),
    default_prompt | llm  # 默认路径
)

根据文档内容长度路由：

from langchain_core.runnables import RunnableBranch

# 定义分支：根据输入长度选择不同处理方式
branch_chain = RunnableBranch(
    (lambda x: len(x["text"]) > 100, prompt_long | llm | output_parser),  # 长文本总结
    (lambda x: len(x["text"]) <= 100, prompt_short | llm | output_parser), # 短文本扩写
    default_prompt | llm | output_parser  # 默认情况
)

# 也可以与管道符结合
full_chain = {"text": RunnablePassthrough()} | branch_chain
result = full_chain.invoke({"text": "这是一个很长的文档内容..."})

判断值条件路由：

from langchain_core.runnables import RunnableBranch

# 假设我们有一个分类器链，能判断用户意图是 'code', 'data' 还是 'general'
# classifier_chain = ... (一个输出意图分类的链)

# 定义针对不同意图的子链
code_chain = ... # 处理编程问题的链
data_chain = ... # 处理数据分析问题的链
general_chain = ... # 处理通用问题的链

# 使用 RunnableBranch 创建路由链
routing_chain = RunnableBranch(
    (lambda x: x["intent"] == "code", code_chain),
    (lambda x: x["intent"] == "data", data_chain),
    general_chain, # 默认分支
)

# 最终链：先分类，再路由
# full_chain = classifier_chain | routing_chain

链嵌套使用路由：

from langchain_core.runnables import RunnableBranch, RunnableLambda

# 定义路由函数
def route_by_topic(info):
    topic = info.get("topic", "").lower()
    if "code" in topic or "编程" in topic:
        return "programming"
    return "general"

# 创建不同分支的链
programming_chain = (
    ChatPromptTemplate.from_template("作为编程专家，回答：{question}")
    | ChatOpenAI()
    | StrOutputParser()
)

general_chain = (
    ChatPromptTemplate.from_template("作为通用助手，回答：{question}")
    | ChatOpenAI()
    | StrOutputParser()
)

# 使用 RunnableBranch 实现条件路由
branch = RunnableBranch(
    (lambda x: route_by_topic(x) == "programming", programming_chain),
    general_chain  # 默认分支
)

# 完整链
full_chain = {
    "topic": lambda x: x["topic"],
    "question": lambda x: x["question"]
} | branch

# 执行
result = full_chain.invoke({
    "topic": "编程问题",
    "question": "Python列表推导式怎么用？"
})

3. 数据传递 (`RunnablePassthrough`)

RunnablePassthrough 就像一个“数据通道”，在复杂的字典流中，它可以原封不动地传递某个值，常与 RunnableParallel 配合使用。

from langchain_core.runnables import RunnablePassthrough

# 构建 RAG 链的典型用法：
# 1. 检索上下文 (retriever)
# 2. 透传原始问题 (RunnablePassthrough())
# 3. 将两者打包成一个字典，供 Prompt 使用
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

4. 自定义逻辑 (RunnableLambda)

RunnableLambda 允许你将任意的 Python 函数包装成一个 Runnable 对象，无缝嵌入 LCEL 链中，实现复杂的数据转换或逻辑判断。

带检索的RAG链：

from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# 假设已有向量数据库
vectorstore = FAISS.load_local("faiss_index", OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 提示模板
template = """基于以下上下文回答问题：
{context}

问题：{question}
"""
prompt = ChatPromptTemplate.from_template(template)

# 模型和解析器
model = ChatOpenAI(model="gpt-4")
parser = StrOutputParser()

# 辅助函数：格式化检索结果
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

# 将其包装为 Runnable 后，就可以像其他组件一样使用
# chain = retriever | RunnableLambda(format_docs) | prompt | llm | parser
# 构建 RAG 链
rag_chain = (
    {
        "context": retriever | format_docs,  # 检索并格式化文档
        "question": RunnablePassthrough()     # 直接传递用户问题
    }
    | prompt
    | model
    | parser
)

# 执行
result = rag_chain.invoke("什么是LCEL？")
print(result)

使用 RunnableLambda 把普通函数包装成 Runnable 对象。

from langchain_core.runnables import RunnableLambda

# 定义一个普通函数
def extract_keywords(text: str) -> list:
    # 这里是一个简单的模拟
    keywords = ["模型", "人工智能", "学习"]
    return [kw for kw in keywords if kw in text]

# 使用 RunnableLambda 将其转换为 Runnable
keyword_extractor = RunnableLambda(extract_keywords)

# 集成到链中：先生成答案，再提取关键词
complex_chain = prompt | model | StrOutputParser() | keyword_extractor

result = complex_chain.invoke({"topic": "深度学习"})
print(result) # 输出: ['模型', '学习']

5. 错误处理 (RunnableWithFallback)

RunnableWithFallback 可以为一个 Runnable 组件配置一个或多个“备胎”。当主组件执行失败时，会自动调用备用组件，提升系统的健壮性。

from langchain_core.runnables import RunnableWithFallback

# 如果主模型（chat_model）调用失败，就使用备用模型（fallback_model）
robust_chain = RunnableWithFallback(
    runnable=chat_prompt | chat_model,
    fallbacks=[chat_prompt | fallback_model]
)

使用注意事项

LCEL表达式语言的管道符|是有严格的先后顺序要求，顺序决定了数据流的类型和结构，错误的顺序会导致运行时错误。

LCEL 中的 | 管道操作符并不是简单的“拼接”，而是将前一个组件的输出作为后一个组件的输入。因此，顺序必须遵循各组件之间 输入/输出类型的匹配规则。

两侧都是Runnable对象

|操作符两侧都必须是Runnable对象。最常见的错误是直接将一个普通 Python 对象（如字符串）放在 | 后面。LCEL 期望每一个操作单元都是一个 Runnable，这是它的设计原则。

输入输出类型匹配

核心规则：类型必须匹配

每个 Runnable 组件（如 PromptTemplate、ChatModel、OutputParser）都有其预期的输入类型和输出类型，所以在写 LCEL 链时，首先明确每个组件的输入类型和输出类型。管道传递的本质就是：A | B 要能工作，必须满足 A.output_type ⊆ B.input_type（即 A 的输出是 B 能接受的输入）。

| 连接的两个组件，左边组件的输出类型必须是**右边组件的输入类型。例如，ChatPromptTemplate 输出 PromptValue，ChatOpenAI 接收 PromptValue 并输出 AIMessage，而 StrOutputParser 接收 AIMessage 并输出 str。

最常见的：PromptTemplate → ChatModel → OutputParser

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("讲一个关于{topic}的笑话")
model = ChatOpenAI()
parser = StrOutputParser()

chain = prompt | model | parser   # 正确顺序
result = chain.invoke({"topic": "程序员"})

输入 {"topic": "程序员"} → PromptTemplate 期望一个 dict，输出一个 PromptValue（或字符串化的消息列表）。
PromptValue → ChatModel 期望 PromptValue 或消息序列，输出 BaseMessage。
BaseMessage → StrOutputParser 期望 BaseMessage，输出 str。

错误1：ChatModel 在前，PromptTemplate 在后

# 错误：类型不匹配
chain = prompt | model | prompt  # ❌ 会报错，model输出是AIMessage，不能直接给prompt

# 正确：使用输出解析器转换
chain = prompt | model | StrOutputParser() | another_prompt

model 输出 BaseMessage，但 prompt 期望输入 dict（包含模板变量）。BaseMessage 无法直接转换为 dict，因此抛出 TypeError。

错误2：OutputParser 在前，ChatModel 在后

1 2	# 错误示例 chain = parser \| model # ❌ 会报错

parser 输出 str（或其他解析后的类型），而 model 通常期望 PromptValue、字符串或消息序列。直接传递 str 在某些模型下可能勉强工作（如旧版 LLM 接口），但不推荐，且会丢失 prompt 模板的能力。更重要的是，如果你的 parser 输出的是复杂结构（如 dict），model 很可能无法处理。

错误3：两个 PromptTemplate 直接连接

prompt1 = ChatPromptTemplate.from_template("先总结：{text}")
prompt2 = ChatPromptTemplate.from_template("再翻译：{summary}")

chain = prompt1 | prompt2   # ❌ 通常无意义且会报错

prompt1 输出 PromptValue，但 prompt2 期望 dict。除非你手动包装一个适配器，否则类型不匹配。

字典格式的正确使用

LCEL 链条中的数据通常以字典（dict）的形式在组件间传递。确保每个组件的输入变量名与上一步输出的键（key）匹配。例如，prompt 需要 {topic}，那么前一个组件的输出或 invoke 的输入就必须包含 topic 这个键。

# 正确：RunnableParallel 或 字典字面量
chain = {
    "context": retriever | format_docs,
    "question": RunnablePassthrough()
} | prompt | model

# 注意：字典的每个值必须是 Runnable 或可调用对象

提示词模板变量匹配

这是最常见的错误之一。如果 PromptTemplate 中定义了 {variable_name}，那么在调用链时，输入的字典中必须包含 variable_name 这个键，否则会抛出 ValueError。

输出解析器失败

输出解析器（如 JsonOutputParser）期望模型输出特定格式。如果模型输出不符合预期，解析会失败。调试时，可以先移除解析器，打印模型的原始输出，检查格式问题，然后通过优化提示词（明确告知模型输出格式）来解决。

上下文窗口超限

如果输入文本过长，加上提示词模板后可能会超出模型的最大 token 限制。可以使用 LangChain 的 TextSplitter 将长文档切分成小块来处理。

版本兼容性与调试

遇到类型错误，可以分段执行（如先 chain1 = prompt | model，再单独 chain1.invoke(...) 观察输出类型）来调试。

LCEL 链本身就像一个“黑盒”，内部的中间结果不容易直接查看。可以使用 langchain_core.callbacks 模块中的 ConsoleCallbackHandler，它能在执行时打印每个组件的输入和输出。

LangChain 迭代非常快，不同版本间可能存在不兼容。建议使用 langchain-core 和 langchain-community 等拆分后的包，并锁定版本。

对于复杂的链条，调试可能比较困难，可以考虑使用 LangSmith 这样的工具进行全链路跟踪和可视化。

优先使用管道符`|`

优先使用 | 而非 .pipe()：管道符 | 是组合 LCEL 链的首选和推荐方式，因为它更简洁、直观，符合 Python 语言习惯。.pipe() 方法作为备选，主要在需要动态构建链时使用，例如在循环中根据条件添加组件。

拥抱现代化API

拥抱现代化API，舍弃旧用法：在 langchain.chains 模块中的 LLMChain、SequentialChain 等旧式链（Legacy Chains）已被标记为过时（deprecated），并将在未来的版本中被移除。请确保学习和实践都基于最新的 LCEL 语法。

异步支持

# LCEL 链自动支持异步
result = await chain.ainvoke({"topic": "AI"})
# 或批量处理
results = await chain.abatch([{"topic": "AI"}, {"topic": "ML"}])

调试与追踪

# 查看链的结构
print(chain.get_graph().print_ascii())

# 开启详细日志
import langchain
langchain.debug = True

状态传递

# 使用 RunnablePassthrough.assign 保留原始输入
from langchain_core.runnables import RunnablePassthrough

chain = RunnablePassthrough.assign(
    summary=(lambda x: x["text"][:100])  # 添加新字段，保留原有字段
) | prompt | model

调整顺序或适配类型

当需要实现非标准顺序时，可以使用 LCEL 提供的适配器组件：

场景	解决方案	示例
前一步输出类型不满足下一步输入	`RunnableLambda` 或 `RunnablePassthrough` 进行转换	`chain = prompt1
需要同时传递多个输入（分支合并）	`RunnableParallel`	`{“summary”: chain1, “topic”: chain2}
在管道中插入一个无输入的函数	`RunnablePassthrough.assign()`	保留原有输入并添加新字段
从 `BaseMessage` 中提取字符串内容	使用 `StrOutputParser()`	`model

LCEL 的 | 管道要求前后类型匹配。当不匹配时，使用 RunnableLambda、RunnablePassthrough、StrOutputParser 等适配器进行类型转换，或重新设计管道逻辑。

示例：修复“两个 prompt 不能直接连”的问题

from langchain_core.runnables import RunnableLambda, RunnablePassthrough

prompt1 = ChatPromptTemplate.from_template("总结：{text}")
prompt2 = ChatPromptTemplate.from_template("将总结翻译成英文：{summary}")

model = ChatOpenAI()

# 错误写法：prompt1 | model | prompt2   ❌ (model输出BaseMessage, prompt2要dict)
# 正确写法：提取 model 输出的文本内容，并重命名为 summary
chain = (
    prompt1 
    | model 
    | StrOutputParser()                     # BaseMessage -> str
    | (lambda summary: {"summary": summary}) # str -> dict {"summary": ...}
    | prompt2 
    | model 
    | StrOutputParser()
)

result = chain.invoke({"text": "LangChain 是一个强大的框架"})
print(result)

LCEL示例解读

示例代码

# 用LCEL完成 总结 → 翻译 的两步任务
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

summarize_prompt = PromptTemplate.from_template("用中文总结这段文本:\n\n{text}")
translate_prompt = PromptTemplate.from_template("将以下内容翻译成英文:\n\n{summary}")

model = ChatOpenAI(model="gpt-4o-mini")

summarize_chain = summarize_prompt | model
translate_chain = translate_prompt | model

# 组合链，并在中间使用一个lambda函数提取文本内容
# (lambda x: {"summary": x.content}) 类型转换桥接：提取内容并包装成字典，满足下游的输入格式
full_chain = summarize_chain | (lambda x: {"summary": x.content}) | translate_chain

# 执行
result = full_chain.invoke({"text": "人工智能正在改变世界..."})
print(result.content)

详细拆解这段代码，理解它在 LCEL（LangChain 表达式语言）中的精妙设计。

1	full_chain = summarize_chain \| (lambda x: {"summary": x.content}) \| translate_chain

这段代码构建了一个两步骤的流水线：先总结，再翻译。它体现了 LCEL 用 | 管道符连接不同组件时的类型转换思想。

三个组成部分

组件	类型	输入类型	输出类型
`summarize_chain`	一个 LCEL 链（通常是 `PromptTemplate	ChatModel	StrOutputParser`）
`lambda x: {"summary": x.content}`	一个 Python 函数，被 `RunnableLambda` 自动包装	取决于上游输出：这里上游是 `summarize_chain` 输出 `str`，所以 `x` 是 `str`	`dict`（例如 `{"summary": "这是总结内容"}`）
`translate_chain`	另一个 LCEL 链，通常期望输入是一个包含 `summary` 键的字典	`dict`（必须包含 `summary` 字段）	`str`（翻译后的文本）

数据流动过程

Step1：执行`summarize_chain`

假设我们最终调用：

1	result = full_chain.invoke({"text": "LangChain 是一个用于构建 LLM 应用的框架..."})

{"text": "..."} 作为输入传给 summarize_chain。

summarize_chain 内部可能是这样的：

1
2
3

summarize_prompt = PromptTemplate.from_template("用中文总结以下内容：\n{text}")
model = ChatOpenAI()
summarize_chain = summarize_prompt | model | StrOutputParser()

它输出一个 字符串，例如：
1
"LangChain 是一个用于构建大语言模型应用的框架。"
这个字符串会作为 lambda 函数的输入 x。

Step2：通过lambda函数进行类型转换

1	lambda x: {"summary": x.content}

注意，这里有一个隐藏陷阱：x 是字符串，而字符串没有 .content 属性！实际上，更常见的写法应该是：

1	lambda x: {"summary": x} # 因为 x 已经是纯文本字符串

但原代码写的是 x.content，这暗示 x 可能是一个 BaseMessage 对象（例如 AIMessage）。

这说明 summarize_chain 的输出很可能**没有经过 StrOutputParser**，而是直接输出了 ChatModel 的原始 AIMessage。

修正理解（更符合实际场景），通常我们写两步链时，会这样设计：

summarize_chain = prompt | model                     # 输出 AIMessage
translate_chain = translate_prompt | model | StrOutputParser()

full_chain = summarize_chain | (lambda msg: {"summary": msg.content}) | translate_chain

summarize_chain 输出 AIMessage，其 content 属性是总结文本。
lambda 函数接收 msg（一个 AIMessage 对象），提取 msg.content，并包装成字典 {"summary": ...}。
这个字典正好是 translate_chain 所期望的输入（它的 prompt 模板中有 {summary} 变量）。

Step3：执行`translate_chain`

translate_chain 收到 {"summary": "LangChain 是一个用于构建大语言模型应用的框架。"}

它内部的 prompt 模板类似：

1	translate_prompt = PromptTemplate.from_template("将以下内容翻译成英文：\n{summary}")

经过模型和解析器后，输出一个字符串，例如：

1	"LangChain is a framework for building large language model applications."

最终 full_chain.invoke(...) 返回这个翻译后的字符串。

lambda函数

核心原因：连接两个对数据结构要求不同的组件。

summarize_chain 输出的是纯文本（或 AIMessage）。
translate_chain 期望的输入是一个字典，键名为 summary。

直接连接 summarize_chain | translate_chain 会失败，因为类型不匹配：

如果 translate_chain 的第一个组件是 PromptTemplate，它要求输入为 dict（提供模板变量 summary）。
而 summarize_chain 输出的是 str 或 AIMessage，不能直接塞进 PromptTemplate。

lambda 函数充当了一个适配器（Adapter），在 LCEL 中被自动提升为 RunnableLambda，负责将上游的输出转换成下游需要的格式。

示例总结

代码片段	作用
`summarize_chain`	第一步：生成总结（输出文本或消息对象）
`	(lambda x: {“summary”: x.content})`
`translate_chain`	第二步：翻译字典中的 `summary` 字段内容

最佳实践总结

从简单开始：先掌握 prompt | model | parser 基础模式
**善用 RunnableParallel**：需要多任务并行时优先考虑
保持链的可读性：复杂逻辑拆分为多个子链
**利用 .bind()**：固定参数如 model.bind(temperature=0)
使用 LangSmith：生产环境务必开启追踪调试

LangChain：LCEL与管道符(|)

http://blog.gxitsky.com/2026/04/11/AI-LangChain-034-Chain-LCEL/

作者

光星

发布于

2026-04-11

更新于

2026-04-12

LangChain：LCEL与管道符(|)

LCEL概念

LCEL是什么

管道符(|)是什么

组件与管道符(|)

LCEL组件

关键组件详解

常用组件的输入输出类型

LCEL优势

LCEL写法

等价写法

优雅写法

LCE使用示例

示例1：基础使用

示例2：复杂逻辑

1. 并行处理 (RunnableParallel)

2. 条件分支 (RunnableBranch)

3. 数据传递 (RunnablePassthrough)

4. 自定义逻辑 (RunnableLambda)

5. 错误处理 (RunnableWithFallback)

使用注意事项

两侧都是Runnable对象

输入输出类型匹配

字典格式的正确使用

提示词模板变量匹配

输出解析器失败

上下文窗口超限

版本兼容性与调试

优先使用管道符|

拥抱现代化API

异步支持

调试与追踪

状态传递

调整顺序或适配类型

LCEL示例解读

示例代码

三个组成部分

数据流动过程

Step1：执行summarize_chain

Step2：通过lambda函数进行类型转换

Step3：执行translate_chain

lambda函数

示例总结

最佳实践总结

作者

发布于

更新于

许可协议

喜欢这篇文章？打赏一下作者吧

评论

目录

管道符(`|`)是什么

3. 数据传递 (`RunnablePassthrough`)

优先使用管道符`|`

Step1：执行`summarize_chain`

Step3：执行`translate_chain`