LCEL is the modern way to write LangChain code. It is a syntax system that lets you connect LangChain components together using the | pipe operator — clean, readable, and powerful.
What does LCEL include?
Full Summary / 完整总结
| Method / 方法 | Purpose | Importance | |
|---|---|---|---|
| Syntax | | | Connect components / 串联组件 | ⭐⭐⭐ |
| Invoke | .invoke() | Single call / 单次调用 | ⭐⭐⭐ |
| Invoke | .stream() | Token by token / 逐字输出 | ⭐⭐⭐ |
| Invoke | .batch() | Multiple inputs / 批量处理 | ⭐⭐ |
| Invoke | .ainvoke() | Async single / 异步单次 | ⭐⭐ |
| Invoke | .astream() | Async stream / 异步流式 | ⭐⭐ |
| Tools | RunnableParallel | Parallel / 并行 | ⭐⭐⭐ |
| Tools | RunnablePassthrough | Pass through / 透传 | ⭐⭐⭐ |
| Tools | RunnableLambda | Wrap function / 包装函数 | ⭐⭐⭐ |
| Tools | RunnableBranch | Conditional / 条件分支 | ⭐⭐ |
| Advanced | .bind() | Pre-set params / 预设参数 | ⭐⭐ |
| Advanced | .with_retry() | Auto retry / 自动重试 | ⭐⭐ |
| Advanced | .with_fallbacks() | Backup model / 备用模型 | ⭐⭐ |
| Advanced | .with_config() | Logging / 日志追踪 | ⭐ |
Part 1 — Syntax / 语法：| 管道符
EN: The | operator connects components in sequence. Output of left becomes input of right.
CN: | 运算符把组件顺序串联。左边的输出变成右边的输入。
python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
parser = StrOutputParser()
prompt = ChatPromptTemplate.from_template(
"Give me a company name for a brand that sells {product}. "
"Only return the name."
)
# | connects them in order / | 按顺序连接
chain = prompt | llm | parser
result = chain.invoke({"product": "eco-friendly water bottles"})
print(result)
# → "AquaGreen Co."Flow：
{"product": "eco-friendly water bottles"}
↓ prompt
"Give me a company name for eco-friendly water bottles..."
↓ llm
AIMessage(content="AquaGreen Co.")
↓ parser
"AquaGreen Co."Part 2 — Invocation Methods / 调用方式
.invoke() — Single call / 单次调用
EN: Call the chain once, wait for the full response.
CN: 调用一次，等待完整回复。
python
result = chain.invoke({"product": "water bottle"})
print(result)
# → "AquaGreen Co." (完整结果一次返回).stream() — Token by token / 逐字输出
EN: Returns response token by token as it generates — like the ChatGPT typing effect.
CN: 边生成边逐字返回 — 就像 ChatGPT 的打字效果。
python
for chunk in chain.stream({"product": "water bottle"}):
print(chunk, end="", flush=True)
# Console / 控制台：
# Aqua...Green...Co... (逐字出现 / appears word by word).batch() — Multiple inputs at once / 批量处理
EN: Run the same chain on multiple inputs simultaneously.
CN: 同时对多个输入运行同一个 chain。
python
results = chain.batch([
{"product": "water bottle"},
{"product": "coffee mug"},
{"product": "lunch box"},
])
print(results)
# → ["AquaGreen Co.", "BrewMaster Inc.", "FreshBox Ltd."]When to use / 什么时候用：
Process large datasets → .batch()
批量处理大量数据 → .batch()
Generate content in bulk → .batch()
批量生成内容 → .batch().ainvoke() / .astream() — Async / 异步
EN: Async versions of .invoke() and .stream(). Use when building web APIs or handling multiple users at the same time.
CN: .invoke() 和 .stream() 的异步版本。构建 Web API 或同时处理多个用户时使用。
python
import asyncio
# Async single call / 异步单次调用
async def generate():
result = await chain.ainvoke({"product": "water bottle"})
print(result)
# Async streaming / 异步流式输出
async def stream():
async for chunk in chain.astream({"product": "water bottle"}):
print(chunk, end="", flush=True)
asyncio.run(generate())When to use / 什么时候用：
FastAPI / web service → .ainvoke()
FastAPI / Web 服务 → .ainvoke()
Handle multiple users → .ainvoke()
同时处理多用户 → .ainvoke()
Real-time streaming UI → .astream()
前端实时打字效果 → .astream()Part 3 — Composition Tools / 组合工具
EN: These are the building blocks you use inside LCEL pipelines.
CN: 这些是你在 LCEL pipeline 里使用的构建积木。
python
from langchain_core.runnables import (
RunnableParallel, # Run multiple chains at once / 并行运行
RunnablePassthrough, # Pass input unchanged / 透传原始输入
RunnableLambda, # Wrap any Python function / 包装Python函数
RunnableBranch, # Conditional routing / 条件分支
)Part 4 — Advanced Methods / 高级方法
.bind() — Pre-set parameters / 预设参数
EN: Lock in fixed parameters on a component so you don’t repeat them every call.
CN: 预先固定组件的参数，不用每次调用都传。
# Without bind / 不用 bind — 每次都要传
llm.invoke(prompt, temperature=0, max_tokens=100)
# With bind / 用 bind — 预先固定
llm_precise = llm.bind(temperature=0, max_tokens=100)
chain = prompt | llm_precise | parser
result = chain.invoke({"product": "water bottle"}).with_retry() — Auto retry / 自动重试
EN: Automatically retry failed LLM calls — network errors, rate limits, timeouts.
CN: LLM 调用失败时自动重试 — 网络错误、限流、超时。
reliable_chain = prompt | llm.with_retry(
stop_after_attempt=3, # Max 3 retries / 最多重试3次
wait_exponential_jitter=True # Exponential backoff / 指数退避
) | parser
result = reliable_chain.invoke({"product": "water bottle"})
# → Fails? Auto retry up to 3 times / 失败？自动重试最多3次.with_fallbacks() — Backup model / 备用模型
EN: If the primary model fails, automatically switch to a backup.
CN: 主模型失败时，自动切换到备用模型。
primary = ChatOpenAI(model="gpt-4o-mini")
fallback = ChatOpenAI(model="gpt-3.5-turbo")
# Primary fails → auto switch to fallback
# 主模型失败 → 自动切换备用
reliable_llm = primary.with_fallbacks([fallback])
chain = prompt | reliable_llm | parser
result = chain.invoke({"product": "water bottle"}).with_config() — Runtime config / 运行时配置
EN: Attach tags, metadata, and run names for logging and tracing in LangSmith.
CN: 附加标签、元数据、运行名称，用于 LangSmith 日志和追踪。
result = chain.with_config(
tags = ["production", "v2"],
metadata = {"user_id": "abc123"},
run_name = "product-name-generator"
).invoke({"product": "water bottle"})Part 5 — Async Support / 异步支持
EN: Every LCEL chain automatically supports async — no extra setup needed.
CN: 每个 LCEL chain 自动支持异步 — 不需要额外设置。
# Sync / 同步
result = chain.invoke(...)
for chunk in chain.stream(...): ...
results = chain.batch([...])
# Async / 异步 — just add 'a' prefix / 只需加 'a' 前缀
result = await chain.ainvoke(...)
async for chunk in chain.astream(...): ...
results = await chain.abatch([...])