今天咱们继续 LangChain 的学习，在之前的内容Langchain学习教程二（提示词模板和 LCEL 语法）中，我们已经初识了 LCEL 语法，那么今天我们更深入的来了解一下，关于 LCEL 的内容还是比较多，我们分为上下两篇文章，今天是上篇，主要是讲解下 LCEL 是怎么运行的以及 Runnable 是啥。好了，废话不多说，我们立马开始。
什么是LCEL
我们还是复用之前的例子
from langchain.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
template = "请将以下内容翻译成{language}：{input}"
prompt = PromptTemplate.from_template(template)
llm = ChatOpenAI(
model="Qwen/Qwen3-32B",
openai_api_key="your_key",
base_url="https://api.siliconflow.cn/v1",
temperature=0
)
parser = StrOutputParser()
# 用 LCEL 方式组合成一个完整链（Prompt → LLM → Parser）
chain = prompt | llm | parser
result = chain.invoke({
"language": "日文",
"input": "我是一个中国人"
})
print(result)
我们看到上面的核心就是 chain = prompt | llm | parser
从逻辑上来说，这里做了三件事
你看起来只是在“管道”中流转数据，其实内部经历了：
在介绍 LCEL 前，咱们可能写成这样：
formatted = prompt.format(...)
result = llm.invoke(formatted)
output = parser.invoke(result)
但是通过这种语法改写，你只需要：
chain = prompt | llm | parser
是不是像在拼积木？这就是 LCEL 的威力。
所以现在我们来回答一下什么是 LCEL。
LCEL，全称 LangChain 表达式语言 (LangChain Expression Language)，是 LangChain 框架中用于组合链 (Chains) 的一种声明式方式。它提供了一种统一且高度灵活的接口，用于构建从简单到复杂的 LLM 应用程序。
核心思想是通过 | (管道) 运算符将不同的 Runnable 组件连接起来，形成一个数据流动的链条
它能解决什么问题？
在没有 LCEL 之前，构建复杂的 LLM 应用链条可能会面临以下挑战：
复杂性管理： 当链条包含多个步骤和条件逻辑时，代码会变得冗长且难以维护。
输入/输出管理： 在不同组件之间传递数据需要手动处理，容易出错。
可观测性： 难以追踪数据在链条中的流动，调试和优化变得困难。
性能： 缺乏对异步、批处理和流式传输的内置支持，影响应用性能和用户体验。
统一接口： 不同类型的组件（LLM、检索器、工具等）可能需要不同的调用方式。
LCEL 通过引入 Runnable 接口和简洁的组合语法，优雅地解决了这些问题，使得构建、调试和部署 LLM 应用程序变得更加高效和直观。
chain = prompt | llm | parser
所以上面的代码核心流程就是：
- 原始输出再经过解析器，最终得到我们需要的结构化数据。
每个步骤都是一个独立的、可复用的 Runnable。
什么是 Runnable？
上面我们提到过好几次这个东西，那么这又是个啥玩意。
官方文档：https://python.langchain.com/docs/concepts/runnables/
在 LCEL 中，Runnable 是核心的抽象概念。简单来说，任何可以被执行并接收输入、产生输出的组件，都是一个 Runnable。
这意味着无论是：
一个 BaseLanguageModel (LLM 实例)
一个 BaseOutputParser (输出解析器)
一个 VectorStoreRetriever (向量存储检索器)
甚至一个普通的 Python 函数 (RunnableLambda)
它们都实现了 Runnable 接口。这种统一的接口是 LCEL 强大组合能力的基础。它确保了无论组件的内部逻辑如何，它们都可以通过相同的方式被调用和连接。
大白话解释
“Runnable”这个词很抽象，光看字面意思可能觉得“能跑的东西”很玄。
你可以把 Runnable 理解成： 一个可以“收货”和“出货”的工人
在 LangChain 这个“流水线系统”里，每个模块都是个“工人”。它的职责是：收到输入 → 加工处理 → 输出结果。
这个“工人”只要满足三个条件：
只要满足这三个条件，它就是一个 Runnable。
Runnable 可以“串起来”，原因是：因为每个工人都有固定的“输入口”和“输出口”：
所以它们就像「一段段水管」一样能接起来：
chain = prompt | llm | parser
Runnable 的执行能力
Runnable 接口定义了一系列标准的方法，用于执行和获取输出。这些方法提供了同步、异步、单次和批处理等多种执行模式，比如下面这些方法
invoke(input, config=None):
作用： 同步地执行 Runnable，接收单个输入，并返回单个输出。
场景： 最常见的调用方式，用于处理单个请求。
示例： chain.invoke({"question": "你好"})
batch(inputs, config=None):
作用： 同步地执行 Runnable，接收一个输入列表，并返回一个输出列表。它会并行或并发地处理这些输入，以提高效率。
场景： 当你需要处理多个独立的请求，并且希望一次性提交以减少开销时。
示例： chain.batch([{"q": "A"}, {"q": "B"}])
stream(input, config=None):
作用： 同步地执行 Runnable，并以流的形式返回输出的块（chunks）。对于 LLM，这意味着你可以逐字逐句地接收文本。
场景： 当你需要实时显示 LLM 的生成过程，或者处理大型数据集时，可以逐步处理数据。
示例： for chunk in chain.stream({"query": "讲个笑话"}): print(chunk)
ainvoke(input, config=None):
作用： 异步地执行 Runnable，接收单个输入，并返回单个输出。
场景： 在异步应用程序（如 FastAPI、Aiohttp）中使用，以避免阻塞主线程。
示例： await chain.ainvoke({"question": "你好"})
abatch(inputs, config=None):
作用： 异步地执行 Runnable，接收一个输入列表，并返回一个输出列表。
场景： 异步地处理多个请求，提高并发性能。
示例： await chain.abatch([{"q": "A"}, {"q": "B"}])
astream(input, config=None):
作用： 异步地执行 Runnable，并以异步流的形式返回输出的块。
场景： 异步地实时显示 LLM 的生成过程。
示例： async for chunk in chain.astream({"query": "讲个笑话"}): print(chunk)
这些方法提供了强大的灵活性，我们能够根据应用程序的需求选择最合适的执行模式。
实现一个 Runnable 函数
接下来我们实现一个最简单的 Runnable 函数，并把它添加到我们的链里面
比如我们上面的翻译工具，我们想在输入内容里面补充一段自定义的内容。
我们先定义一个函数
def append_signature(text: str) -> str:
return text + "\n\n—— 由 AI 翻译，hockor 提供能力"
这个函数的输入是字符串，输出也是字符串，非常适合接在 StrOutputParser() 后面。
然后我们使用官方提供的RunnableLambda将我们的函数作为参数传入
from langchain_core.runnables import RunnableLambda
def append_signature(text: str) -> str:
return text + "\n\n—— 由 AI 翻译，hockor 提供能力"
test_node = RunnableLambda(append_signature)
现在 test_node 就是一个合法的 Runnable，你可以把它当作“工人”拼接到链里了。
也就是
chain = prompt | llm | parser | test_node
这样就实现了在 LLM 输出后再加一层自定义处理逻辑。
所以完整代码就是
from langchain.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI
template = "请将以下内容翻译成{language}：{input}"
prompt = PromptTemplate.from_template(template)
llm = ChatOpenAI(
model="Qwen/Qwen3-32B",
openai_api_key="your_key",
base_url="https://api.siliconflow.cn/v1",
temperature=0
)
parser = StrOutputParser()
# 自定义函数：在结果后追加一行签名
def append_signature(text: str) -> str:
return text + "\n\n—— 由 AI 翻译，hockor 提供能力"
# 把它包装成 Runnable
signature_node = RunnableLambda(append_signature)
chain = prompt | llm | parser | signature_node
result = chain.invoke({
"language": "日文",
"input": "我是一个中国人"
})
print(result)
打印结果就是
私は中国人です。
—— 由 AI 翻译，hockor 提供能力
RunnableLambda 接受一个普通函数，它就像一个“中间处理器”，你可以用它来做：
判断一个对象是不是 Runnable
from langchain_core.runnables.base import Runnable
isinstance(my_component, Runnable) # True 就说明它能串进链
串行和并行
LCEL 不仅支持简单的顺序执行，还提供了强大的并行组合能力，让你可以构建更复杂的逻辑流。
顺序组合 (|)
顺序组合是最常见和直观的组合方式，使用 | (管道) 运算符连接 Runnable。数据从左到右依次传递。
原理就是前一个 Runnable 的输出作为后一个 Runnable 的输入。
特点： 简单、直观，适用于线性的处理流程。
并行组合 (RunnableParallel)
并行组合允许你同时执行多个 Runnable，并将它们的输出合并。这对于需要同时从多个源获取信息或同时生成不同类型输出的场景非常有用。
原理： RunnableParallel 接收一个字典，其键是输出的名称，值是对应的 Runnable。它会同时执行这些 Runnable，并将它们的输出收集到一个字典中。
非常适用于需要合并多个独立结果的场景。
示例： 假设我们想同时生成一个关于“猫”的笑话和一个关于“狗”的笑话。
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel
from langchain_openai import ChatOpenAI
joke_prompt_cat = ChatPromptTemplate.from_template("请讲一个关于猫的笑话")
joke_prompt_dog = ChatPromptTemplate.from_template("请讲一个关于狗的笑话")
llm = ChatOpenAI(
model="Qwen/Qwen3-32B",
openai_api_key="your_key",
base_url="https://api.siliconflow.cn/v1",
temperature=0
)
parser = StrOutputParser()
# 定义并行链
parallel_jokes = RunnableParallel(
cat_joke=joke_prompt_cat | llm | parser,
dog_joke=joke_prompt_dog | llm | parser
)
执行并行链
result = parallel_jokes.invoke({})
print(result)
在这个例子中：
RunnableParallel 接收一个字典，键 cat_joke 对应一个生成猫笑话的链，键 dog_joke 对应一个生成狗笑话的链。
当 parallel_jokes 被调用时，这两个内部链会同时开始执行。
parallel_jokes 会等待两个链都完成后，将它们的输出收集到一个新的字典中，键为 cat_joke 和 dog_joke，值为各自链的最终输出。
{
'cat_joke': '\n\n**笑话：** \n主人问猫：“你为什么总爱钻纸箱？” \n猫头也不抬：“因为它们是宇宙中唯一免费的五星级SPA房——躺进去之前，我得先确认你有没有把零食藏在里面。” \n\n**解析：** \n这个笑话结合了猫的日常行为（钻纸箱）和人类的幽默视角，用“五星级SPA房”夸张纸箱的“豪华”，再反转揭示猫的“自私动机”——检查零食，既符合猫的狡黠性格，又制造了反差萌。',
'dog_joke': '\n\n好的，这里有一个轻松幽默的狗笑话：\n\n**《超市奇遇记》**\n\n主人对狗说：“今天表现好，晚饭后带你去超市！” \n狗一听，耳朵立刻竖起来，尾巴摇得像螺旋桨：“汪汪！（好呀！）有肉骨头卖吗？” \n\n主人笑着摸摸它的头：“你想得美，我是去买东西。” \n狗歪着头想了想，突然跳上餐桌，把肉骨头全扫进碗里，叼着满嘴骨头冲到门口：“汪汪汪！（不，这次你得带我去——我得把骨头藏好！）” \n\n主人：……“你是不是搞错了什么？” \n狗（严肃脸）：“你出门不带狗，超市的骨头会笑死的。” \n\n—— \n希望这个笑话能让你笑一笑！如果需要更多，我可以继续编～ 🐶'
}
组合使用：
LCEL 的强大之处在于你可以将顺序组合和并行组合灵活地结合起来。例如，你可能需要并行地从多个数据库检索信息，然后将所有检索到的信息合并，再传递给一个 LLM 进行总结。
from langchain_core.runnables import RunnableParallel, RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
# 模拟两个不同的检索器
def retrieve_docs_from_db1(query):
return f"DB1关于'{query}'的文档：[文档A, 文档B]"
def retrieve_docs_from_db2(query):
return f"DB2关于'{query}'的文档：[文档C, 文档D]"
# 定义检索 Runnable
retriever1 = RunnableLambda(retrieve_docs_from_db1)
retriever2 = RunnableLambda(retrieve_docs_from_db2)
# 并行检索
parallel_retrieval = RunnableParallel(
docs_from_db1=retriever1,
docs_from_db2=retriever2
)
# 组合检索结果的 Runnable
combine_docs = RunnableLambda(lambda x: f"所有相关文档：{x['docs_from_db1']} {x['docs_from_db2']}")
# 总结 Prompt
summary_prompt = ChatPromptTemplate.from_template("根据以下文档进行总结：\n{docs}\n总结：")
llm = ChatOpenAI(temperature=0.5)
parser = StrOutputParser()
# 构建完整的链条
full_chain = (
{"query": RunnableLambda(lambda x: x)} # 确保输入是字典形式
| parallel_retrieval
| {"docs": combine_docs} # 将并行结果合并到 'docs' 键下
| summary_prompt
| llm
| parser
)
# 执行链条
# result = full_chain.invoke("LangChain")
# print(result)
在这个复杂的例子中：
首先，并行地调用两个模拟的检索器 retriever1 和 retriever2，它们都接收相同的 query 输入。
parallel_retrieval 的输出是一个字典，包含 docs_from_db1 和 docs_from_db2 两个键。
combine_docs 这个 RunnableLambda 接收这个字典，并将两个文档字符串合并成一个。
合并后的文档字符串作为 summary_prompt 的 {docs} 变量的输入。
最后，LLM 生成总结，并通过解析器得到最终的字符串结果。
通过 LCEL 的顺序和并行组合能力，开发者可以清晰、高效地构建出高度模块化和可维护的 LLM 应用程序。
粤ICP备14082021号
免费POC，
零成本试错
