介绍

LCEL（LangChain Expression Language） 是 LangChain 官方推出的声明式链式编排语法，核心用 | 管道符串联所有“可运行单元（Runnable）”，让流程像流水线一样清晰、可组合、可流式、可并行。

一句话：用写命令管道的方式，写 LLM 应用工作流。

设计理念

一切皆 Runnable

• 提示词模板、模型、解析器、工具、自定义函数——只要实现 Runnable 接口，都能进管道。
• 统一接口：invoke/ainvoke（同步/异步）、batch/abatch（批量）、stream/astream（流式）。

声明式而非命令式

你只描述“数据怎么走”，LangChain 负责优化执行（并行、缓存、批处理）。

原生支持流式与异步

链级天然支持逐字输出（stream），延迟接近直接调用模型。

基础语法

1）管道符 |（RunnableSequence）

顺序执行，前一个输出作为后一个输入：

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-3.5-turbo")
parser = StrOutputParser()

# LCEL 链式
chain = prompt | model | parser

# 调用
result = chain.invoke({"topic": "LCEL"})
print(result)

等价于：

result = parser.invoke(model.invoke(prompt.invoke({"topic": "LCEL"})))

2）字典 → RunnableParallel（并行）

字典会自动转为并行执行，多分支同时跑，结果合并：

from langchain_core.runnables import RunnableParallel

chain = RunnableParallel({
    "summary": prompt | model | parser,
    "joke": prompt | model | StrOutputParser()
})

out = chain.invoke({"topic": "AI"})
# out = {"summary": "...", "joke": "..."}

LCEL 简写（自动识别字典）：

chain = {
    "summary": prompt | model | parser,
    "joke": prompt | model | parser
} | RunnablePassthrough()

3）常用 Runnable 工具

• RunnableLambda：把任意 Python 函数包装成 Runnable

  from langchain_core.runnables import RunnableLambda
  upper = RunnableLambda(lambda x: x.upper())
  chain = prompt | model | parser | upper
  

• RunnablePassthrough：透传输入，用于保留原始数据或分支合并

  chain = {"question": RunnablePassthrough()} | prompt | model
  

• RunnableAssign：向字典追加字段

  chain = RunnableAssign({"answer": model})
  

• RunnableBranch：条件分支（if-else）

  from langchain_core.runnables import RunnableBranch
  branch = RunnableBranch(
  (lambda x: len(x["question"]) > 10, long_chain),
  short_chain
  )
  

执行模式（统一接口）

1）同步调用

result = chain.invoke({"topic": "LangChain"})

2）异步调用

import asyncio
async def main():
    result = await chain.ainvoke({"topic": "LangChain"})
asyncio.run(main())

3）批量处理

results = chain.batch([{"topic": "AI"}, {"topic": "Python"}])

4）流式输出（核心优势）

for chunk in chain.stream({"topic": "LCEL"}):
    print(chunk, end="", flush=True)

典型场景示例

场景1：检索增强（RAG）

# retrieval_chain = 检索器 | 格式化文档
# rag_chain = {"context": retrieval_chain, "question": RunnablePassthrough()} | prompt | model | parser

场景2：多工具 Agent

from langchain.agents import create_tool_calling_agent, AgentExecutor
tools = [tool1, tool2]
agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)
# LCEL 可直接嵌入：chain = prompt | agent_executor

场景3：多分支并行

chain = {
    "code": code_prompt | model | code_parser,
    "explanation": explain_prompt | model | StrOutputParser()
} | RunnablePassthrough()

LCEL 优势总结

1. 简洁易读：用 | 替代多层嵌套，逻辑一目了然。
2. 性能更好：自动并行、流式、批处理优化，减少延迟。
3. 可组合强：任意 Runnable 自由拼接，支持复杂分支与循环。
4. 调试友好：每一步可单独测试，RunnablePassthrough 方便查看中间结果。
5. 官方主推：LangChain 0.1+ 核心演进方向，旧 Chain 逐步弃用。

与旧 Chain 对比

使用注意

1. 字典不是 Runnable：{"a": ...} 不能直接调用 .invoke，必须放在管道中或显式 RunnableParallel。
2. 输入输出类型匹配：前一个输出必须和后一个输入兼容，否则报错。
3. 流式时异常处理：stream 中出错会中断，建议外层加 try/except。

完整可运行代码

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

# 1. 组件定义
prompt = ChatPromptTemplate.from_template("用3句话介绍一下 {topic}")
model = ChatOpenAI(model="gpt-3.5-turbo", api_key="YOUR_KEY")
parser = StrOutputParser()

# 2. LCEL 链
chain = prompt | model | parser

# 3. 调用
if __name__ == "__main__":
    # 同步
    print("=== 同步 ===")
    print(chain.invoke({"topic": "LCEL"}))

    # 流式
    print("\n=== 流式 ===")
    for chunk in chain.stream({"topic": "LCEL"}):
        print(chunk, end="", flush=True)