LCEL与Runnable

把 prompt、model、parser 写成嵌套函数调用，改一步就要动三处签名；要流式输出、批量跑积压、主模型超时换备用，还得各写一套编排。LCEL（LangChain Expression Language）用 | 把实现 Runnable 的对象串成管道，输入从左进、输出从右出；组合成的仍是 Runnable，因此 invoke / stream / batch / ainvoke 共用一条链。

段末注释：LCEL = 用 | 连接 Runnable 的声明式语法；Runnable = 实现 invoke/stream 等协议的可组合单元。

社区方案即官方 langchain-core 的 Runnable 协议（2023 年起替代 LLMChain / SequentialChain 的关键字拼装）。适用线性流水线、需要流式与批量时；风险是把业务 if/else 全塞进 RunnableLambda 会失去声明式优势，有环、检查点应换图编排。

---

1. 一句话定位

维度	内容
角色	能力层组合：声明式串联 prompt / model / parser / 自定义步骤
输入 → 输出	链首类型 → 链尾类型（如 dict → str）
典型调用入口	chain = a | b | c、chain.invoke(x)、chain.stream(x)、chain.batch([...])
与 LangGraph	线性无环流水线用 LCEL 足够；有环 / checkpoint 用 LangGraph

出现背景：旧 Chain 类把 prompt、llm 藏在构造器参数里，换组件要改类层次；LCEL 把每一步提成 Runnable，| 只表达数据依赖。v1 后旧 Chain 迁入 langchain-classic，新代码默认走本篇写法。

---

2. 实现逻辑

1. prompt = ChatPromptTemplate(...)
2. chain = prompt | model | StrOutputParser()
3. chain.invoke({"question": "..."})
   Step A: dict → list[BaseMessage]
   Step B: messages → AIMessage
   Step C: AIMessage → str (content)
4. chain.stream(...) 逐步产出 parser 前/后的 chunk（视实现）
5. chain.batch([{...}, {...}]) 并行多输入

字段级变形：

{"city": "北京"}
  → |prompt|  → [SystemMessage, HumanMessage]
  → |model|   → AIMessage(content="...")
  → |parser|  → "..."

---

3. 原理说明

3.1 Runnable 协议

Runnable（抽象类）
功能：统一 invoke / batch / stream / ainvoke 等入口。| 构造 RunnableSequence。
默认值：config=None。
最小维度：单步 invoke 的 input 由该 Runnable 的 InputType 决定，不能是 None（除非类型允许）。

invoke(input, config=None)（方法）
功能：同步跑完整条。返回该步 OutputType。

batch(inputs, config=None, *, return_exceptions=False)（方法）
默认 return_exceptions=False。inputs len≥1；空列表 → []。

stream(input, config=None)（方法）
产出 chunk；prompt 等非流式步骤往往一次吐完整中间值。

chain = prompt | model | parser
chain.invoke({"q": "hi"})

__or__ / RunnableSequence（组合）
功能：left | right，左输出 = 右输入。
最小维度：至少 2 个 Runnable；单对象不必 |。

3.2 组合即协议继承（LCEL 相对手写的核心差）

手写 parser.parse(model.invoke(prompt.format(x))) 只有同步单次。要 stream/batch/ainvoke 得各写一遍。| 的结果仍是 Runnable：框架转发 chunk、线程池跑 batch、把 ainvoke 落到各步。声明数据流一次，六种调用入口跟来。

段末注释：协议继承 = 子 Runnable 组合成新 Runnable 后，仍暴露同一套 invoke/stream/batch/async 方法。

3.3 RunnableParallel

RunnableParallel（类，也可写作 dict 字面量进链）
功能：同一输入扇出到多支路，合并为 dict。独立 I/O 总延迟接近最慢一支。

项	类型	默认值	最小维度
支路	Mapping[str, Runnable]	必填	≥1 个键；2 键才有并行意义
输出	dict	—	键 = 支路名

RunnableParallel(text=RunnablePassthrough(), draft=summarize)
# 等价：{"text": RunnablePassthrough(), "draft": summarize}

RunnablePassthrough（类）
功能：原样返回输入。无字段。最小输入 = 上游类型（常为 dict 或 str）。

RunnablePassthrough().invoke({"q": "x"})  # → {"q": "x"}

3.4 RunnableLambda

RunnableLambda(func)（类）
功能：包装任意 callable。异常中断整条链，除非外层 fallback。
默认值：无；func 必填。
最小维度：func 至少 1 个位置参数（input）；可选第二参 RunnableConfig。

from langchain_core.runnables import RunnableLambda
RunnableLambda(lambda x: x["a"] + x["b"]).invoke({"a": 1, "b": 2})  # 3

Lambda 按整段输入/输出工作，不会把 model 的 token chunk 自动变成自定义增量逻辑。

3.5 与 LangGraph 的边界

LCEL 无环。条件分支用 RunnableBranch（至少 1 条条件 + 1 个 default）。多轮工具环、持久化 state 应上图。

---

4. 最小可运行示例

pip install -U langchain-openai

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

model = ChatOpenAI(
    model="qwen3.5:9b",
    api_key="ollama",
    base_url="http://localhost:11434/v1",
    temperature=0,
)
prompt = ChatPromptTemplate.from_messages([("human", "用一句话总结：{text}")])
summarize = prompt | model | StrOutputParser()

# 并行：原文原样留下 + 生成摘要
chain = RunnableParallel(text=RunnablePassthrough(), draft=summarize)
out = chain.invoke({"text": "LCEL 用竖线把 Runnable 串成管道，组合成的对象仍是 Runnable。"})
print(list(out.keys()), out["draft"])
# 预期：keys 为 ['text', 'draft']；draft 是一句话摘要（措辞随模型变）

print(summarize.invoke({"text": "列表推导式用一行从可迭代对象生成列表。"}))
print("".join(summarize.stream({"text": "列表推导式用一行从可迭代对象生成列表。"})))
# 预期：invoke 与 stream 拼起来同义；stream 为多个 str chunk

out["text"] 是并行支路里 Passthrough 的输入 dict。生产异步：await summarize.ainvoke({"text": "..."})，参数与同步一致。

---

5. 案例：客服工单草稿（对照手写 vs LCEL）

教学场景：用户提交一条工单，系统要 检索相似历史 与 打紧急度（二者互不依赖），再生成给客服的回复草稿。真实线上还要：聊天窗口流式打字、夜间 batch 清积压、主模型超时切备用。

5.1 手写编排：只有 invoke

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

def ollama_chat(model: str = "qwen3.5:9b") -> ChatOpenAI:
    """指向本地 Ollama 的 ChatOpenAI。输入为模型名，输出为客户端。"""
    return ChatOpenAI(
        model=model, api_key="ollama",
        base_url="http://localhost:11434/v1", temperature=0,
    )

def retrieve_similar(ticket: str) -> str:
    kb = {
        "发票": "历史#12：核对开票主体后重发 PDF。",
        "登录": "历史#7：清缓存后走 SSO 重置。",
    }
    for k, v in kb.items():
        if k in ticket:
            return v
    return "无相似工单。"

model = ollama_chat()
draft_prompt = ChatPromptTemplate.from_template(
    "工单：{ticket}\n紧急度：{urgency}\n相似处理：{similar}\n写一条给用户的回复草稿，不超过两句。"
)

def handle_ticket(ticket: str) -> str:
    similar = retrieve_similar(ticket)          # 先等检索
    urgency = model.invoke(
        ChatPromptTemplate.from_template(
            "只输出一个字：高、中或低。工单：{ticket}"
        ).format_messages(ticket=ticket)
    ).content
    return model.invoke(
        draft_prompt.format_messages(ticket=ticket, similar=similar, urgency=urgency)
    ).content

print(handle_ticket("发票抬头开错了"))
# 预期形态：草稿提到核对开票主体或重发 PDF（措辞随模型变）

检索与分类在函数里串行；要 stream / batch / ainvoke / 降级，必须改 handle_ticket 的签名与内部实现（通常再抄 2～3 份）。

5.2 LCEL：同一条链覆盖并行 + 三种调用

零件与 5.1 相同（检索函数、草稿 prompt、同一 Ollama），只改编排。

from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnableParallel
from openai import APIConnectionError, APITimeoutError

def ollama_chat(model: str = "qwen3.5:9b", **kwargs) -> ChatOpenAI:
    """指向本地 Ollama 的 ChatOpenAI。kwargs 可覆盖 base_url / timeout。"""
    return ChatOpenAI(
        model=model, api_key="ollama",
        base_url="http://localhost:11434/v1", temperature=0, **kwargs,
    )

def retrieve_similar(ticket: str) -> str:
    kb = {
        "发票": "历史#12：核对开票主体后重发 PDF。",
        "登录": "历史#7：清缓存后走 SSO 重置。",
    }
    for k, v in kb.items():
        if k in ticket:
            return v
    return "无相似工单。"

model = ollama_chat()
classify = (
    ChatPromptTemplate.from_template("只输出一个字：高、中或低。工单：{ticket}")
    | model
    | StrOutputParser()
)
draft_prompt = ChatPromptTemplate.from_template(
    "工单：{ticket}\n紧急度：{urgency}\n相似处理：{similar}\n写一条给用户的回复草稿，不超过两句。"
)
# 主路径失败时切到另一本地模型；健康时不会走到 backup
backup = ollama_chat("llama3.1:8b")
primary = model.with_fallbacks(
    [backup],
    exceptions_to_handle=(APIConnectionError, APITimeoutError, TimeoutError, OSError),
)

prep = RunnableParallel(
    ticket=RunnableLambda(lambda x: x["ticket"]),
    similar=RunnableLambda(lambda x: retrieve_similar(x["ticket"])),
    urgency=classify,
)
chain = prep | draft_prompt | primary | StrOutputParser()

ticket = {"ticket": "发票抬头开错了"}
print(chain.invoke(ticket))
print("".join(chain.stream(ticket)))
print(chain.batch([
    {"ticket": "无法登录，提示 401"},
    {"ticket": "发票抬头开错了"},
], config={"max_concurrency": 2}))
# 预期：invoke 与 stream 拼起来同义；batch 两条分别贴近「登录/SSO」与「发票/PDF」

classify 与 similar 互不依赖，Parallel 同时跑；下游 prompt 一次吃齐 {ticket, similar, urgency}。要把主路径换成不可达地址以观察降级，只需 ollama_chat(base_url="http://127.0.0.1:9", timeout=5).with_fallbacks([backup])，调用点仍是 chain.invoke / stream / batch。

能力	5.1 手写 handle_ticket	5.2 LCEL chain
检索 ∥ 分类	串行，自己改才并行	RunnableParallel 默认并发
单次调用	有	invoke
流式打字机	要重写函数	stream（及 astream）
夜间积压	自管线程池	batch（max_concurrency）
FastAPI 非阻塞	再写 async def	await ainvoke
主模型降级	try/except 包一层	with_fallbacks 仍是同一 Runnable
换 parser / prompt	改函数体与所有调用点	只换 `

---

重要配置参数

参数（API 名）	类型 / 默认值	功能说明	作用与影响	参考起点	配置指导
config	RunnableConfig，invoke 第二参	单次运行的 tags/metadata/callbacks/configurable	观测与租户字段放这里，不写进构造器	invoke 第二参	空 {} 合法，走默认
max_concurrency	int，batch/并行时	同时跑多少条输入或 Parallel 支路	过大打满 API 限流；过小吞吐低	5～20	受 rate limit 约束
RunnableSequence 顺序	`	` 左右顺序	规定数据从哪一站流到哪一站	乱序则类型不兼容或 parser 吃到未生成文本	prompt→model→parser
stream 模式	方法，无额外默认	按 token/块增量产出，而不是等整段	下游 parser 不支持 chunk 会解析失败	用户可见打字机	StrOutputParser 可增量
with_fallbacks	链级/单步方法	失败时换另一套 Runnable	备路径输出形态须与主路径兼容	主备各 1 个	备模型也要能接同一 input
bind	链组装时	把 temperature 等打进模型副本	构造期固定；单次调用用 config	在 | 前 bind	与运行时 config 分工
Parallel 键名	str，必填	RunnableParallel 输出 dict 的键，供下游 prompt 引用	与 {var} 不一致则缺变量	与模板变量逐字相同	先定 prompt 再定键名

---

6. 易踩坑

1. Parallel 键名与 prompt 变量不一致：{"ctx": ...} 但模板写 {context}。
2. Lambda 里改输入类型：下一步 model 期望 messages 却收到 dict。
3. 以为 LCEL 自动做工具循环：| 不会执行 tool；Agent 需单独循环或 create_agent。
4. 在 Lambda 里重写编排：又回到 5.1，stream/batch 不会自动出现。

适用：prompt → model → parser、RAG 组装、独立 I/O 可并行、需要 stream/batch 的线性流水线。
不适用：多轮 tool 环、人工审批、会话 checkpoint——那些不是 | 能表达的控制流。

---

小结

• LCEL 用 | 连接 Runnable；组合后仍是 Runnable，invoke / stream / batch / ainvoke 共用。
• RunnableParallel 把互不依赖的步骤并发；工单案例里检索与紧急度应扇出，而不是写进一个函数里排队。
• RunnableLambda 插入自定义逻辑，但不要用它取代管道去手写编排。
• 线性流水线用本篇；有环用 LangGraph。

---

参考链接

• LCEL 概念
• Runnable API
• RunnableParallel