LangChain 核心概念：LCEL 链式调用

LangChain 的文档里有一段示例代码：

python

chain = prompt | model | output_parser
result = chain.invoke({"question": "什么是向量数据库？"})

其中 | 符号把两个对象连在一起，然后直接 .invoke() 就出结果了。这个 | 不是普通的按位或，而是 LangChain 重写了组件的 __or__ 方法，把它变成了"管道"操作符。

这就是 LCEL（LangChain Expression Language，LangChain 的表达式语言：用 | 符号把各种 AI 组件串联成处理流水线的一套语法）。搞懂它，LangChain 就通了一大半。

1. 为什么需要 LCEL：从手动拼接到管道组合

LCEL 链式调用结构 — Prompt | LLM | OutputParser 的管道组合

在 LCEL 出现之前（LangChain v0.1 时代），写一个简单的 LLM 调用链需要这样：

python

# 旧的方式（不推荐，已废弃）
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import OpenAI

llm = OpenAI()
prompt = PromptTemplate(
    input_variables=["concept"],
    template="请用100字解释：{concept}"
)
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run("向量数据库")

这种写法有几个问题：

LLMChain 是一个固定的封装，想加一个 OutputParser 就要用另一种 Chain
不同的 Chain 类型（LLMChain、ConversationChain 等）接口不一致
想并行执行两个 Chain，没有标准方式

LCEL 用统一的 Runnable 接口解决了这些问题。所有组件（Model、Prompt、Parser、Retriever 等）都实现同一套接口，可以自由组合：

python

# LCEL 方式（推荐）
chain = prompt | llm | output_parser
# 想加 Retriever？直接加进来：
rag_chain = retriever | prompt | llm | output_parser

这就是 Unix 管道思想的精髓：每个组件做好一件事，然后自由组合。

2. LangChain 到底是什么

LangChain 不是大语言模型，它不做模型训练，也不做推理计算。它是一个框架，专门用来把模型和各种外部工具、数据源串起来。

打个比方：GPT-4、Claude、DeepSeek 是发动机，而 LangChain 是底盘和变速箱。可以换不同的发动机（换模型），但底盘的逻辑保持不变——输入数据、处理数据、调用模型、解析结果、返回给用户。

没有 LangChain，构建一个 RAG 应用（检索增强生成：先从知识库里检索相关内容，再把检索结果交给 LLM 生成回答），需要自己处理文档分块、向量化、存储、检索、拼 Prompt、调 API、解析输出……每一步都要手写。有了 LangChain，这些组件都封装好了，只需要把它们连起来。

连起来这件事，就是 LCEL 干的。

3. LCEL：像 Linux 管道一样串组件

LCEL 的核心思想来自 Unix 管道。在 Linux 里，可以这样写：

bash

cat file.txt | grep "error" | sort | uniq

每个命令处理上一个命令的输出，结果传给下一个命令。LCEL 把同样的思路搬到了 Python 里：

python

chain = prompt | model | output_parser
result = chain.invoke({"question": "什么是向量数据库？"})

左边的输出，直接传给右边的输入。这个设计很优雅——不用关心中间怎么传，只管定义"从哪来，到哪去"。

Runnable 接口：所有组件的共同语言

LCEL 能工作，是因为所有组件都实现了 Runnable 接口。这个接口定义了三个核心方法：

.invoke(input)：同步调用，返回结果
.stream(input)：流式调用，逐步返回
.batch(inputs)：批量调用，并发执行

任何实现了 Runnable 的对象，都可以用 | 连接到链里。这是 LangChain 最核心的设计决策——统一接口让组合成为可能。

4. 三个核心组件

LCEL 里最常用的三个组件是 PromptTemplate、ChatModel 和 OutputParser，它们构成了最基础的链。

PromptTemplate：负责把用户输入组装成完整的 Prompt。定义一个模板，留几个占位符，调用时传入参数，它帮你填好。

ChatModel：负责调用大语言模型。LangChain 封装了 OpenAI、Anthropic、DeepSeek、通义千问等几十个模型，换模型只需要改一行代码。

OutputParser：负责解析模型输出。模型返回的是 AIMessage 对象，StrOutputParser 帮你把它转成普通字符串；JsonOutputParser 帮你把 JSON 格式的输出转成字典。

5. 完整示例：从零写一个 LCEL 链

下面这段代码是完整可运行的，用 DeepSeek 模型（OpenAI 兼容接口）：

python

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

# 配置 DeepSeek（OpenAI 兼容接口）
llm = ChatOpenAI(
    model="deepseek-chat",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com",
    temperature=0.7,
)

# 定义 Prompt 模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位专业的技术文章作者，用简洁清晰的语言解释技术概念。"),
    ("human", "请用 100 字以内解释：{concept}"),
])

# 定义输出解析器
output_parser = StrOutputParser()

# 用 LCEL 把三个组件串起来
chain = prompt | llm | output_parser

# 调用链
result = chain.invoke({"concept": "向量数据库"})
print(result)

运行这段代码发生了什么：

prompt 接收 {"concept": "向量数据库"}，生成完整的消息列表（系统消息 + 用户消息）
llm 接收消息列表，调用 DeepSeek API，返回 AIMessage 对象
output_parser 接收 AIMessage，提取 .content 字段，返回纯字符串

整个流程，只写了三行核心逻辑。

LCEL 组件失败时的恢复策略

LCEL 链中任何一个组件抛出异常，整个链都会终止。生产环境中有两种常见处理方式：

方式一：在链外用 try-except 捕获，适合对调用方透明地处理错误：

python

# ⚠️ 生产环境建议补充：
# import time
# from openai import RateLimitError, APITimeoutError
#
# try:
#     result = chain.invoke({"concept": "向量数据库"})
# except RateLimitError:
#     time.sleep(60)  # 限流，等待后重试
#     result = chain.invoke({"concept": "向量数据库"})
# except APITimeoutError:
#     raise  # 超时，向上抛出
# except Exception as e:
#     raise  # 其他错误向上抛出，由调用方决定如何处理

方式二：用 .with_fallbacks() 设置备用链，适合需要降级处理的场景（如主模型限流时切换到备用模型）：

python

# 备用链示例：主链失败时自动切换到备用模型
# fallback_llm = ChatOpenAI(model="gpt-4o-mini", ...)
# fallback_chain = prompt | fallback_llm | output_parser
# chain_with_fallback = chain.with_fallbacks([fallback_chain])
# result = chain_with_fallback.invoke({"concept": "向量数据库"})

预期输出示例：

code

向量数据库是一种专门存储和检索向量数据的数据库系统。它将文本、图像等非结构化数据转换为高维数值向量后存储，并通过计算向量相似度（如余弦相似度）来实现语义搜索，是 RAG 和 AI 应用的核心基础设施。

6. invoke / stream / batch：三种调用方式

LCEL 链支持三种调用方式，适用场景不同。

invoke：同步调用，等模型返回全部结果再输出。适合后台批处理任务，不需要实时反馈的场景。

python

result = chain.invoke({"concept": "向量数据库"})
print(result)

stream：流式输出，模型生成一个 token 就返回一个，用户不需要等全部生成完才看到内容。做 ChatBot 时几乎必用，不然用户盯着空白框等 10 秒，体验很差。

python

for chunk in chain.stream({"concept": "向量数据库"}):
    print(chunk, end="", flush=True)
print()  # 换行

batch：批量调用，一次传入多个输入，并发执行，返回结果列表。处理大量数据时用，比循环调用 invoke 快得多。

python

results = chain.batch([
    {"concept": "向量数据库"},
    {"concept": "RAG"},
    {"concept": "Embedding"},
])
for r in results:
    print(r)
    print("---")

三种方式的选择原则：

场景	推荐方式	原因
ChatBot 对话	stream	实时显示，体验好
API 后端（单次请求）	invoke	简单直接
批量处理数据	batch	并发执行，速度快
后台任务	invoke / batch	不需要实时反馈

7. RunnableParallel：并行执行多个链

有时候需要对同一个输入做多件事，比如同时生成中文摘要和英文摘要，不需要等一个做完再做另一个。这时用 RunnableParallel：

python

from langchain_core.runnables import RunnableParallel

# 定义两个独立的链
zh_prompt = ChatPromptTemplate.from_messages([
    ("human", "用中文总结以下内容（50字以内）：{text}")
])
en_prompt = ChatPromptTemplate.from_messages([
    ("human", "Summarize the following in English (within 50 words): {text}")
])

zh_chain = zh_prompt | llm | output_parser
en_chain = en_prompt | llm | output_parser

# 并行执行
parallel_chain = RunnableParallel(
    chinese=zh_chain,
    english=en_chain,
)

result = parallel_chain.invoke({
    "text": "LangChain 是一个用于构建 LLM 应用的开源框架，提供了链式调用、记忆管理、工具集成等功能。"
})

print("中文摘要：", result["chinese"])
print("英文摘要：", result["english"])

两个 API 调用会并发执行，总耗时约等于耗时最长的那个，而不是两个相加。对于需要从多个角度处理同一输入的场景，这是重要的性能优化。

8. RunnablePassthrough 和 RunnableLambda

除了 RunnableParallel，还有两个常用的 Runnable 工具类：

RunnablePassthrough：直接把输入传递给下一个组件，不做任何处理。常用于把原始输入和处理结果一起传给下游：

python

from langchain_core.runnables import RunnablePassthrough

# 构建 RAG 链：把问题和检索结果同时传给 LLM
rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | output_parser
)

RunnableLambda：把任意 Python 函数包装成 Runnable，使其可以嵌入链中：

python

from langchain_core.runnables import RunnableLambda

# 把一个普通函数变成可以加入链的 Runnable
def add_prefix(text: str) -> str:
    return f"[处理结果] {text}"

chain = prompt | llm | output_parser | RunnableLambda(add_prefix)
result = chain.invoke({"concept": "向量数据库"})
# 结果会带上 "[处理结果]" 前缀

9. LCEL 链的调试：如何看清每步发生了什么

LCEL 的一个常见困惑是：出了问题不知道哪一步出的。解决方法有三个：

方法一：在链中间插入 print

python

from langchain_core.runnables import RunnableLambda

def debug_print(x):
    print(f"[DEBUG] 当前输出类型：{type(x)}, 内容：{str(x)[:100]}")
    return x

# 在 prompt 和 llm 之间插入 debug
chain = prompt | RunnableLambda(debug_print) | llm | output_parser

方法二：分步调用

python

# 分步调用，每步单独检查
prompt_output = prompt.invoke({"concept": "向量数据库"})
print("Prompt 输出：", prompt_output)

llm_output = llm.invoke(prompt_output)
print("LLM 输出：", llm_output)

final_output = output_parser.invoke(llm_output)
print("最终输出：", final_output)

方法三：接入 LangSmith

python

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-key"

# 之后的所有 chain.invoke() 调用都会自动追踪
result = chain.invoke({"concept": "向量数据库"})
# 在 LangSmith 平台可以看到完整的调用链路

LangSmith 是调试 LangChain 应用最强大的工具，第 11 篇会专门介绍。

10. 为什么用 LCEL 而不是直接调 API

直接调 OpenAI API 不是不行，但用 LCEL 有几个实际好处：

可组合性：每个组件都实现了同一套接口（Runnable），可以随意拼接。今天用 DeepSeek，明天换 Claude，只改一行模型初始化代码，其他不动。

内置 streaming 支持：不需要手写 SSE 解析，链里任意一个组件支持 streaming，整个链就支持。

可复用：定义好一个链，可以在多个地方 .invoke()，也可以作为子链嵌入更大的链里。

可观测性：LangChain 内置了回调系统，接入 LangSmith 后，每次链的调用，输入、输出、耗时、Token 消耗，全部可追踪。

批处理能力：LCEL 链原生支持 .batch() 并发调用，不需要自己写 asyncio。

直接调 API，这些都需要自己实现。

11. 小结

LCEL 的 | 符号背后是一套统一的 Runnable 接口——模型、Prompt、解析器、检索器都实现了它，所以可以任意串联。

记住这几个关键点：

| 把组件连成链，左边的输出是右边的输入
.invoke() 同步调用，.stream() 流式调用，.batch() 批量调用
RunnableParallel 并行执行，RunnableLambda 包装自定义函数
出了问题先分步调用，或者接入 LangSmith 看完整链路

LangChain 不是一个大而全的黑盒，而是一堆可组合的积木，拼法由你决定。后面讲 Memory、Retriever、RAG，都建立在 LCEL 这个基础上。