2.3 传送带：链与表达式语言LCEL

2.3 传送带：链与表达式语言 LCEL

本节摘要：链是 LangChain 装配线的本体——把提示、模型、解析器锁成一道工序。本节沿着传送带的两次进化讲：旧式 LLMChain 与顺序链如何解决"单工序"问题，LCEL 管道又如何用统一接口把任何零件自由串联、分流、汇流。

第一代传送带：LLMChain

最早的问题很具体：工艺单和发动机每次都要手工对接，能不能锁死成一道工序？LLMChain 就是这个"锁扣"：

from langchain_openai import OpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain

llm = OpenAI(temperature=0.7)

prompt = PromptTemplate(
    input_variables=["product"],
    template="给一家生产{product}的公司起一个好名字，只输出名字",
)

# 把工艺单与动力单元锁成一道工序
chain = LLMChain(llm=llm, prompt=prompt, verbose=True)
print(chain.run(product="彩色袜子"))
# 输出（verbose 会先打印组装好的提示词）：
# > Entering new LLMChain chain...
# > Finished chain.
# 彩织坊
​

单道工序不够用时，把多条链首尾相接，前一道的产品当后一道的原料——这就是顺序链：

from langchain.chains import SimpleSequentialChain

# 第一道：起名
prompt1 = PromptTemplate(input_variables=["product"],
                         template="给生产{product}的公司起名，只输出名字")
chain1 = LLMChain(llm=llm, prompt=prompt1)

# 第二道：基于名字写标语
prompt2 = PromptTemplate(input_variables=["company_name"],
                         template="为{company_name}写一句12字内标语")
chain2 = LLMChain(llm=llm, prompt=prompt2)

# 两道工序串成产线：输入原料，末端出成品
overall = SimpleSequentialChain(chains=[chain1, chain2], verbose=True)
print(overall.run("彩色袜子"))
# 输出示例：让每一天从彩色开始
​

SimpleSequentialChain 只传"单件流"——上一道的全部输出就是下一道的唯一输入。要多路进出，就用带变量表的 SequentialChain：

from langchain.chains import SequentialChain

# 每道工序声明自己的进料与出品变量名
c1 = LLMChain(llm=llm,
              prompt=PromptTemplate(input_variables=["topic"],
                                    template="为{topic}生成3个中文标题"),
              output_key="titles")
c2 = LLMChain(llm=llm,
              prompt=PromptTemplate(input_variables=["titles"],
                                    template="从这些标题里选最有创意的一个：{titles}"),
              output_key="best")

flow = SequentialChain(chains=[c1, c2],
                       input_variables=["topic"],
                       output_variables=["titles", "best"])
print(flow.invoke({"topic": "城市夜跑"}))
# 输出示例：{'topic': '城市夜跑', 'titles': '1 晚风跑道 2 霓虹配速 3 月光里程', 'best': '霓虹配速'}
​

第二代传送带：LCEL 管道

旧式链每加一种组合就要发明一个新类（顺序的、路由的、检索的），类爆炸是它最大的历史包袱。LCEL（LangChain 表达式语言）换了一个思路：让所有零件实现同一个可运行接口，再用管道符 | 任意串联：

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

prompt = ChatPromptTemplate.from_template(
    "给生产{product}的公司起名，只输出名字")

# 管道符就是传送带的对接卡槽：工艺单 | 动力单元 | 出口夹具
chain = prompt | llm | StrOutputParser()
print(chain.invoke({"product": "彩色袜子"}))
# 输出示例：彩织坊
​

三行完成了 LLMChain 的工作，但能力远不止串联。三个新零件撑起了全部分流汇流操作：

from langchain_core.runnables import RunnableLambda, RunnablePassthrough

# RunnableLambda：把普通函数编进产线
# RunnablePassthrough：原样放行，常用于"保住原料"
length_checker = RunnableLambda(lambda x: {"name": x, "len": len(x)})

# 顺序链需求：起名 → 记录长度
line = (ChatPromptTemplate.from_template("给生产{p}的公司起名，只输出名字")
        | llm | StrOutputParser() | length_checker)
print(line.invoke({"p": "彩色袜子"}))
# 输出示例：{'name': '彩织坊', 'len': 3}
​

汇流用字典语法声明并行工位，各走各的线，最后按键合并：

from langchain_core.runnables import RunnableParallel

# 并行：左线起名，右线写卖点，两线完成后打包交付
full_line = RunnableParallel(
    name=(ChatPromptTemplate.from_template("给{p}起公司名，只输出名字")
          | llm | StrOutputParser()),
    pitch=(ChatPromptTemplate.from_template("用20字说明{p}的卖点")
           | llm | StrOutputParser()),
)
print(full_line.invoke({"p": "彩色袜子"}))
# 输出示例：{'name': '彩织坊', 'pitch': '一双袜子，一天好心情，颜色自己选。'}
​

RunnablePassthrough.assign 则是"在保留原料的基础上追加新工序"的标准写法，第2.8节的检索链会反复用到它。

# 原料一路向下传递，同时追加一道新工序的产出
enrich = RunnablePassthrough.assign(
    upper=RunnableLambda(lambda x: x["name"].upper()))
print(enrich.invoke({"name": "彩织坊"}))
# 输出：{'name': '彩织坊', 'upper': '彩织坊'.upper() 形式}
​

分流：路由链

真实产线要按工件类型走不同支线——技术问题走工程师工位，售后问题走客服工位。LCEL 里用分支判断实现：

from langchain_core.runnables import RunnableLambda

support = RunnableLambda(lambda q: "已转人工客服，工单号881")
tech    = RunnableLambda(lambda q: "已转技术工程师，预计2小时响应")

def route(question: str):
    # 路由规则：命中关键词走对应支线
    return support if "退款" in question else tech

branch = RunnableLambda(
    lambda q: (support if "退款" in q else tech).invoke(q))
print(branch.invoke("我要申请退款"))
# 输出：已转人工客服，工单号881
print(branch.invoke("接口报错502"))
# 输出：已转技术工程师，预计2小时响应
​

旧式写法里对应的是 LLMRouterChain（让模型自己判断走哪条支线），思路相同，但 LCEL 版本可以手工路由与模型路由混用，调试时能先把规则写死、上线前再换成模型判断，这是旧式类做不到的渐进式改造。

⚠️ 两代传送带别混接：LCEL 管道里直接塞一个 LLMChain 实例，多数情况下能靠兼容层跑通，但流式与批量行为会变得不可预期。迁移时先把旧链整体翻成管道写法，再接入新零件。

💡 判断传送带好坏的土办法：看它是否免费获得 batch 与 stream。LCEL 管道天然继承这两种能力，而手工循环拼出来的"链"只有 invoke——这也是为什么说 LCEL 把"组装"本身变成了标准件。

传送带进化对照

本节要点回顾

• 链是锁扣：把多道工序锁成一道，是复用的最小单位；
• 两代传送带：LLMChain 系列解决有无，LCEL 管道解决自由组合；
• 四个新零件：RunnableLambda 编入函数、RunnablePassthrough 保料、assign 追加工序、字典语法开并行线；
• 路由即分流：规则路由与模型路由可渐进替换，调试期用规则，运行期交给模型；
• 迁移纪律：新旧两代不要在一条带上混接。

传送带会单向流，但真实对话是多轮往返——下一节先看给产线装上"会自己挑工具"的自治工位，2.5 节再解决"记得上一轮"的记录问题。