LangChain LCEL 表达式语言介绍（链式调用）-马育民老师

# 介绍

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

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

---

# 设计理念

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

### **声明式而非命令式**

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

### **原生支持流式与异步**

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

# 基础语法

### 1）管道符 `|`（RunnableSequence）

顺序执行，前一个输出作为后一个输入：
```python
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)
```
等价于：
```python
result = parser.invoke(model.invoke(prompt.invoke({"topic": "LCEL"})))
```

### 2）字典 → RunnableParallel（并行）
字典会自动转为**并行执行**，多分支同时跑，结果合并：
```python
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 简写（自动识别字典）：
```python
chain = {
    "summary": prompt | model | parser,
    "joke": prompt | model | parser
} | RunnablePassthrough()
```

### 3）常用 Runnable 工具

- **RunnableLambda**：把任意 Python 函数包装成 Runnable
```python
from langchain_core.runnables import RunnableLambda
upper = RunnableLambda(lambda x: x.upper())
chain = prompt | model | parser | upper
```
- **RunnablePassthrough**：透传输入，用于保留原始数据或分支合并
```python
chain = {"question": RunnablePassthrough()} | prompt | model
```
- **RunnableAssign**：向字典追加字段
```python
chain = RunnableAssign({"answer": model})
```
- **RunnableBranch**：条件分支（if-else）
```python
from langchain_core.runnables import RunnableBranch
branch = RunnableBranch(
(lambda x: len(x["question"]) > 10, long_chain),
short_chain
)
```

---

# 执行模式（统一接口）
### 1）同步调用
```python
result = chain.invoke({"topic": "LangChain"})
```

### 2）异步调用
```python
import asyncio
async def main():
    result = await chain.ainvoke({"topic": "LangChain"})
asyncio.run(main())
```

### 3）批量处理
```python
results = chain.batch([{"topic": "AI"}, {"topic": "Python"}])
```

### 4）流式输出（核心优势）
```python
for chunk in chain.stream({"topic": "LCEL"}):
    print(chunk, end="", flush=True)
```

---

# 典型场景示例
### 场景1：检索增强（RAG）
```python
# retrieval_chain = 检索器 | 格式化文档
# rag_chain = {"context": retrieval_chain, "question": RunnablePassthrough()} | prompt | model | parser
```

### 场景2：多工具 Agent
```python
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：多分支并行
```python
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 对比
| 特性 | 旧 Chain（SequentialChain） | LCEL |
|---|---|---|
| 语法 | 嵌套、样板代码多 | `|` 管道，极简 |
| 流式 | 支持差 | 原生逐字输出 |
| 并行 | 需手动写 | 字典自动并行 |
| 异步 | 支持有限 | 原生 `async/await` |
| 可扩展性 | 弱 | 强，任意 Runnable 组合 |

---

# 使用注意

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

---

# 完整可运行代码
```python
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)
```

原文出处：/show_1GW3N4nfF8Jx.html