LangChain Expression Language (LCEL) 深度解析
从链式调用到流式输出，全面掌握 LangChain 的声明式编程范式，构建高性能 LLM 应用。
一、LCEL 是什么？
LangChain Expression Language（LCEL）是 LangChain 推出的声明式语言，用于轻松组合各种组件构建 LLM 应用。它借鉴了 Unix 管道的思想，让组件之间通过|操作符串联。
1.1 传统方式 vs LCEL
传统方式：手动管理数据流转
# 传统方式：繁琐且容易出错prompt=ChatPromptTemplate.from_template("告诉我关于{topic}的笑话")model=ChatOpenAI()output_parser=StrOutputParser()# 手动调用每一步chain_input={"topic":"程序员"}prompt_value=prompt.invoke(chain_input)message=model.invoke(prompt_value)result=output_parser.invoke(message)print(result)LCEL 方式：声明式组合
# LCEL 方式：简洁优雅chain=prompt|model|output_parser result=chain.invoke({"topic":"程序员"})print(result)1.2 LCEL 的核心优势
| 特性 | 说明 |
|---|---|
| 流式输出 | 自动支持，无需额外代码 |
| 异步支持 | 同步/异步 API 自动生成 |
| 并行执行 | 自动优化并行调用 |
| 重试机制 | 内置错误重试 |
| 可观测性 | 无缝集成 LangSmith |
| 类型安全 | 完整的类型提示 |
二、核心组件详解
2.1 Runnable 接口
所有 LCEL 组件都实现了Runnable接口，这是 LCEL 的基石：
fromlangchain_core.runnablesimportRunnableclassRunnable:# 同步调用definvoke(self,input,config=None)->Output:...# 异步调用asyncdefainvoke(self,input,config=None)->Output:...# 批量调用defbatch(self,inputs,config=None)->List[Output]:...# 流式输出defstream(self,input,config=None)->Iterator[Output]:...# 链式组合def__or__(self,other)->RunnableSequence:...2.2 RunnableSequence：链式调用
使用|操作符创建链：
fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParser# 创建组件prompt=ChatPromptTemplate.from_template("你是一个{role}，请用专业角度分析：{question}")model=ChatOpenAI(model="gpt-4-turbo")parser=StrOutputParser()# 组合成链chain=prompt|model|parser# 调用result=chain.invoke({"role":"资深架构师","question":"微服务架构的优缺点"})执行流程：
输入 {"role": "...", "question": "..."} ↓ PromptTemplate → 生成 PromptValue ↓ ChatOpenAI → 生成 AIMessage ↓ StrOutputParser → 输出字符串2.3 RunnableParallel：并行执行
当多个分支可以并行时，使用RunnableParallel：
fromlangchain_core.runnablesimportRunnableParallel# 定义多个并行分支chain=RunnableParallel(# 分支1：生成摘要summary=prompt_summary|model|parser,
