What is LCEL & Runnable?
- Runnable is a standard interface (protocol) that represents anything that can be “run” — it takes an input and produces an output. It is the basic building block of LangChain.
是一个标准接口（协议），代表任何“可运行”的东西——它接收输入并产生输出。它是 LangChain 的基础构建块 - LCEL (LangChain Expression Language) is a declarative way to compose chains using the pipe operator
|. It allows you to string together multiple Runnables into a processing pipeline
是一种声明式语言，通过管道操作符|来组合链。它允许你将多个 Runnable 串联成一个处理流水线
Core Content
Runnable Interface
Runnable is an abstract interface that defines five core methods.
| Method | Type | Purpose |
|---|---|---|
invoke(input) | 同步 | 单次调用，阻塞式返回完整结果 |
ainvoke(input) | 异步 | 单次异步调用，非阻塞 |
stream(input) | 同步 | 流式输出，逐 token 生成结果 |
batch([inputs]) | 同步 | 批量处理多个输入 |
abatch([inputs]) | 异步 | 异步批量处理 |
LCEL (LangChain Expression Language)
LCEL is the syntax/rules for composing Runnables. The core syntax is:
chain = component1 | component2 | component3
result = chain.invoke(input)Data flows left to right: output of component1 → input of component2 → output of component2 → input of component3
Key Runnable Primitives
Code Implementation
Basic LCEL Chain
# ============================================================
# 导入必要的模块
# Import necessary modules
# ============================================================
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
# ============================================================
# 1. 创建各个组件（每个都实现了 Runnable 接口）
# 1. Create individual components (all implement Runnable interface)
# ============================================================
# 提示模板：定义与 LLM 对话的格式
# Prompt template: defines the format for talking to the LLM
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant that translates {source_lang} to {target_lang}."),
("human", "{text}")
])
# LLM 模型：实际的语言模型
# LLM model: the actual language model
model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 输出解析器：将 LLM 的字符串输出原样返回（或可解析为其他格式）
# Output parser: returns the LLM's string output as-is (or can parse to other formats)
parser = StrOutputParser()
# ============================================================
# 2. 用 LCEL 的管道操作符 | 组合成链
# 2. Compose into a chain using LCEL pipe operator |
# ============================================================
# 数据流：prompt -> model -> parser
# 每个组件的输出自动成为下一个组件的输入
# Data flow: prompt -> model -> parser
# Each component's output automatically becomes the next component's input
translation_chain = prompt | model | parser
# ============================================================
# 3. 调用链（invoke 是 Runnable 接口的标准方法）
# 3. Invoke the chain (invoke is the standard method of Runnable interface)
# ============================================================
result = translation_chain.invoke({
"source_lang": "English",
"target_lang": "Chinese",
"text": "Hello, how are you today?"
})
print(result)
# 输出: 你好，今天你好吗？Parallel Execution
# ============================================================
# RunnableParallel: 并行执行多个任务
# RunnableParallel: Execute multiple tasks in parallel
# ============================================================
from langchain_core.runnables import RunnableParallel
# ============================================================
# 定义两个独立的提示模板
# Define two independent prompt templates
# ============================================================
prompt_summary = ChatPromptTemplate.from_messages([
("system", "You are a summarizer. Summarize the following text in one sentence."),
("human", "{text}")
])
prompt_keywords = ChatPromptTemplate.from_messages([
("system", "You are a keyword extractor. Extract 3 keywords from the following text."),
("human", "{text}")
])
# ============================================================
# 创建两个独立的链
# Create two independent chains
# ============================================================
summary_chain = prompt_summary | model | parser
keywords_chain = prompt_keywords | model | parser
# ============================================================
# RunnableParallel 并行执行两个链
# RunnableParallel executes both chains in parallel
# ============================================================
# 注意：两个链共享同一个输入 {"text": ...}
# Note: Both chains share the same input {"text": ...}
parallel_chain = RunnableParallel({
"summary": summary_chain,
"keywords": keywords_chain
})
# ============================================================
# 调用并行链 - 两个任务同时执行
# Invoke the parallel chain - both tasks execute simultaneously
# ============================================================
result = parallel_chain.invoke({
"text": "LangChain is a framework for developing applications powered by language models. "
"It provides tools for building chains, agents, and retrieval systems."
})
print(f"Summary: {result['summary']}")
print(f"Keywords: {result['keywords']}")RunnablePassthrough
# ============================================================
# RunnablePassthrough: 传递数据或进行转换
# RunnablePassthrough: Pass data through or transform it
# ============================================================
from langchain_core.runnables import RunnablePassthrough
# ============================================================
# 场景：在输入进入 LLM 之前，对输入进行预处理
# Scenario: Preprocess the input before it goes into the LLM
# ============================================================
# 方式1：RunnablePassthrough 原样传递数据
# Method 1: RunnablePassthrough passes data through unchanged
# 这里用 RunnablePassthrough() 占位，表示"把输入原样传给下一步"
# Here RunnablePassthrough() is a placeholder meaning "pass input to next step as-is"
passthrough_chain = (
RunnablePassthrough() # 输入原样传递 / Pass input through
| prompt # 然后进入 prompt / Then into prompt
| model # 然后进入 model / Then into model
| parser # 最后解析 / Finally parse
)
# ============================================================
# 方式2：用字典 + lambda 函数进行数据转换
# Method 2: Use dict + lambda functions for data transformation
# ============================================================
# RunnablePassthrough.assign() 可以在传递数据的同时添加/修改字段
# RunnablePassthrough.assign() can add/modify fields while passing data
from langchain_core.runnables import RunnablePassthrough
# 假设我们想：如果输入文本超过100字符，就截断
# Suppose we want to: truncate input text if it exceeds 100 characters
preprocessing_chain = (
{
# 对 "text" 字段应用转换：如果超过100字符则截断
# Apply transformation to "text" field: truncate if > 100 chars
"text": lambda x: x["text"][:100] + "..." if len(x["text"]) > 100 else x["text"],
# "source_lang" 和 "target_lang" 原样传递
# "source_lang" and "target_lang" pass through unchanged
"source_lang": lambda x: x["source_lang"],
"target_lang": lambda x: x["target_lang"],
}
| prompt
| model
| parser
)RunnableBranch
# ============================================================
# RunnableBranch: 基于条件选择不同的执行路径
# RunnableBranch: Choose different execution paths based on conditions
# ============================================================
from langchain_core.runnables import RunnableBranch
from langchain_core.prompts import ChatPromptTemplate
# ============================================================
# 定义不同场景的提示模板
# Define prompt templates for different scenarios
# ============================================================
prompt_short = ChatPromptTemplate.from_messages([
("system", "You are a concise assistant. Answer briefly in under 20 words."),
("human", "{text}")
])
prompt_long = ChatPromptTemplate.from_messages([
("system", "You are a detailed assistant. Provide comprehensive answers."),
("human", "{text}")
])
# ============================================================
# 创建对应的链
# Create corresponding chains
# ============================================================
short_chain = prompt_short | model | parser
long_chain = prompt_long | model | parser
# ============================================================
# RunnableBranch: (条件, 链) 对 的列表
# RunnableBranch: list of (condition, chain) pairs
# ============================================================
# 如果文本长度 < 50 字符，用 short_chain，否则用 long_chain
# If text length < 50 chars, use short_chain, otherwise use long_chain
branch_chain = RunnableBranch(
(lambda x: len(x["text"]) < 50, short_chain), # 条件1：短文本 / Condition 1: short text
long_chain # 默认分支 / Default branch
)
# ============================================================
# 测试：短文本走 short_chain，长文本走 long_chain
# Test: short text goes to short_chain, long text goes to long_chain
# ============================================================
short_result = branch_chain.invoke({"text": "What is AI?"})
long_result = branch_chain.invoke({"text": "Explain the complete history of artificial intelligence from 1950 to today."})