概述

LangChain Expression Language (LCEL) 是一种声明式的链式调用语法，通过管道操作符 | 将多个 Runnable 组件串联起来，实现复杂的工作流。LCEL 是 LangChain v0.1+ 推荐的编写链式调用的方式。

基础语法

管道操作符 |

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4")
parser = StrOutputParser()

# 使用管道操作符串联
chain = model | parser

# 执行
result = chain.invoke("什么是 LCEL？")

Runnable 接口

所有支持 LCEL 的组件都实现了 Runnable 接口：

# Runnable 的核心方法
chain.invoke(input)           # 同步调用
chain.batch([input1, input2]) # 批量调用
chain.stream("prompt")       # 流式调用

常用组件

Prompt 模板

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个{language}助手"),
    ("human", "{question}")
])

# 串联
chain = prompt | model | parser

result = chain.invoke({
    "language": "Python",
    "question": "解释生成器是什么"
})

输出解析器

from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel

class Response(BaseModel):
    answer: str
    confidence: float

parser = JsonOutputParser(pydantic_object=Response)

chain = prompt | model | parser
result = chain.invoke({...})
# result 是 dict: {"answer": "...", "confidence": 0.9}

组合模式

并行处理

from langchain_core.runnables import RunnableParallel

chain1 = prompt | model | parser
chain2 = summary_prompt | model | summary_parser

combined = RunnableParallel(
    detail=chain1,
    summary=chain2
)

result = combined.invoke({"topic": "LangChain"})
# result = {"detail": "...", "summary": "..."}

条件路由

from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: "simple" in x["query"], simple_chain),
    (lambda x: "complex" in x["query"], complex_chain),
    default_chain
)

完整示例：RAG 链

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_chroma import Chroma

# 向量存储
vectorstore = Chroma("docs", OpenAIEmbeddings(), "...")
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# RAG 链
template = """基于以下上下文回答问题：

上下文：
{context}

问题：{question}"""

prompt = ChatPromptTemplate.from_template(template)
model = ChatOpenAI(model="gpt-4")
parser = StrOutputParser()

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | model
    | parser
)

result = rag_chain.invoke("LangChain 的核心概念是什么？")

常见问题

Q1: 为什么用 | 而不是 .pipe()？

• | 更简洁，符合 Unix 管道直觉
• 语义清晰：数据从左向右流动

Q2: LCEL 支持异步吗？

• 原生支持，.ainvoke() / .abatch() 自动使用异步

Q3: 如何调试 LCEL 链？

• 使用 .with_config({"run_name": "StepName"}) 添加名称
• 在中间插入检查点查看输出

参考资料

• LangChain LCEL 文档
• LangChain GitHub