LangChain LCEL 语法入门与实战：用“管道”搭出大模型应用

版本：Python 3.9+、LangChain ≥ 0.1
目标：10 分钟掌握 LCEL（LangChain Expression Language）声明式语法，写出可维护、可扩展、可并发的链式调用。

---

1. 什么是 LCEL？

LCEL = LangChain Expression Language，官方 2023 起主推的链式表达式语法。
核心思想：所有组件都实现 Runnable 接口，用 |（管道符）把组件像拼乐高一样串起来：

chain = component1 | component2 | component3

• 无需继承基类、无需显式写 LLMChain
• 自动类型推导 + 原生支持 流式 / 异步 / 批处理
• 底层基于 Python __or__ 重载，等价于 RunnableSequence

---

2. 最简三件套：Prompt | LLM | Parser

文件：2_minimal_lcel.py

# 2_minimal_lcel.py
import os
from pathlib import Path
from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# ---------- 0. 环境 ----------
dotenv_path = Path(__file__).with_name(".env")   # 同目录下的 .env
load_dotenv(dotenv_path)

# ---------- 1. 三件套 ----------
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
    openai_api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL") or None,
)

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

# ---------- 2. 一条 LCEL 链 ----------
chain = prompt | llm | StrOutputParser()

if __name__ == "__main__":
    print(chain.invoke({"question": "什么是LCEL？"}))

终端打印：

LCEL（LangChain Expression Language）是 LangChain 提供的声明式链式语法……

---

3. 把链当函数：RunnableLambda 自定义逻辑

文件：3_runnable_lambda.py

from langchain_core.runnables import RunnableLambda

def add_length(original: str) -> str:
    return f"长度={len(original)}\n{original}"

chain = prompt | llm | StrOutputParser() | RunnableLambda(add_length)

print(chain.invoke({"question": "Hello"}))

输出：

长度=5
Hello

要点：

• 任意 Python 函数 → RunnableLambda(fn) 即可插进链
• 支持 同步 / 异步 统一接口

---

4. 并行分支：RunnableParallel 一次调多个模型

文件：4_parallel.py

from langchain_core.runnables import RunnableParallel

joke_chain = prompt | llm | StrOutputParser()
poem_chain = (
    ChatPromptTemplate.from_template("写一首关于{question}的诗") | llm | StrOutputParser()
)

map_chain = RunnableParallel({
    "joke": joke_chain,
    "poem": poem_chain
})

print(map_chain.invoke({"question": "春天"}))

输出：

{'joke': '春天来了，小草说：我终于可以合法地绿了！', 
 'poem': '春风十里，不如你……'}

---

5. 流式输出：逐字打印，内存零等待

文件：5_stream.py

chain = prompt | llm | StrOutputParser()

for chunk in chain.stream({"question": "用 20 字介绍 LangChain"}):
    print(chunk, end="", flush=True)

终端效果（逐字蹦）：

LangChain是编排大模型与外部数据的开发框架

---

6. 异步并发：asyncio 友好

文件：6_async.py

import asyncio

async def main():
    chain = prompt | llm | StrOutputParser()
    result = await chain.ainvoke({"question": "异步有什么好处？"})
    print(result)

asyncio.run(main())

---

7. 批处理：一次传 N 条，自动并行

文件：7_batch.py

questions = [{"question": f"第{i}条测试"} for i in range(1, 4)]
for ans in chain.batch(questions):
    print(ans)

---

8. 原理 1 张图记住

调用阶段	LCEL 做了什么
prompt | llm	把 Dict → List[BaseMessage] → AIMessage
llm | parser	把 AIMessage → str / dict / datetime …
整体	返回新的 Runnable，可继续 .pipe() 无限拼接

底层就是 __or__ 重载：

def __or__(self, other):
    return RunnableSequence(self, other)

---

9. 迁移旧代码：传统 chain → LCEL 对照

老写法（命令式）	LCEL 写法（声明式）
LLMChain(llm=llm, prompt=prompt).run(...)	(prompt | llm | StrOutputParser()).invoke(...)
多步骤手动传参	一条管道自动传参
流式需额外封装	原生 .stream() / .astream()

---

10. 小结 & 最佳实践

1. 所有组件默认实现 Runnable：Prompt、LLM、Parser、Retriever、Tool…
2. 用 | 连接，可读性 > 嵌套函数
3. 同一条链可同时支持 sync / async / stream / batch 四种调用方式，无需改代码
4. 复杂 DAG（并行、条件、循环）用 RunnableParallel / RunnableBranch 即可
5. 日志 & 监控：配合 RunnableBinding 或 LangSmith，一行 .with_config({"run_name": "xxx"}) 搞定

---

现在你已经会用 LCEL 搭链、并行、流式、异步、批处理 —— 一行代码即可上线。
点赞 + 收藏，下篇带你用 LCEL 手撕 RAG + Agent！