LangChain 实践笔记：LCEL 与 Runnable

chain = prompt | model | output_parser只有一行，但三者的输入输出并不相同，却能被同一个|串起来。

背后的关键，就是 LangChain 统一的 Runnable 抽象。

理解 Runnable 之后，链式组合、插入自定义函数、并行执行、批量处理和流式输出，其实都是同一套机制的不同用法。

#一、`|` 做了什么

| 是一次构造动作的简写。 下面两行等价：

from langchain_core.runnables import RunnableSequence

# 顺序执行每一个节点
chain = RunnableSequence(prompt, model, out)
# 与上一行等价
chain = prompt | model | out

RunnableSequence 把多个组件按顺序连成一条链，| 是它常用的组合写法。链描述的是数据在组件之间的流动。上一个组件的输出，会作为下一个组件的输入。

能串起来的关键，在于这些组件都遵循统一的 Runnable 接口。常用的调用方式有：

invoke   接受输入，返回输出
batch    接受批量输入，返回输出列表
stream   以流式方式返回结果

这三个方法在模型那一层已经见过。 之所以 | 能用于不同组件，是因为提示模板、模型、解析器等都遵循统一的 Runnable 接口。

chain.invoke({"topic": "人工智能"})   # 与调用模型是同一个写法

更重要的是 RunnableSequence 本身也是一个 Runnable，因此组合出来的链还能继续放进另一条链中，和其它 Runnable 再次组合。LCEL 的嵌套组合能力，就是这样一层层建立起来的。

#二、可以插进链里的三类对象

提示词模板、模型、解析器是框架提供的组件。写业务代码时，链上还会出现另外三类东西。

#2.1 `RunnableLambda`：把自写函数接进链

链里有一步没有现成组件时，可以直接写一个普通 Python 函数。再用 RunnableLambda 包装。

from langchain_core.runnables import RunnableLambda

# 输入字符串，返回字符串长度
def length_function(text):
    return len(text)

chain = RunnableLambda(length_function)
chain.invoke("hello")              # 5

RunnableLambda 把普通函数变成一个 Runnable，这样它就能参与 LCEL 组合，并使用 invoke、batch、stream 等统一接口。

也可以用 @chain：

from langchain_core.runnables import chain

@chain
def multiple_length_function(_dict):
    return len(_dict["text1"]) * len(_dict["text2"])

@chain 同样是把函数变成一个 Runnable。

函数放进链之后，它接收到什么输入，取决于上游传下来的数据。如果上游传来的是字典，而某个步骤需要使用到其中某个字段时，可以用 itemgetter。

from operator import itemgetter

{"a": itemgetter("k1") | RunnableLambda(length_function),
 "b": itemgetter("k2") | RunnableLambda(length_function)}

这里两个分支都会接到同一个输入字典，再分别取出 k1 和 k2 处理；这种字典写法在 LCEL 中会被组合成并行分支。

#2.2 `RunnableParallel`：一份输入分给多个任务

RunnableSequence 会按顺序执行各个步骤。 几个任务彼此独立时，可以用 RunnableParallel 同时执行，减少不必要的等待。

from langchain_core.runnables import RunnableParallel

chain = RunnableParallel(
    a=add_one,
    b=mul_two,
    c=mul_three,
)
chain.invoke(1)      # {'a': 2, 'b': 2, 'c': 3}

RunnableParallel 接收一份输入，把它同时交给多个分支执行，最后按字段名把各分支结果合成一个字典。

RunnableMap 是 RunnableParallel 的别名，两者可以等价使用。

它也可以接在链中间：

chain2 = RunnableLambda(add_one) | RunnableParallel(a=mul_two, b=mul_three)

上游的一份输出会同时进入两个分支，最终整理成一个字典继续向下传。

#2.3 `RunnablePassthrough`：保留原有数据，需要时增加字段

有时中间需要做新的计算，但后边还要使用原始数据，这时可以用 RunnablePassthrough。

from langchain_core.runnables import RunnablePassthrough

chain = RunnablePassthrough().assign(modified=lambda x: x["k1"] + "!!!")
chain.invoke({"k1": "hello world"})
# {'k1': 'hello world', 'modified': 'hello world!!!'}

RunnablePassthrough 会把输入原样传下去。 .assign() 则会在保留原字段的基础上，再加入新的计算结果。

三者可以简单区分，自定义处理用 RunnableLambda，并行分支用 RunnableParallel，保留原始输入用 RunnablePassthrough。

#三、同一条链的三种用法

三个方法前面已经介绍过，这里只看它们在整条链上的区别。

invoke 处理一次输入。

single_results = [chain.invoke({"topic": topic}) for topic in topics]

有多份输入的时候，也可以交给 batch：

inputs = [{"topic": topic} for topic in topics]
batch_results = chain.batch(inputs)

两种方式调用的是同一条链，区别在于调用方式。batch 会一次接收多个输入，并尽可能高效地批量处理。

使用时需要注意两点：每个输入都要满足链要求的输入格式；如果任务之间共享状态，还要注意并发带来的状态冲突。

stream 逐块获取输出。

for chunk in chain.stream({"topic": "人工智能"}):
    print(chunk, end="", flush=True)   # 调用侧逐块接收

stream 返回的是一个迭代器，可以逐块读取结果。如果链上的结果都支持流式处理，数据可以逐步向下传递；如果中间某一步不支持流式，可能会在那里等待处理完成。

流式最大的意义，是不用等完整结果生成完，用户就能先看到已经产生的内容。

#四、把三类对象串进一条链

前面的每一样都是零件。下面这条链是一次完整装配，需求来自一个电商客户反馈处理场景：

1. 情感分析：判断反馈的情感倾向
2. 问题分类：识别反馈中的问题类型
3. 紧急程度评估：根据内容判断处理优先级
4. 生成回复草稿：根据分析结果生成初步回复

三件事彼此独立，先并行分析，再把结果汇总给最后一步。

extract_chain = RunnableParallel(
    order_id=RunnableLambda(extract_order_id),
    original_feedback=lambda x: x,
)

analysis_chain = RunnableParallel(
    sentiment=RunnableLambda(analyze_sentiment),
    categories=RunnableLambda(classify_issue),
    urgency=RunnableLambda(assess_urgency),
)

processing_chain = (
    extract_chain
    | RunnablePassthrough.assign(analysis=itemgetter("original_feedback") | analysis_chain)
    | {
        "original_feedback": lambda x: x["original_feedback"],
        "order_id": lambda x: x["order_id"]["order_id"],
        "sentiment": lambda x: x["analysis"]["sentiment"].get("sentiment", "NEUTRAL"),
        "categories": lambda x: x["analysis"]["categories"]["categories"],
        "urgency": lambda x: x["analysis"]["urgency"]["urgency"],
        "sla_hours": lambda x: x["analysis"]["urgency"]["sla_hours"],
    }
    | RunnableLambda(generate_response)
)

这条链上有三处值得停一下。

第一处，extract_chain 同时提取订单号并保留原文。两路接收同一份输入，结果合成一个字典继续往下传。

第二处，.assign() 在原数据基础上增加 analysis。原文先交给 analysis_chain，情感、分类、紧急程度三路并行执行，结果统一挂到 analysis 字段。

第三处，中间的字典负责整理数据结构。 它从嵌套结果中取出后续步骤真正需要的字段，再交给 generate_response。

这条链把三类 Runnable 组合到了一起：RunnableLambda 负责自定义处理，RunnableParallel 负责分支并行，RunnablePassthrough 负责保留原数据并追加结果。每一步只负责一件事，复杂链也就更容易读懂和维护。

#五、总结

回到开头那个问题。| 之所以能通用，是因为参与组合的对象都遵循统一的 Runnable 抽象。组合出来的链本身也是 Runnable，因此还能继续组合。

三句话带走：

| 用来组合 Runnable，多个步骤按顺序组合后形成 RunnableSequence；
三类对象各有侧重：接自定义函数用 RunnableLambda，并行任务用 RunnableParallel，保留原始输入数据用 RunnablePassthrough；
invoke／batch／stream 是 Runnable 的常用调用方式，batch 要注意输入格式和共享状态下的并发问题。

到这里，链处理的仍然只是当前输入和模型已有的知识。如果答案来自企业文档、知识库或私有资料，就还需要先把这些内容取出来，再交给模型。

#一、| 做了什么

#二、可以插进链里的三类对象

#2.1 RunnableLambda：把自写函数接进链

#2.2 RunnableParallel：一份输入分给多个任务

#2.3 RunnablePassthrough：保留原有数据，需要时增加字段