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 已经输出字符串：

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 中非常常见的模式。

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 表达式语言）中的精妙设计。

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

假设我们最终调用：

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

1. {"text": "..."} 作为输入传给 summarize_chain。
2. summarize_chain 内部可能是这样的：

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

3. 它输出一个 字符串，例如：

   "LangChain 是一个用于构建大语言模型应用的框架。"

   这个字符串会作为 lambda 函数的输入 x。

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

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

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

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 模板类似：

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

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

  "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 字段内容

最佳实践总结

1. 从简单开始，逐步复杂化：先掌握 prompt | model | parser 基础模式，确保跑通。然后逐步加入工具、分支、记忆等组件。不要试图一开始就构建一个完美的复杂Agent。
2. **善用RunnableParallel**：需要多任务并行时优先考虑
3. 保持链的可读性：复杂逻辑拆分为多个子链
4. **利用 .bind()**：固定参数如 model.bind(temperature=0)
5. 使用LangSmith：生产环境务必开启追踪调试