Agent 开发笔记（六）：用 LCEL 将组件组合为执行链

在 LangChain 应用中，Prompt、ChatModel 和 Parser 分别调用并不复杂，但当流程变长后，开发者需要反复处理不同的输入输出类型，也不便于插入或替换组件。LCEL 用统一的 Runnable 协议和管道符组织这些步骤，让执行链更紧凑，也为流式、批量与异步调用提供一致接口。

Agent 开发学习笔记（六）：LCEL：把组件串成链

前言

前面几篇，我们把三件套拆开讲完了：Prompt 负责组装提示词，ChatModel 负责对话，Parser 负责把模型输出转成结构化数据。

单独用都没问题，但真串起来跑一次，你得手动调三次 invoke，还得记住每一步的输入输出类型不一样——prompt 出来是 PromptValue，ch@t 出来是 AIMessage，parser 才拿到字符串。

LangChain 提供了一个更顺的写法，用管道符 | 把组件串成一条链。前几篇里凡是出现 | 的地方，都写了「详见 LCEL 篇」——就是这里。

这一篇要回答的是：| 凭什么能把组件串起来？串起来之后怎么跑？除了三件套，还有哪些东西能被串？怎么调用、怎么流式、怎么批量？出错了怎么办？

1. LCEL 是什么，为什么有它

1.1 手动三件套 vs LCEL

前面四篇，我们把三件套拆开讲完了：Prompt 负责组装提示词，ChatModel 负责对话，Parser 负责把模型输出转成结构化数据。

单独用都没问题，但真串起来跑一次，是这样的：

prompt_value = prompt.invoke({"topic": "康师傅绿茶"})
ai_message = [email protected](prompt_value)
result = output_parser.invoke(ai_message)

三步，每步的输入输出类型还不一样：prompt 出来是 PromptValue，ch@t 出来是 AIMessage，parser 才拿到字符串。你得记住每一步吃进去什么、吐出来什么，中间任何一步想加点日志、换个解析器，都得手动改这三行。

LCEL 是 LangChain Expression Language 的缩写，中文叫「LangChain 表达式语言」。它是 LangChain 提供的一套声明式组合语法，核心就是用 | 把组件串成链。

LCEL 把三行压成一行：

chain = prompt | ch@t | output_parser
result = chain.invoke({"topic": "康师傅绿茶"})

| 就是那个管道符。这不是字符串拼接，是 LangChain 定义的一套组合语法。前几篇里凡是出现 | 的地方，都写了「详见 LCEL 篇」——就是这里。

1.2 | 做了什么：Runnable

先看 | 本身

| 有个硬性要求：两边都得是 Runnable。

Runnable 是什么？你可以把它想成家里的插座规格。插座规定了两孔还是三孔、220V、什么形状——只要电器符合这个规格，插上去就能用，你不用管它是台灯、充电器还是电视。

Runnable 就是 LangChain 里的「插座规格」。它规定：只要一个对象实现了 invoke / stream / batch / ainvoke 这四个方法，它就符合规格，就能被 | 串进链里。

从底层看，Runnable 是一个抽象基类 langchain_core.runnables.base.Runnable，定义了这四个方法：

class Runnable:
    def invoke(self, input, config=None): ...
    def stream(self, input, config=None): ...
    def batch(self, inputs, config=None): ...
    async def ainvoke(self, input, config=None): ...

谁继承它、实现这些方法，谁就是 Runnable。实际使用中你不需要自己写 class MyThing(Runnable)，因为：

• Prompt、ChatModel、Parser 这些官方组件已经继承好了，直接用就行。
• 你自己写的普通函数，用 RunnableLambda 包一下，就自动变成 Runnable。
• 你自己写的生成器，用 RunnableGenerator 包一下，同理。

| 返回了什么

| 这个操作符，底层是 Runnable.__or__。写 a | b 时，Python 调用 a.__or__(b)，返回一个新的 Runnable，类型是 RunnableSequence。

所以：

chain = prompt | ch@t | output_parser

等价于：

chain = prompt.__or__(ch@t).__or__(output_parser)

先 prompt.__or__(ch@t) 得到一个 RunnableSequence，再拿它 .__or__(output_parser)，得到最终的 RunnableSequence。每一步 | 都在套一层「先跑左边、再跑右边」。

RunnableSequence 是什么

RunnableSequence 外表是一条链，内里是一个装着所有组件的 steps 列表。

chain = prompt | ch@t | output_parser 的结果，就是一个 steps = [prompt, ch@t, output_parser] 的 RunnableSequence。它的本质就一句话：按顺序跑 steps 里的每一个 Runnable，上一节的输出直接喂给下一节。

它的 invoke 逻辑极其简单，核心就是一个 for 循环：

# 简化后的 RunnableSequence.invoke 逻辑
input = 你传入的输入
for step in self.steps:
    input = step.invoke(input)   # 上一节的输出，直接变成下一节的输入
return input                     # 最后一节的输出，就是整条 chain 的返回值

没有类型检查，没有「识别」这一步。它只是把 input 这个变量反复赋值——先传给第一步，拿到结果；把这个结果再传给第二步，拿到新结果；再传给第三步……直到最后一节，直接把那个结果返回。

chain 怎么跑：一次具体的接力

用具体例子看：

chain = prompt | ch@t | output_parser

chain.invoke({"topic": "康师傅绿茶"})

这一行 invoke 内部依次发生：

输入：{"topic": "康师傅绿茶"}          ← 整条 chain 的输入，字典
  ↓ prompt.invoke({"topic": "康师傅绿茶"})
中间：PromptValue                     ← prompt 的输出
  ↓ [email protected](PromptValue)
中间：AIMessage                       ← ch@t 的输出，文本在 .content
  ↓ output_parser.invoke(AIMessage)
输出："康师傅绿茶，清新解渴……"         ← 整条 chain 的输出，字符串

由此得到两个结论。

第一，整条 chain 的输入类型由第一节决定，输出类型由最后一节决定。 上面这条链，输入必须是 prompt 要的字典 {"topic": ...}，输出必然是 output_parser 吐出的字符串。

如果链里没有 parser：

chain = prompt | ch@t
chain.invoke({"topic": "康师傅绿茶"})   # 返回 AIMessage

最后一节是 ch@t，[email protected](PromptValue) 返回 AIMessage，整条 chain 就返回 AIMessage。RunnableSequence 根本不关心最后一节是什么，它只做一件事：把最后一节的 invoke 返回值原样抛出去。

第二，chain 自己也是 Runnable。 因为它有明确的输入类型和输出类型，内部实现了 invoke / stream / batch / ainvoke，所以它符合 Runnable 协议。既然符合，就能继续用 | 接下一节：

bigger_chain = chain | another_step

bigger_chain 的 steps 变成 [prompt, ch@t, output_parser, another_step]。调用时，output_parser 吐出的字符串会原样喂给 another_step.invoke，继续接力。这就是 chain 可以无限拼接的原因。

类型不匹配怎么办

既然不识别类型，那如果中间某一节的输出，下一节根本吃不下，会怎样？

答案是：运行时才会炸，而且炸的是下一节的 invoke，不是 RunnableSequence。

比如你写 ch@t | ch@t。第一个 ch@t 输出 AIMessage，RunnableSequence 把这个 AIMessage 直接传给第二个 ch@t 的 invoke。第二个 ch@t 期望收到 PromptValue 或消息列表，收到 AIMessage 时自己内部会报错。RunnableSequence 只是「接力」，不负责「判断接力棒合不合适」。

四个方法各自怎么用，留到第 4 节。这里只记住：符合这份规格的东西，就能进管道；组合出来的 chain 也是其中一员。

小结：到这里，| 的机制就说完了：Runnable 是一份要求实现四个方法的协议，| 把符合协议的组件组合成 RunnableSequence，后者靠 for 循环接力执行——上一节吐什么，下一节就接什么。chain 自己也是 Runnable，所以能继续拼下去。

1.3 PipelinePromptTemplate 的退场

| 能把 prompt、ch@t、parser 串起来，但 Prompt 自己也有「组装」的需求。

假设你要拼一个这样的提示词：

你是一个翻译助手。

请回答：今天天气怎么样？

前半句是角色设定，后半句是用户问题。在 LCEL 之前，LangChain 有个专门的类来干这件事，叫 PipelinePromptTemplate。

它这样用。先定义零件：

from langchain_core.prompts import PromptTemplate, PipelinePromptTemplate

final = PromptTemplate.from_template("{greeting}nn{question}")
greeting = PromptTemplate.from_template("你是一个{role}。")
question = PromptTemplate.from_template("请回答：{q}")

再用它把零件串起来：

pipeline = PipelinePromptTemplate(
    final_prompt=final,
    pipeline_prompts=[
        ("greeting", greeting),
        ("question", question),
    ]
)

result = pipeline.invoke({"role": "翻译助手", "q": "今天天气怎么样？"})

它做的事就是：先渲染 greeting、question，把结果分别塞进 final 的 {greeting}、{question} 占位符，最后输出完整提示词。

这个类现在被官方废弃了，后续版本直接删除。原因很简单：过度设计。 它本质上就是「依次渲染几个 Prompt 再拼接」，用一个专用类来做，反而增加了调试难度——出了问题你很难判断是哪个子模板渲染错了，变量依赖关系也不透明。

更重要的是，LCEL 已经能覆盖它的全部功能，而且更灵活。用 LCEL 替代它，写法是：

from langchain_core.runnables import RunnableParallel, RunnableLambda

composed = RunnableParallel(
    greeting=greeting,
    question=question,
)

def format_and_strip_text(inputs):
    greeting_str = inputs['greeting'].text
    question_str = inputs['question'].text
    return f"{greeting_str}nn{question_str}".strip()

full_prompt = RunnableLambda(format_and_strip_text)
pipeline_chain = composed | full_prompt

这里用到了 RunnableParallel 和 RunnableLambda——它们是第 3 节的主角。现在只需要知道：LCEL 不只是「串三件套」，它还能表达 Prompt 内部的组装逻辑，所以 PipelinePromptTemplate 没有存在的必要了。

小结

这一节回答了两个问题：LCEL 是什么——它用 | 把 Runnable 串成链，一次 invoke 跑完全程；为什么有它——手动三步变成一行，而且连 Prompt 内部的组装逻辑也能用 Runnable 表达，PipelinePromptTemplate 这类专用类因此退场。

2. Runnable 体系

1.2 里四个方法只列了名字，这一节给每个方法一句话说明，具体用法和差异留到第 4 节。

2.1 Runnable 协议：四个方法

协议要求实现四个方法：

• invoke(input)：单次调用，传入一个输入，返回一个输出。
• stream(input)：流式调用，传入一个输入，返回一个迭代器，逐块产出输出。
• batch(inputs)：批量调用，传入一组输入，返回一组输出。
• ainvoke(input)：异步单次调用，invoke 的 async 版本。

四个方法的具体用法和差异，第 4 节展开。这里先记住：只要是 Runnable，就一定有这四个方法，所以调用方式永远统一。

2.2 Runnable 家族清单

按用途分组：

核心三件套

组件	作用	本篇
Prompt	组装提示词	已在 Prompt 篇讲
ChatModel	对话模型	已在 Model 篇讲
Parser	解析模型输出	已在 Parser 篇讲

组合类

组件	作用	本篇
RunnableParallel	并行跑多个分支，再合并结果	第 3 节展开
RunnablePassthrough	输入原样透传	第 3 节展开

自定义类

组件	作用	本篇
RunnableLambda	把普通函数接进链	第 3 节展开，Parser 篇已讲基础用法
RunnableGenerator	把生成器接进链，支持流式	第 3 节展开，Parser 篇已讲基础用法

本篇不讲的

Retriever、Tool 等也是 Runnable，但属于检索、工具调用话题，不在本篇范围。

小结：这些就是 LCEL 能串的全部对象：核心三件套、组合类、自定义类，以及本篇不展开的 Retriever 等。记住一点：只要是 Runnable，就有统一的四个方法，调用方式永远一致——这正是第 4 节要展开的。

3. RunnableParallel / RunnablePassthrough / RunnableLambda

第 2 节的清单里，Prompt、ChatModel、Parser 是核心三件套，前面几篇已经讲透了。这一节讲另外三个：它们不是「三件套」，而是组合和改造数据流的工具。理解它们，才算真正会用 LCEL。

3.1 RunnableLambda：把普通函数接进链

RunnableLambda 解决一个很实际的问题：链里想插一步自定义处理，但这一步不是官方组件，就是一个普通函数。

比如把模型输出的大小写翻转：

from langchain_core.messages import AIMessage

def parse(ai_message: AIMessage) -> str:
    return ai_message.content.swapcase()

chain = ch@t | parse
chain.invoke("hello")

这里 parse 就是个普通函数。用 | 把它拼进链里时，LangChain 会自动把它包装成 RunnableLambda，所以它能接在 ch@t 后面。

Parser 篇里讲过用 RunnableLambda 写自定义解析器。那是它的一个应用场景。放到 Runnable 体系里看，它的定位更清楚：它是「官方组件」和「你自己写的函数」之间的适配器。 任何一步逻辑，只要输入输出能对上，都能用 RunnableLambda 包一下接进链。

3.2 RunnablePassthrough：输入原样透传

RunnablePassthrough 做的事就一件：拿到什么，原样返回什么。 看起来像什么都没干，但它解决的是「输入结构不匹配」。

看这条链：

chain = (
    {"input": RunnablePassthrough()}
    | prompt
    | llm
    | json_parser
)

prompt 要的输入是一个字典，形如 {"input": "用户的原话"}。但用户调用时，手上往往只有一句裸字符串：

chain.invoke("John is 20 years old...")

{"input": RunnablePassthrough()} 这一步的作用，就是把这句裸字符串包成 prompt 要的字典。它的执行过程是：

输入："John is 20 years old..."
  ↓ {"input": RunnablePassthrough()}
中间：{"input": "John is 20 years old..."}
  ↓ prompt.invoke(...)
...

RunnablePassthrough() 本身把裸字符串原样返回，外层的 {"input": ...} 给它套了一个键名。合起来就是「把输入包成字典」。

所以 RunnablePassthrough 的典型用途是：链的第一节输入格式和后面组件要的不一致时，用它做一层适配，同时保留原始输入不变。

3.3 RunnableParallel：并行跑多个分支

RunnableParallel 解决的是「一份输入，要同时喂给多个分支，再把结果合并」。

最直观的例子是 Prompt 组装。假设要拼一个这样的提示词：

你正在冒充 Elon Musk。

下面是一个交互示例：
Q：你最喜欢什么车？
A：Tesla

现在正式开始！
Q：您最喜欢的社交媒体网站是什么？
A：

它由三个零件组成：introduction、example、start。三个零件都只依赖输入里的某几个变量，互相不依赖，可以并行渲染。

from langchain_core.runnables import RunnableParallel, RunnableLambda
from langchain_core.prompts import PromptTemplate

introduction_prompt = PromptTemplate.from_template("你正在冒充{person}。")

example_prompt = PromptTemplate.from_template(
    "下面是一个交互示例：nnQ：{example_q}nA：{example_a}"
)

start_prompt = PromptTemplate.from_template("现在正式开始！nnQ：{input}nA：")

composed = RunnableParallel(
    introduction=introduction_prompt,
    example=example_prompt,
    start=start_prompt,
)

composed 拿到输入字典后，会同时跑三个分支：introduction_prompt.invoke(...)、example_prompt.invoke(...)、start_prompt.invoke(...)，然后把三个结果按名字合并成一个新字典：

输入：{"person": "Elon Musk", "example_q": ..., "example_a": ..., "input": ...}
  ↓ RunnableParallel 并行跑三个分支
中间：{
    "introduction": PromptValue(...),
    "example": PromptValue(...),
    "start": PromptValue(...),
}

拿到这个字典后，再用 RunnableLambda 把三个 PromptValue 的 .text 取出来、拼成完整字符串：

def format_and_strip_text(inputs):
    intro_str = inputs['introduction'].text
    example_str = inputs['example'].text
    start_str = inputs['start'].text
    return f"{intro_str}nn{example_str}nn{start_str}".strip()

full_prompt = RunnableLambda(format_and_strip_text)
pipeline_chain = composed | full_prompt

这就是 1.3 里那个 PipelinePromptTemplate 替代方案的完整形态。RunnableParallel 负责并行渲染三个子 Prompt，RunnableLambda 负责拼接，两者串起来，就替代了那个被废弃的专用类。

再往后接上模型和解析器，就是一条完整的链：

chain = composed | full_prompt | ch@t | output_parser
chain.invoke({
    "person": "Elon Musk",
    "example_q": "你最喜欢什么车？",
    "example_a": "Tesla",
    "input": "您最喜欢的社交媒体网站是什么？",
})

小结：这一节做的事，拆开看是三步：RunnableParallel 并行渲染三个 Prompt，拿到三个 PromptValue；RunnableLambda 把三个 PromptValue 的 .text 取出来，拼成完整字符串；因为拼接函数被 RunnableLambda 包装过，它本身也是 Runnable，所以能继续用 | 接上 ch@t 和 parser。

3.4 三者对比

Runnable	作用	典型场景
RunnableLambda	把普通函数包成 Runnable	链里插一步自定义处理
RunnablePassthrough	输入原样透传	输入结构和下游组件不匹配，做适配
RunnableParallel	并行跑多个分支，合并结果	一份输入同时喂多个组件

三者经常一起出现：RunnableParallel 负责分叉，RunnableLambda 负责处理分叉后的结果，RunnablePassthrough 负责在最前面把输入包成需要的形状。

小结

这三个 Runnable 让 LCEL 不只能串三件套，还能改造数据流：RunnableLambda 负责插入自定义处理，RunnablePassthrough 负责适配输入形状，RunnableParallel 负责分叉并行。三者经常一起出现，前面 Elon Musk 那条链就是例子。

4. invoke / stream / batch / ainvoke

第 2.1 节说 Runnable 协议要求实现四个方法，这一节交代它们各自怎么用。

四个方法的关系，一句话概括：invoke 是基础，stream 是它的流式版，batch 是它的批量版，ainvoke 是它的异步版。 四个方法跑的是同一条 chain，只是「怎么喂输入、怎么拿输出」不同。

4.1 invoke：单次调用

前面所有例子用的都是它：

chain = prompt | ch@t | output_parser
result = chain.invoke({"topic": "康师傅绿茶"})

传一个输入，跑完整条 chain，返回一个输出。它的本质，第 1.2 节已经拆过了：RunnableSequence.invoke 拿输入去 for 循环跑 steps，上一节的输出喂给下一节，最后一节的结果作为整条 chain 的返回值。

invoke 是最常用的方法。如果只想拿一个完整结果，用它就够了。

4.2 stream：流式调用

stream 和 invoke 传参一样，区别在返回：invoke 返回一个完整结果，stream 返回一个迭代器，逐块产出。

for chunk in chain.stream({"topic": "康师傅绿茶"}):
    print(chunk, end="")

关键问题是：流式在 chain 里怎么传下去？ 不是只有最后一节在流，而是每一节都支持流式，chunk 一节一节往下传。

输入：{"topic": "康师傅绿茶"}
  ↓ prompt.stream(...)
中间：PromptValue
  ↓ [email protected](...)
中间：AIMessageChunk 一块一块产出
  ↓ output_parser.stream(...)
输出：字符串一块一块产出

stream 的好处是「边说边处理」。模型还没说完，前面的 chunk 就已经流到解析器了，不用等整句生成完。

这一点在解析器上特别明显。JSON 解析器的流式输出长这样：

for s in chain.stream({"query": "请给我介绍学习中国历史的经典书籍"}):
    print(s)

输出：

{}
{'title': ''}
{'title': '中国'}
{'title': '中国通'}
{'title': '中国通史'}
{'title': '中国通史', 'author': ''}
{'title': '中国通史', 'author': '吕'}
{'title': '中国通史', 'author': '吕思'}
{'title': '中国通史', 'author': '吕思勉'}
...

字典不是一次性出现的，而是随着模型一块一块吐字，字段一个一个长出来。这就是流式——解析器不是等模型说完才解析，而是边流边解析。

4.3 RunnableLambda 和 RunnableGenerator 的区别

stream 有一个容易踩的坑：不是链里每一节都能「边来边处理」。 这就要回到 Parser 篇讲过的 RunnableLambda 和 RunnableGenerator。

先看一个场景。模型回答「你好」，它不是一次性吐完，而是分几次产出：

第 1 次，模型产出：你
第 2 次，模型产出：好
第 3 次，模型产出：！

问题来了：你的代码，是每次产出一处理一次，还是等全部产出完再处理一次？这就是两个组件的区别。

RunnableGenerator：每次产出一处理一次

def streaming_parse(chunks):
    for chunk in chunks:              # 上游每产出一块
        yield chunk.content.swapcase()  # 就处理这一块，产出一块

streaming_parse 收到的是整条 chunk 流。模型第 1 次产出「你」，chunks 里立刻多一块，for 循环马上拿到它、处理它、yield 出去，外面立刻看到结果。第 2 次、第 3 次同理。

模型分三次产出，它就分三次处理，外面分三次看到。 这就是逐块处理。

RunnableLambda：攒完再处理一次

def parse(ai_message):
    return ai_message.content.swapcase()

parse 收到的不是「你」「好」「！」三块，而是等模型全部生成完之后、组装好的一个完整 AIMessage，里面是「你好！」。所以 parse 只跑一次、return 一次，外面只看到一次输出。

那三次产出，在这里被攒成了一个整体。模型分三次产出，它只处理一次，外面只看到一次。

区别的后果：流式效果会不会被挡住

「流式」说的是外面最后拿到的是一次结果，还是分次结果。

• 用 RunnableGenerator：模型分三次产出，外面分三次拿到，流式效果保住。
• 用 RunnableLambda：模型分三次产出，但被 parse 攒成一次，外面只拿到一次，流式效果被挡住。

注意：RunnableLambda 不是「不能调 stream」。chain.stream(...) 照样能跑，只是流到它这一节被攒住了，外面看不到「一块一块」的效果。

而且挡住的不是整条 chain，是这一节。如果链里某一节是 RunnableLambda 这种「攒完再处理」，流式效果到它这就断了；换成 RunnableGenerator，流式效果就能一路传到外面。

对比表

	RunnableLambda	RunnableGenerator
处理时机	模型全部生成完再处理	模型每产出一块就处理
输入	AIMessage（完整对象）	AIMessageChunk（碎片流）
写法	普通函数，return	生成器，yield
流式效果	被挡住，外面只看到一次输出	保住，外面分次看到输出
典型调用	invoke()	stream()

小结：自己写的处理逻辑，想保住流式效果，就用 RunnableGenerator 写生成器；只处理完整结果、不在乎流式，用 RunnableLambda 写普通函数。官方 Parser 已经内部实现了 stream，不用你自己操心。

4.4 batch：批量调用

batch 是 invoke 的批量版：传一组输入，返回一组输出。

results = chain.batch([
    {"topic": "康师傅绿茶"},
    {"topic": "可口可乐"},
    {"topic": "农夫山泉"},
])

它和「写个 for 循环挨个 invoke」的区别是：batch 内部会并发处理这批输入，不用一个跑完再跑下一个。输入多的时候，比手动循环快。

4.5 ainvoke：异步调用

ainvoke 是 invoke 的异步版，用在 async 环境里：

result = await chain.ainvoke({"topic": "康师傅绿茶"})

它的行为和 invoke 一样，只是返回协程，配 await 用。

链里最耗时的是模型输出，它是「等 IO」的活——发请求出去，等几秒，结果回来，这段时间 CPU 是闲着的。同步接口 def 用 chain.invoke(...)，等模型时线程被占着；异步接口 async def 用 await chain.ainvoke(...)，等模型时线程能去处理别的请求。注意别在 async def 里用同步 invoke，会卡住事件循环。对应的 stream 有 astream，batch 有 abatch，需要时再查。

小结

四个方法跑的是同一条 chain，只是调用方式不同：

方法	用途	返回
invoke	单次调用	一个完整结果
stream	流式调用	逐块产出的迭代器
batch	批量调用	一组结果，内部并发
ainvoke	异步单次调用	协程，配 await

流式效果会不会被挡住，取决于链里每一节是「边来边处理（逐块处理）」还是「攒完再处理」。RunnableGenerator 边来边处理，流式效果一路传下去；RunnableLambda 攒完再处理，流式效果到它就断了。

5. 完整链：prompt | ch@t | parser

前面四节把零件都讲过了：Runnable 协议、Runnable 家族、三个组合工具、四个调用方法。这一节把它们拼成完整链，看三种典型形态。

5.1 最小链

最基础的链，就是把三件套按顺序串起来：

chain = prompt | ch@t | output_parser
result = chain.invoke({"topic": "康师傅绿茶"})

它的结构，第 1.2 节拆过：

输入：{"topic": "康师傅绿茶"}
  ↓ prompt
中间：PromptValue
  ↓ ch@t
中间：AIMessage
  ↓ output_parser
输出：字符串

这条链是所有链的原型。后面两条，都是在它的某个位置插入了额外处理。

5.2 带输入适配的链

有时候链的第一节要的输入，和你手上有的对不上。比如 prompt 要的是 {"input": ...} 字典，但用户只给你一句裸字符串：

chain = (
    {"input": RunnablePassthrough()}
    | prompt
    | llm
    | json_parser
)

chain.invoke("John is 20 years old. He is a student...")

{"input": RunnablePassthrough()} 这一步，就是第 3.2 节讲的输入适配：RunnablePassthrough() 把裸字符串原样返回，外层的 {"input": ...} 给它套上键名，包成 prompt 要的字典。

输入："John is 20 years old..."
  ↓ {"input": RunnablePassthrough()}
中间：{"input": "John is 20 years old..."}
  ↓ prompt
中间：PromptValue
  ↓ llm
中间：AIMessage
  ↓ json_parser
输出：{'name': 'John', 'age': 20, 'description': '...'}

这条链的特点：第一节不是 prompt，而是一个字典构造，用来把输入变成 prompt 要的形状。

5.3 带并行组装的链

如果 prompt 本身由多个零件拼成，就用 RunnableParallel 并行渲染：

composed = RunnableParallel(
    introduction=introduction_prompt,
    example=example_prompt,
    start=start_prompt,
)

def format_and_strip_text(inputs):
    intro_str = inputs['introduction'].text
    example_str = inputs['example'].text
    start_str = inputs['start'].text
    return f"{intro_str}nn{example_str}nn{start_str}".strip()

full_prompt = RunnableLambda(format_and_strip_text)

chain = composed | full_prompt | ch@t | output_parser

这条链的结构：

输入：{"person": ..., "example_q": ..., "example_a": ..., "input": ...}
  ↓ RunnableParallel（并行渲染三个 Prompt）
中间：{"introduction": PromptValue, "example": PromptValue, "start": PromptValue}
  ↓ RunnableLambda（取 .text 拼接）
中间：完整提示词字符串
  ↓ ch@t
中间：AIMessage
  ↓ output_parser
输出：字符串

它比 5.1 多了前半段：用 RunnableParallel 分叉渲染，用 RunnableLambda 合并。 这就是 1.3 里 PipelinePromptTemplate 的 LCEL 替代方案，也是第 3.3 节的完整形态。

5.4 format_messages vs invoke

前面一直留着一个对比，这里收掉。

ChatModel 除了 invoke，还有一个方法叫 format_messages。两者的分工：

• format_messages：把输入转成模型要的消息格式，只做格式转换，不调模型。
• invoke：把输入转成消息格式，再调模型，返回 AIMessage。

看代码最清楚：

# 只格式化，不调模型
messages = [email protected]_messages("请问2只兔子有多少条腿？")
# messages 是 [HumanMessage(content='请问2只兔子有多少条腿？')]

# 格式化 + 调模型
ai_message = [email protected]("请问2只兔子有多少条腿？")
# ai_message 是 AIMessage(content='2只兔子有8条腿。')

format_messages 的用途是调试：你想看看「输入经过 ChatModel 会变成什么消息」，又不想真的调模型花钱，就用它。日常开发用 invoke 就行。

小结

Runnable 家族

类别	组件	作用
核心三件套	Prompt	组装提示词
	ChatModel	对话模型
	Parser	解析模型输出
组合类	RunnableParallel	并行跑多个分支，合并结果
	RunnablePassthrough	输入原样透传，适配输入形状
自定义类	RunnableLambda	把普通函数接进链，攒完处理
	RunnableGenerator	把生成器接进链，逐块处理

四个调用方法

方法	用途	返回
invoke	单次调用	一个完整结果
stream	流式调用	逐块产出的迭代器
batch	批量调用	一组结果，内部并发
ainvoke	异步单次调用	协程，配 await

三种典型链

链	特点
prompt | ch@t | output_parser	最小链，所有链的原型
{"input": RunnablePassthrough()} | prompt | llm | parser	第一节做输入适配
composed | full_prompt | ch@t | parser	前半段并行渲染 + 合并

一条主线：LCEL 用 | 把 Runnable 串成链。Runnable 是统一协议，符合协议的组件都能进链；链自己也是 Runnable，能继续拼。所谓「串链」，就是把三件套、组合工具、自定义处理，按输入输出能对上的顺序接起来，然后用 invoke / stream / batch / ainvoke 中的一种去跑。

6. 出错了怎么办

链跑起来，出错的地方通常不在 prompt，也不在 parser 的本地处理，而在模型输出和模型输出能不能被解析这两处。这一节讲三层处理：某一节内部纠错、某一节失败后换路、偶发失败重试。

6.1 解析失败：OutputFixingParser 在链里的位置

Parser 篇讲过 OutputFixingParser：外层包一个常规解析器 + 一个 llm，解析失败时让模型重新生成。这里看它在链里的位置。

parser = PydanticOutputParser(pydantic_object=Actor)
new_parser = OutputFixingParser.from_llm(parser=parser, llm=ch@t)

chain = prompt | ch@t | new_parser

结构上，它就在链的最后一环：

prompt | ch@t | new_parser
                  ↑
            纠错发生在这里

执行过程：

输入：{...}
  ↓ prompt
中间：PromptValue
  ↓ ch@t
中间：AIMessage（格式可能不对）
  ↓ new_parser
      ├─ 先用内部 parser 解析
      │    ├─ 成功 → 返回结构化结果
      │    └─ 失败 → 把坏输出 + 格式要求交给内部 llm，重新生成
      │             再解析一次 → 返回结构化结果
输出：结构化结果

从链的角度看：纠错发生在最后一节内部，链本身没中断。 new_parser 对外仍然是一个 Runnable，输入 AIMessage，输出结构化结果。它内部多做了一次「失败就重来」，外面感知不到。

这就是「某一节内部纠错」这一层。

6.2 某一节失败：with_fallbacks 换路

OutputFixingParser 解决的是「解析格式不对」，但如果某一节彻底失败——比如模型服务挂了、超时了——就需要换一条路。

Runnable 协议提供 with_fallbacks：给某一节配一个备选，主路失败时自动走备选。

chain = prompt | ch@t | parser
fallback_chain = prompt | ch@t | StrOutputParser()

robust_chain = chain.with_fallbacks([fallback_chain])

robust_chain 先跑 chain。如果 chain 里任何一节抛异常，就自动改跑 fallback_chain。这里 fallback_chain 用 StrOutputParser，不要求结构化输出，至少能返回一个字符串，不至于整个请求失败。

注意，with_fallbacks 不是「拿失败那一步的中间结果去补救」，而是整条链作废、换备选链从头跑。备选链会重新执行自己的每一节，包括重新调模型。

with_fallbacks 可以配多个备选，按顺序尝试，前一个失败就试下一个。

这是「某一节失败就换路」这一层。

6.3 偶发失败：with_retry 重试

有时候失败是偶发的——网络抖一下、模型服务瞬时 500。这种不需要换路，重试一次可能就好了。

Runnable 协议还提供 with_retry：

chain = prompt | [email protected]_retry(stop_after_attempt=3) | parser

ch@t 这一节失败时自动重试，最多 3 次。和接口超时自动重试是一个思路——不换路，原地再来一次。本篇不展开，知道有这能力即可。

6.4 出错处理的层次

出错情况	处理方式	作用范围
解析格式不对	OutputFixingParser	解析器内部纠错，链不中断
某一节彻底失败	with_fallbacks	换一条备选链
偶发失败	with_retry	同一节自动重试

三层从内到外：解析器内部纠错 → 整节换路 → 同节重试。 实际用的时候按需组合，不是每一层都必须上。

小结

链里最容易出错的是模型输出和解析这两步。处理思路分三层：格式问题在解析器内部修，整节失败用 with_fallbacks 换路，偶发失败用 with_retry 重试。三者都是 Runnable 协议自带的能力，配在链的对应位置上即可。

7 小结

这一篇讲的是 LCEL——用 | 把组件串成链。

| 是什么。 | 要求两边都是 Runnable。Runnable 是一份协议，规定实现 invoke / stream / batch / ainvoke 四个方法。Prompt、ChatModel、Parser 都符合这份协议，所以能串。| 底层是 Runnable.__or__，返回 RunnableSequence——外表一条链，内里一个 steps 列表，invoke 就是 for 循环跑这个列表，上一节的输出喂给下一节。链的输入由第一节决定，输出由最后一节决定。

能串什么。 核心三件套之外，还有三个组合工具：RunnableLambda 把普通函数接进链，RunnablePassthrough 做输入适配，RunnableParallel 并行跑多个分支再合并。链自己也是 Runnable，能继续拼下去。

怎么调用。 四个方法跑的是同一条链：invoke 单次、stream 流式、batch 批量、ainvoke 异步。流式效果会不会被挡住，取决于每一节是「边来边处理」还是「攒完再处理」——RunnableGenerator 逐块处理，流式效果一路传下去；RunnableLambda 攒完处理，流式效果到它就断了。异步接口用 ainvoke，别在 async def 里用同步 invoke。

典型链。 最小链 prompt | ch@t | output_parser 是所有链的原型。带输入适配的链在最前面加 {"input": RunnablePassthrough()}。带并行组装的链用 RunnableParallel 分叉渲染、RunnableLambda 合并。

出错怎么办。 三层：格式不对用 OutputFixingParser 在解析器内部现场修；某一节彻底失败用 with_fallbacks 换链重跑；偶发失败用 with_retry 原地重试。

一句话：LCEL 把「组件」和「组合」分开——组件各管各的，组合交给 |。只要符合 Runnable 协议，就能进链；串出来的链自己也是 Runnable，能继续拼。这就是它比 PipelinePromptTemplate 那类专用类更通用的原因。