LangChain Chains：LCEL 的真正用法

LCEL（LangChain Expression Language）是 LangChain 里最值得花时间掌握的东西。

不是因为它让代码变短了（虽然确实变短了），而是因为用 | 串联起来的 Chain，自动获得了流式输出、异步调用和批量并发——不用改任何代码，换个方法名就行。这个设计很聪明，我第一次意识到这点时觉得挺惊喜的。

这页重点讲 LCEL 的各种组合模式，以及它们在什么场景下真正有用。

---

#基础：三件东西用 | 串起来

| 在 Python 里是位运算符，LangChain 把它重载成了管道操作符。理解这点后，你就知道它不是什么魔法，只是运算符重载。

python复制
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("将 {text} 翻译成 {language}")
llm = ChatOpenAI()
parser = StrOutputParser()

chain = prompt | llm | parser

result = chain.invoke({"text": "Hello, world!", "language": "中文"})
# 你好，世界！

四种调用方式，改方法名就行：

python复制
# 同步，等完整结果
result = chain.invoke({"text": "Hello", "language": "中文"})

# 流式，边生成边返回
for chunk in chain.stream({"text": "Hello", "language": "中文"}):
    print(chunk, end="", flush=True)

# 异步，配合 FastAPI
result = await chain.ainvoke({"text": "Hello", "language": "中文"})

# 批量并发，比 for 循环快很多
results = chain.batch([
    {"text": "Hello", "language": "中文"},
    {"text": "Goodbye", "language": "中文"},
])

---

#顺序链：第一步的输出作为第二步的输入

最常见的模式——步骤 1 的结果要传给步骤 2：

python复制
from langchain_core.runnables import RunnablePassthrough

# 翻译 → 总结
translate_chain = (
    ChatPromptTemplate.from_template("将以下文本翻译成英文：\n{text}")
    | ChatOpenAI()
    | StrOutputParser()
)

summarize_chain = (
    ChatPromptTemplate.from_template("用一句话总结：\n{translated}")
    | ChatOpenAI()
    | StrOutputParser()
)

# 把翻译结果传给总结
full_chain = (
    {"translated": translate_chain}
    | summarize_chain
)

result = full_chain.invoke({"text": "这是一段很长的中文文本..."})

---

#并行链：同时做多件事

这个功能我用得相当频繁。比如分析一段文字，同时提取摘要、关键词和情感——三个 LLM 调用并发进行，总时间接近单次调用，而不是三倍。对于内容分析类任务，这个优化非常明显。

python复制
from langchain_core.runnables import RunnableParallel

parallel_chain = RunnableParallel(
    summary=(
        ChatPromptTemplate.from_template("用两句话总结：{text}")
        | ChatOpenAI()
        | StrOutputParser()
    ),
    keywords=(
        ChatPromptTemplate.from_template("提取 5 个关键词，逗号分隔：{text}")
        | ChatOpenAI()
        | StrOutputParser()
    ),
    sentiment=(
        ChatPromptTemplate.from_template("判断情感倾向（正面/负面/中立）：{text}")
        | ChatOpenAI(temperature=0)
        | StrOutputParser()
    ),
)

result = parallel_chain.invoke({"text": "这是一篇关于 AI 发展的文章..."})
print(result["summary"])    # 摘要
print(result["keywords"])   # 关键词
print(result["sentiment"])  # 情感

---

#RunnablePassthrough：透传原始输入

做 RAG 时经典用法——检索文档的同时，把原始问题原样传给下一步：

python复制
from langchain_core.runnables import RunnablePassthrough

rag_chain = (
    {
        "context": retriever | format_docs,  # 用问题检索文档
        "question": RunnablePassthrough(),   # 问题本身原样透传
    }
    | prompt
    | llm
    | StrOutputParser()
)

RunnablePassthrough() 就是一个"直通管道"，什么都不做，把输入原样送到下一步。

---

#RunnableLambda：插入自定义 Python 逻辑

有时候需要在 Chain 中间做一些纯 Python 处理（格式化、过滤、转换），用 RunnableLambda 包一下就能接进去：

python复制
from langchain_core.runnables import RunnableLambda

def clean_text(text: str) -> str:
    """预处理：去掉多余空格和换行"""
    return " ".join(text.split())

def add_timestamp(result: str) -> dict:
    """后处理：给结果加时间戳"""
    from datetime import datetime
    return {"result": result, "timestamp": datetime.now().isoformat()}

chain = (
    RunnableLambda(clean_text)      # 前处理
    | prompt
    | llm
    | StrOutputParser()
    | RunnableLambda(add_timestamp)  # 后处理
)

---

#带回退的 Chain：主模型挂了自动切备用

GPT-4o 偶尔会有服务中断，给生产环境的 Chain 配一个备用模型：

python复制
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

main_chain = (
    prompt
    | ChatOpenAI(model="gpt-4o")
    | StrOutputParser()
)

fallback_chain = (
    prompt
    | ChatAnthropic(model="claude-3-5-sonnet-20241022")
    | StrOutputParser()
)

# 主链失败时自动用备用链
robust_chain = main_chain.with_fallbacks([fallback_chain])

---

#调试：在 Chain 中加日志

有时候需要看中间步骤的输入输出，插一个 RunnableLambda 打印就行：

python复制
def debug_log(x, label=""):
    print(f"\n[DEBUG {label}]: {str(x)[:200]}")
    return x

chain = (
    RunnableLambda(lambda x: debug_log(x, "输入"))
    | prompt
    | RunnableLambda(lambda x: debug_log(x, "Prompt后"))
    | llm
    | parser
)

更推荐的方式是用 LangSmith，能看到完整的调用链，不用在代码里加 print。但开发阶段这种临时 log 够用。

---

#条件逻辑：简单分支用 RunnableBranch

python复制
from langchain_core.runnables import RunnableBranch

branch = RunnableBranch(
    (lambda x: x["type"] == "code", code_review_chain),
    (lambda x: x["type"] == "text", text_analysis_chain),
    default_chain  # 默认分支
)

说实话，RunnableBranch 的语法我觉得挺别扭的。对于简单的 if/else 路由还行，但一旦有多层嵌套或者需要循环，可读性会很快变差。遇到复杂分支逻辑，我更倾向于直接用 LangGraph——可读性好很多，也能处理循环。

---

#组合的原则

几个实际使用中总结的原则：

把长 Chain 拆成命名子 Chain。step1 | step2 | step3 | step4 | step5 | step6 这种写法在出问题时很难定位是哪一步。拆成 preprocess_chain | process_chain | postprocess_chain，调试和测试都容易很多。

并行能用就用。如果多个 LLM 调用之间没有依赖关系，RunnableParallel 能让总延迟接近单次调用。这是做内容分析类应用时最值得用的优化。

流式输出从一开始就考虑。如果你的应用需要实时展示生成内容，Chain 里所有环节都要支持流式——Parser 用 StrOutputParser 而不是 JsonOutputParser（后者要等完整 JSON 才能解析，流式就没意义了）。

---

#小结

1. LCEL 的 | 不只是语法糖，它让 Chain 自动支持 invoke/stream/ainvoke/batch 四种调用方式。
2. RunnableParallel 是做多路 LLM 调用最干净的方式，并发执行，速度接近单次调用。
3. RunnablePassthrough 用于透传原始输入，RAG 场景里必用。
4. RunnableLambda 把任意 Python 函数接入 Chain，适合预处理/后处理。
5. 复杂分支逻辑别强行用 RunnableBranch，上 LangGraph 更清晰。

---

下一步：Agents 代理系统 — 给 AI 配工具，让它自主决定调用哪个

相关参考：LCEL 官方文档↗ | Runnable 接口↗