前言

在前两篇文章中，我们已经完成了两条完整的大模型处理流程：先是使用 FewShotChatMessagePromptTemplate 让模型模仿示例格式输出，再使用 Output Parser 将模型返回的文本解析成 Python 列表或结构化对象。

当时的代码采用分步调用的方式：

final_prompt = prompt_template.invoke(input_data)
response = model.invoke(final_prompt)
result = output_parser.invoke(response)

这种写法非常适合初学阶段，因为每一步都可以单独打印查看。但当处理流程逐渐变长时，我们会希望把多个组件连接成一个可以反复调用的整体。LangChain 中的 Chain 正是用来解决这个问题的。

本文基于课堂 Notebook 06 Chain.ipynb，使用一个非常直观的案例：

• 接收国家名称
• 生成包含格式要求的提示词
• 调用通义千问模型
• 把模型文本解析成 Python 列表

我们将分别使用嵌套调用和 Chain（管道符） 两种写法，最终得到相同的结果，并深入理解 Chain 的工作原理与优势。

本文最终效果：无论使用嵌套调用还是 Chain，都能得到类似 ['比亚迪', '吉利', '长城', '奇瑞', '长安'] 的 Python 品牌列表。

项目效果展示

嵌套调用结果： ['比亚迪', '吉利', '长城', '奇瑞', '长安']
Chain 调用结果： ['比亚迪', '吉利', '长城', '奇瑞', '长安']

两种写法输出完全一致，说明 Chain 只是用更清晰的方式组织了同样的三个步骤。

完整工程代码

以下为完整可运行代码，可直接复制到 Python 文件或 Notebook 中执行：

import os

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


# 1. 创建提示词模板
prompt_template = ChatPromptTemplate.from_messages([
    ("system", "{parser_instructions}"),
    ("human", "列出5个{subject}生产的汽车的品牌。")
])

# 2. 创建列表解析器
output_parser = CommaSeparatedListOutputParser()
parser_instructions = output_parser.get_format_instructions()

# 3. 创建模型
model = ChatOpenAI(
    model="qwen-plus",
    openai_api_key=os.getenv("DASHSCOPE_API_KEY"),
    openai_api_base=os.getenv(
        "DASHSCOPE_BASE_URL",
        "https://dashscope.aliyuncs.com/compatible-mode/v1"
    )
)

# 准备输入数据
input_data = {
    "subject": "中国",
    "parser_instructions": parser_instructions
}

# ========== 写法一：嵌套调用 ==========
print("=" * 30)
print("嵌套调用")
result_nested = output_parser.invoke(
    model.invoke(
        prompt_template.invoke(input_data)
    )
)
print("结果：", result_nested)

# ========== 写法二：Chain（管道符） ==========
print("=" * 30)
print("Chain 调用")
chat_model_chain = prompt_template | model | output_parser
result_chain = chat_model_chain.invoke(input_data)
print("结果：", result_chain)

预期输出（模型具有一定随机性，品牌名称或顺序可能略有变化）：

==============================
嵌套调用
结果： ['比亚迪', '吉利', '长城', '奇瑞', '长安']
==============================
Chain 调用
结果： ['比亚迪', '吉利', '长城', '奇瑞', '长安']

技术原理：什么是 LangChain Chain

1. 从数据流的角度理解 Chain

在 Output Parser 的学习中，我们已经熟悉了三个核心组件：

组件	课堂对象	作用
Prompt	prompt_template	将输入变量填入消息模板
Model	model	将提示词发送给大模型
Parser	output_parser	将模型文本转换成 Python 列表

这三个对象虽然用途不同，却有一个共同点：都可以通过 invoke() 接收输入并产生输出。

在 LangChain 中，这类可调用、可组合的组件被称为 Runnable。

可以把 Runnable 理解成一个标准加工环节：

接收一种输入 → 完成自己的工作 → 产生下一种输出

只要前一个组件的输出能够作为后一个组件的输入，它们就可以串联起来。

图2 数据流与三种写法对比（建议放置：展示分步调用、嵌套调用、Chain 调用三种写法的并列对比图）

2. 管道符 | 在 LangChain 中的含义

在普通 Python 表达式中，| 常用于集合或字典合并。在 LangChain 中，相关对象重载了这个运算符，使它表示组件连接。

chat_model_chain = prompt_template | model | output_parser

可以读成：

先执行 prompt_template，然后把结果交给 model，最后把模型结果交给 output_parser。

它和命令行中的管道思想非常相似：前一步的输出自动成为后一步的输入。

代码逐模块详细解析

模块一：导入所需组件

功能说明

导入模型客户端、列表解析器和提示词模板。

完整代码

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

核心实现逻辑

三个组件分别负责三个阶段：

• ChatPromptTemplate：组织提示词，将输入变量填入模板
• ChatOpenAI：通过 OpenAI 兼容接口调用通义千问
• CommaSeparatedListOutputParser：将模型返回的逗号分隔文本解析为 Python 列表

这三个组件都有一个共同点——它们都实现了 invoke() 方法，因此可以被串联起来。

模块二：创建提示词模板

功能说明

定义消息模板，包含系统消息（格式要求）和人类消息（具体问题）。

完整代码

prompt_template = ChatPromptTemplate.from_messages([
    ("system", "{parser_instructions}"),
    ("human", "列出5个{subject}生产的汽车的品牌。")
])

模板变量说明

变量名	来源	作用
{parser_instructions}	output_parser.get_format_instructions()	告诉模型返回逗号分隔的列表
{subject}	调用时传入	指定国家名称

调用时传入 subject="中国"，人类消息就会变成：

列出5个中国生产的汽车的品牌。

模块三：创建列表输出解析器

功能说明

创建解析器实例，并获取格式说明文本。

完整代码

output_parser = CommaSeparatedListOutputParser()
parser_instructions = output_parser.get_format_instructions()

格式说明内容

Your response should be a list of comma separated values,
eg: `foo, bar, baz` or `foo,bar,baz`

这段说明随后会被填入模板中的 {parser_instructions}，告诉模型应该如何组织回答。

模块四：创建模型

功能说明

初始化 Qwen 模型客户端。

完整代码

import os

model = ChatOpenAI(
    model="qwen-plus",
    openai_api_key=os.getenv("DASHSCOPE_API_KEY"),
    openai_api_base=os.getenv(
        "DASHSCOPE_BASE_URL",
        "https://dashscope.aliyuncs.com/compatible-mode/v1"
    )
)

参数说明

参数	类型	作用
model	str	指定模型名称，此处为 qwen-plus
openai_api_key	str	从环境变量 DASHSCOPE_API_KEY 读取密钥
openai_api_base	str	API 端点地址，优先使用环境变量，否则使用公共兼容地址

调试提示：如果密钥读取不到，检查环境变量是否设置正确。可以通过 print(os.getenv("DASHSCOPE_API_KEY") is not None) 验证，但不要打印完整密钥。

模块五：嵌套调用

完整代码

result = output_parser.invoke(
    model.invoke(
        prompt_template.invoke({
            "subject": "中国",
            "parser_instructions": parser_instructions
        })
    )
)

执行顺序详解

这段代码从最内层开始执行：

第一步：执行最里面的提示词模板

prompt_template.invoke({
    "subject": "中国",
    "parser_instructions": parser_instructions
})

接收一个字典，输出填充完成的聊天消息（ChatPromptValue）。

第二步：把消息传给模型

model.invoke(...)

模型返回一个 AIMessage，其中包含类似下面的文本：

比亚迪, 吉利, 长城, 奇瑞, 长安

第三步：把模型消息交给输出解析器

output_parser.invoke(...)

解析器按照逗号拆分内容，最终得到 Python 列表：

['比亚迪', '吉利', '长城', '奇瑞', '长安']

注意：代码虽然从左侧看到的是 output_parser，但 Python 会先执行最内层的 prompt_template.invoke()，再逐层向外。

模块六：使用管道符创建 Chain

完整代码

chat_model_chain = prompt_template | model | output_parser
result = chat_model_chain.invoke({
    "subject": "中国",
    "parser_instructions": parser_instructions
})

核心实现逻辑

prompt_template | model | output_parser 创建了一条由三个组件串联而成的 Chain。

调用 chain.invoke(input_data) 时：

1. 输入字典首先进入 prompt_template，填充模板变量，输出 ChatPromptValue
2. ChatPromptValue 进入 model，调用 API，输出 AIMessage
3. AIMessage 进入 output_parser，解析为 Python 列表

整个过程中，数据自动从前一个组件流向下一个组件，调用者只需要提供最初的输入变量即可。

嵌套调用 vs Chain：对比与选择

功能对比

对比项	嵌套调用	Chain 调用
执行顺序	从内到外，不符合阅读习惯	从左到右，符合阅读习惯
代码可读性	括号嵌套多，可读性差	管道清晰，易于理解
可复用性	每次需要重新编写	可命名后反复调用
扩展性	难以增加中间步骤	可方便地插入新组件
调试友好	需拆开才能打印中间结果	同样需拆开调试
逻辑等价性	✅ 完全等价	✅ 完全等价

何时使用 Chain

推荐使用 Chain 的场景：

• 流程已经调试稳定
• 需要多次复用同一流程
• 代码需要保持简洁和可维护性
• 未来可能需要扩展（分支、并行、重试）

适合拆开调用的场景：

• 第一次编写流程，需要观察每个阶段
• 模型输出不稳定，需要检查原始回答
• 解析器报错，需要定位问题位置
• 教学时需要理解数据类型变化

实用学习顺序

先拆开运行，确认每一步
          ↓
再组合成 Chain
          ↓
最后封装成可复用功能

使用 partial 简化重复参数

问题：每次都传 parser_instructions

课堂调用时每次都传入：

{
    "subject": "中国",
    "parser_instructions": parser_instructions
}

其中 subject 会变化，但 parser_instructions 通常是固定的。每次都传入显得有些冗余。

解决方案：使用 partial 固定格式说明

prompt_template = ChatPromptTemplate.from_messages([
    ("system", "{parser_instructions}"),
    ("human", "列出5个{subject}生产的汽车的品牌。")
]).partial(parser_instructions=parser_instructions)

然后创建 Chain：

chain = prompt_template | model | output_parser

调用时只需要提供真正会变化的变量：

result = chain.invoke({"subject": "中国"})

这样既可以减少重复参数，也能避免某次调用时忘记传入格式说明。

Chain 的更多调用方式

1. 使用 batch 批量处理

当需要查询多个国家时，可以使用 batch()：

inputs = [
    {"subject": "中国", "parser_instructions": parser_instructions},
    {"subject": "日本", "parser_instructions": parser_instructions},
    {"subject": "德国", "parser_instructions": parser_instructions},
]

results = chat_model_chain.batch(inputs)

for country, brands in zip(["中国", "日本", "德国"], results):
    print(f"{country}: {brands}")

2. 使用 ainvoke 异步调用

在异步程序中，可以使用 ainvoke()：

import asyncio

async def main():
    result = await chat_model_chain.ainvoke({
        "subject": "中国",
        "parser_instructions": parser_instructions
    })
    print(result)

asyncio.run(main())

ainvoke() 不会用同步方式一直占用当前任务，适合异步 Web 服务或需要同时处理多个请求的场景。

常见问题与排查方法

问题1：提示缺少 parser_instructions

现象：

KeyError: 'parser_instructions'

原因：模板中包含了 {parser_instructions}，但调用时没有传入。

解决方法：

• 方法一：每次调用时同时传入两个变量
• 方法二：使用 partial() 提前固定 parser_instructions

问题2：解析器没有得到逗号分隔结果

现象：

OutputParserException: Could not parse output

原因：模型输出了序号列表或包含额外解释：

1. 比亚迪
2. 吉利
3. 长城

解决方法：

• 检查 parser_instructions 是否确实传入了系统消息
• 在提示词中增加更明确的格式要求

("system", "{parser_instructions} 只返回品牌名称，严格使用英文逗号分隔，不要添加序号和解释。")

问题3：API Key 读取不到

现象：

AuthenticationError: Invalid API key

原因：环境变量 DASHSCOPE_API_KEY 未设置或读取失败。

解决方法：

验证环境变量是否存在：

print(os.getenv("DASHSCOPE_API_KEY") is not None)

如果输出 False，检查环境变量是否正确设置，并重启终端或 IDE。

问题4：老师的专属地址为什么不能照搬

课堂中使用的 API 地址是业务空间专属域名：

https://ws-qgmmlec1ui8j56ah.cn-beijing.maas.aliyuncs.com/compatible-mode/v1

这个地址与特定账号或空间相关，不应直接复制使用。自己的程序应使用：

• 阿里云百炼公共 OpenAI 兼容地址；或
• 自己控制台显示的业务空间专属地址

文章和公开仓库中最好通过 DASHSCOPE_BASE_URL 环境变量读取，避免暴露个人空间信息。

问题5：如何确定错误发生在哪一段

Chain 运行失败时，可以暂时拆开定位：

input_data = {
    "subject": "中国",
    "parser_instructions": parser_instructions
}

# 第一步：检查提示词
final_prompt = prompt_template.invoke(input_data)
print("提示词：", final_prompt)

# 第二步：检查模型回答
response = model.invoke(final_prompt)
print("模型回答：", response.content)

# 第三步：检查解析
result = output_parser.invoke(response)
print("解析结果：", result)

先定位问题属于模板、模型还是解析器，再恢复 Chain 写法。

从 Chain 看 LangChain 的设计思想

每个组件只负责一件事

在课堂案例中：

• PromptTemplate 不负责调用模型
• ChatModel 不负责解析文本
• OutputParser 不负责组织用户问题

每个组件只完成自己的职责，然后通过统一接口（invoke）连接起来。

这样做的好处是容易替换。例如，不改变 Prompt 和 Parser，可以更换模型；不改变 Prompt 和 Model，可以换成 JSON 解析器。

复杂应用也是从简单 Chain 开始

后续学习的 RAG、Agent 和工具调用看起来更复杂，但基本思路仍然是数据在多个组件之间流动：

RAG：问题 → 检索文档 → 组织上下文 → 模型 → 答案

Agent：用户问题 → 模型判断 → 调用工具 → 返回结果 → 模型总结

结构化抽取：原始文本 → Prompt → 模型 → Parser → 业务对象

本节课的三个组件虽然简单，却已经展示了 LangChain 最核心的组合思想。

总结

本文通过一个“列出国家汽车品牌”的简单案例，完整展示了 LangChain Chain 的两种实现方式：

嵌套调用：

result = output_parser.invoke(
    model.invoke(
        prompt_template.invoke(input_data)
    )
)

Chain 调用：

chain = prompt_template | model | output_parser
result = chain.invoke(input_data)

两种写法逻辑完全等价，都得到了相同的品牌列表。

本节课最重要的知识点可以概括为：

Runnable 让不同组件拥有统一调用方式
             ↓
管道符让前一个组件的输出进入后一个组件
             ↓
Chain 把多个步骤封装成可重复调用的整体

学习建议：初学时先把每一步拆开运行，理解输入和输出类型；确认流程正确后，再组合成 Chain。这样既能看懂底层数据流，也能写出简洁、可维护的 LangChain 程序。

参考资料

• LangChain Expression Language (LCEL)
• LangChain Runnable Interface
• LangChain Prompt Templates
• LangChain Output Parsers

版权说明

本文为原创技术文章，转载请注明出处。

文中代码均经过实际测试，可根据项目需求进行修改与扩展。

如文章存在疏漏或错误，欢迎在评论区交流讨论。