LangChain教程 - 9 LangChain的Chain链

9.1 Chain简介

1 什么是LangChain的Chain

一句话解释：Chain（链）就是把多个组件按顺序连接起来，形成一个自动执行流程。

可以理解成：

输入 → 组件1 → 组件2 → 组件3 → 输出

每个组件都实现了 Runnable 接口，在执行的时候，前一个组件的输出作为下一个组件的输入。

2 为什么需要Chain

如果没有 Chain，我们在使用模板和调用模型的时候，使用如下代码：

python

prompt = chat_prompt_template.invoke({...})
res = model.stream(prompt)

我们需要手动生成提示词，手动传给模型，并自己管理流程。

如果流程变复杂时，代码会变的复杂，越来越乱：

生成提示词 → 调模型 → 处理输出 → 再调模型 → 再处理

如果使用 Chain，我们可以将多个组件使用 | 将多个组件进行相连，可以通过 Chain 的 invoke() 或 stream() 方法触发整个链条的执行。

当调用 chain.invoke() 方法，链条里所有组件依次会被调用 invoke() 方法，一次性返回完整结果；
当调用 chain.stream() 方法，只有大模型组件被调用 stream() （流式输出），其他组件（Prompt/Parser 等）仍调用 invoke()；
当调用 chain.stream() 时，如果链条中的组件支持流式输出，则会调用其 stream() 方法；如果组件不支持流式输出，则会自动使用 invoke() 方法执行。在实际使用中，通常只有大模型组件支持流式输出，因此大多数情况下会触发调用大模型的 stream() 方法，调用其他组件的 invoke() 方法。

举个栗子：

python

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_ollama import ChatOllama

# 1. 创建聊天提示词模板，在模板中使用MessagesPlaceholder占位符，后面传入数据，生成最终的提示词
chat_prompt_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一位专业的健身教练，回答要简单实用。"),
        MessagesPlaceholder("history"),
        ("human", "{input}")
    ]
)

# 2. 准备历史对话数据
history_data = [
    ("human", "我想减肥，有什么建议？"),
    ("ai", "建议控制饮食，并配合每周3次有氧运动。"),
    ("human", "我平时中午吃米饭可以吗？"),
    ("ai", "可以，但建议减少分量，多搭配蔬菜和蛋白质。")
]

# 3. 创建聊天模型
model = ChatOllama(model="qwen3:1.7b")

# 4. 组成链，每个组件都是Runnable接口的子类
chain = chat_prompt_template | model

# 调用模型，插入提示词
res = chain.stream({"history": history_data, "input": "那我晚饭应该怎么吃？"})
for chunk in res:
    print(chunk.content, end="", flush=True)

chain = chat_prompt_template | model 形成的链本质上是一个 RunnableSequence（执行链），它内部维护了一个有序组件列表，也是 RunnableSerializable 对象的实现（RunnableSequence是RunnableSerializable的子类），都是 Runnable 接口的子类对象。
当调用 chain.stream() 时，执行链会按照组件顺序依次运行：首先把输入参数 {"history": history_data, "input": "那我晚饭应该怎么吃？"} 交给第一个组件 ChatPromptTemplate，生成完整的对话消息对象；然后将 chat_prompt_template 的执行结果自动传递给下一个组件 model 作为输入；
由于使用的是 stream() 方法，模型在生成回答时会持续产出内容片段（chunk），执行链不会等待全部完成，而是边生成边向外输出。
我们现在只连接了两个组件，后面我们会学习多个组件，进行多个组件的链接，每个组件实现 Runnable 接口就可以链接。

9.2 链接运算符

执行链的组件为什么可以使用 | 运算符进行连接？

这是因为执行链的各个组件重写了 | 运算符。

在 Python 中，运算符的行为由类的魔方方法决定的，例如：

python

a + b

对象 a + 对象 b，本质调用的是 a.__add__(b) 方法。

同样：

python

a | b

对象 a | 对象 b，本质调用的是 a.__or__(b) 方法。

在前面说了组件必须实现了 Runnable 接口，是因为在 Runnable 基类内部实现了：

python

def __or__(self, other):
    return RunnableSequence(self, other)

所以当组件对象使用 | 链接，得到的是 RunnableSequence 对象，RunnableSequence 对象也是 Runnable 接口的实现，所以依然支持使用 | 链接，所以后面无论怎么链接，得到的都是 RunnableSequence 对象，这也是 Runnable 接口可以链接操作的原因。

9.3 字符串输出解析器

如果我们想让第一次模型输出的结果，重新输入给模型，那么按照链的操作，应该如下：

python

from langchain_core.prompts import PromptTemplate
from langchain_ollama import ChatOllama
from langchain_core.messages import AIMessage

# 1. 创建提示词模板
prompt_template = PromptTemplate.from_template("我的邻居姓{lastname}，刚生了一个{gender}，帮忙取个名字，简单回答")

# 2. 创建模型
model = ChatOllama(model="qwen3:1.7b")

# 3. 创建执行链，将模型的输出结果重新交给模型
chain = prompt_template | model | model

# 调用模型，通过 invoke 方法提问
response: AIMessage = chain.invoke({"lastname":"李", "gender":"儿子"})
print(response.content)

通过 prompt_template | model | model 继续在后面添加 | model ，将输出交给模型。

但是运行会报错：

python

ValueError: Invalid input type <class 'langchain_core.messages.ai.AIMessage'>. Must be a PromptValue, str, or list of BaseMessages.

这是因为模型输出的结果数据类型为 AIMessage 类型，将该类型重新输入给模型是不行的，因为模型的输入需要是 PromptValue 、str 或 BaseMessages 列表类型。

所以需要转换一下，将 AIMessage 转换为模型可以接收的数据类型，我们可以使用字符串输出解析器 StrOutputParser ， StrOutputParser 是 LangChain 内置的字符串解析器，它可以将 AIMessage 解析为字符串，而且它是 Runnable 接口的子类，可以加入执行链。

我们只需要将模型的输出结果输入给 StrOutputParser ， StrOutputParser 可以将 AIMessage 中的内容提取出来解析为字符串，然后再将字符串输入给模型，所以链条如下：

python

from langchain_core.output_parsers import StrOutputParser

# 创建字符串解析器
parser = StrOutputParser()

# 重新交给模型
chain = prompt_template | model | parser | model

# 调用模型，通过 invoke 方法提问
response: AIMessage = chain.invoke({"lastname":"李", "gender":"儿子"})
print(response.content)

所以只需要在两个模型中间添加一个转换器就可以了，将模型输出转换为字符串，字符串可以作为模型输入。

我们还可以在 chain 链的最后，添加一下 StrOutputParser ，这样可以直接获取模型输出结果中的 content 字符串，可以打印出来：

python

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_ollama import ChatOllama

# 创建提示词模板
prompt_template = PromptTemplate.from_template("我的邻居姓{lastname}，刚生了一个{gender}，帮忙取个名字，简单回答")

# 创建模型
model = ChatOllama(model="qwen3:1.7b")

# 创建字符串解析器
parser = StrOutputParser()

# 重新交给模型
chain = prompt_template | model | parser | model | parser

# 调用模型，结果是字符串类型
response: str = chain.invoke({"lastname":"李", "gender":"儿子"})
print(response)

最后又添加了一个 StrOutputParser，得到的结果就是字符串了。

9.4 JsonOutputParser

上面在使用字符串输出解析器的时候，将模型输出的数据重新传递给模型，这样其实是有点问题的。

因为第一次调用大模型，大模型给起了一个名字，然后将名字重新传递给大模型，大模型也懵逼了，你想要做什么呢？

所以我们第一次从大模型得到结果后，应该是做一些处理，再传递给大模型：

python

提示词模板 → 模型 → [数据处理] → [提示词模板] → 模型 → 解析器 → 结果

应该将第一次模型的输出结果进行处理，然后传递给提示词模板，生成新的提示词，然后再传递给模型，这样第二次调用模型，模型才能知道你要做什么。
第一次模型输出的结果是 AIMessage 对象，而提示词模板的输入是字典类型，所以拿到第一次模型输出结果，需要进行数据处理，将 AIMessage 对象转换为字典类型，传递给提示词模板，作为第二次调用模型的输入。

此时使用 StrOutputParser 就做不到了，我们需要使用 JsonOutputParser 将 AIMessage 转换为字典对象。

但是首先需要注意， JsonOutputParser 是将 Json 字符串传唤为字典，所以我们需要在第一次调用模型的时候，告诉模型，输出的结果需要是 JSON 格式的，然后才能调用 JsonOutputParser 。

python

from langchain_core.prompts import PromptTemplate
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser

# 第一次生成名字，并要求返回 JSON
first_prompt_template = PromptTemplate.from_template(
    """
    我的邻居姓{lastname}，刚生了一个{gender}，
    请帮忙取一个名字，并按照如下 JSON 格式返回：

    {{
        "name": "名字"
    }}

    只返回 JSON，不要包含任何额外文字。
    """
)

# 创建模型
model = ChatOllama(model="qwen3:1.7b")

# 第二个提示词
second_prompt_template = PromptTemplate.from_template(
    "请详细解释名字 {name} 的寓意，字数控制在 100 字以内"
)

# 创建 JSON 解析器
json_parser = JsonOutputParser()
# 创建字符串解析器
str_parser = StrOutputParser()

# 构建链条
chain = first_prompt_template | model | json_parser | second_prompt_template | model | str_parser

# 调用模型，结果是字符串类型
response = chain.stream({"lastname": "王", "gender": "儿子"})
for chunk in response:
    print(chunk, end="", flush=True)

首先在上面的例子中，在模板中，JSON 格式需要是格式的，因为在 PromptTemplate 中，单个 {} 是变量占位符，如果想写普通 JSON 的 {}，必须用进行转义。

python

chain = first_prompt_template | model | json_parser | second_prompt_template | model | str_parser

上面整个链条，首先通过 chain.stream() 调用，将参数传递给第一个模板，第一个模板输出的结果传递给第一次模型调用，得到的结果是 JSON 格式的，然后传递给 JsonOutputParser，然后经过 JsonOutputParser 解析得到字典，然后作为参数传递给第二个模板，第二个模板输出作为第二次调用模型的输入，最终将第二次模型的输出传递给 StrOutputParser，从 AIMessage 解析 content 属性为字符串输出。

9.5 RunnableLambda

什么是 RunnableLambda？

RunnableLambda 是 LangChain 内置的类，可以将普通的 Python 函数，转换成 Runnable 接口实例。

这样就可以将自定义的普通 Python 函数可以加入到链条中，实现自定义的数据转换逻辑。

下面使用 RunnableLambda 实现上面 JsonOutputParser 实现的功能，我们不需要让模型的第一次输出为 JSON 格式了，只需要输出一个名字就可以了，然后我们自己实现一个函数，让函数返回字典就可以了。

所以需要两个步骤：

首先定义一个转换的函数，函数返回字典数据类型；
然后使用 RunnableLambda 将函数转换为 Runnable 实例即可，然后就可以加入到执行链了。

python

from langchain_core.prompts import PromptTemplate
from langchain_ollama import ChatOllama
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda
from langchain_core.messages import AIMessage

# 第一次提示词模板，只要求模型输出名字
first_prompt_template = PromptTemplate.from_template(
    "我的邻居姓{lastname}，刚生了一个{gender}，帮忙取个名字，简单回答，只给名字"
)

# 第二个提示词模板
second_prompt_template = PromptTemplate.from_template(
    "请详细解释名字 {name} 的寓意，字数控制在 100 字以内"
)

# 创建模型
model = ChatOllama(model="qwen3:1.7b")

# 字符串解析器
str_parser = StrOutputParser()


# 自定义函数：将名字转换为字典
def convert_to_dict(message: AIMessage) -> dict:
    return {
        "name": message.content
    }


# 将普通函数转换为 Runnable
lambda_runnable = RunnableLambda(convert_to_dict)

# 构建链条
chain = first_prompt_template | model | lambda_runnable | second_prompt_template | model | str_parser

# 调用
response = chain.stream({"lastname": "王", "gender": "儿子"})
for chunk in response:
    print(chunk, end="", flush=True)

通过上面两个步骤，就相当于使用自定义函数替换了 JsonOutputParser，代码更清晰了，而且不需要模型按照 JSON 格式返回数据。

其实可以直接将普通函数加入到执行链的：

python

chain = first_prompt_template | model | convert_to_dict | second_prompt_template | model | str_parser

因为直接将函数加入到执行链，函数会自动转换为 RunnableLambda 对象，但是还是推荐使用 RunnableLambda 显示转换一下。

LangChain教程 - 9 LangChain的Chain链 ​

9.1 Chain简介 ​

1 什么是LangChain的Chain ​

2 为什么需要Chain ​

9.2 链接运算符 ​

9.3 字符串输出解析器 ​

9.4 JsonOutputParser ​

9.5 RunnableLambda ​

LangChain教程 - 9 LangChain的Chain链

9.1 Chain简介

1 什么是LangChain的Chain

2 为什么需要Chain

9.2 链接运算符

9.3 字符串输出解析器

9.4 JsonOutputParser

9.5 RunnableLambda