Chain

Chain 的基本使用

Chain 的基本概念

Chain：链，用于将多个组件（提示模板、LLM 模型、记忆、工具等）连接起来，形成可复用的工作流，完成复杂的任务。

Chain 的核心思想是通过组合不同的模块化单元，实现比单一组件更强大的功能。比如：

• 将 LLM 与 Prompt Template（提示模板）结合
• 将 LLM 与 输出解析器 结合
• 将 LLM 与 外部数据 结合，例如用于问答
• 将 LLM 与 长期记忆 结合，例如用于聊天历史记录
• 通过将 第一个 LLM 的输出作为 第二个 LLM 的输入，……，将多个 LLM 按顺序结合在一起

LCEL 及其基本构成

使用 LCEL，可以构造出结构最简单的 Chain。

LangChain 表达式语言（LCEL，LangChain Expression Language）是一种声明式方法，可以轻松地将多个组件链接成 AI 工作流。它通过 Python 原生操作符（如管道符 |）将组件连接成可执行流程，显著简化了 AI 应用的开发。

LCEL 的基本构成：提示（Prompt）+ 模型（Model）+ 输出解析器（OutputParser），即：

# 在这个链条中，用户输入被传递给提示模板，然后提示模板的输出被传递给模型，然后模型的输出被传
递给输出解析器。
chain = prompt | model | output_parser
chain.invoke({"input":"What's your name?"})

• Prompt：Prompt 是一个 BasePromptTemplate，这意味着它接受一个模板变量的字典并生成一个 PromptValue。PromptValue 可以传递给 LLM（它以字符串作为输入）或 ChatModel（它以消息序列作为输入）。
• Model：将 PromptValue 传递给 model。如果我们的 model 是一个 ChatModel，这意味着它将输出一个 BaseMessage。
• OutputParser：将 model 的输出传递给 output_parser，它是一个 BaseOutputParser，意味着它可以接受字符串或 BaseMessage 作为输入。
• chain：我们可以使用 | 运算符轻松创建这个 Chain。| 运算符在 LangChain 中用于将两个元素组合在一起。
• invoke：所有 LCEL 对象都实现了 Runnable 协议，保证一致的调用方式（invoke / batch / stream）

| 符号类似于 shell 里面管道操作符，它将不同的组件链接在一起，将前一个组件的输出作为下一个组件的输入，这就形成了一个 AI 工作流。

Runnable

Runnable 是 LangChain 定义的一个抽象接口（Protocol），它强制要求所有 LCEL 组件实现一组标准方法：

class Runnable(Protocol):
    def invoke(self, input: Any) -> Any: ...  # 单输入单输出
    def batch(self, inputs: List[Any]) -> List[Any]: ... # 批量处理
    def stream(self, input: Any) -> Iterator[Any]: ...  # 流式输出
    # 还有其他方法如 ainvoke（异步）等...

任何实现了这些方法的对象都被视为 LCEL 兼容组件。比如：聊天模型、提示词模板、输出解析器、检索器、代理(智能体)等。

每个 LCEL 对象都实现了 Runnable 接口，该接口定义了一组公共的调用方法。这使得 LCEL 对象链也自动支持这些调用成为可能。

为什么需要统一调用方式？

传统问题

假设没有统一协议：

• 提示词渲染用 .format()
• 模型调用用 .generate()
• 解析器解析用 .parse()
• 工具调用用 .run()

代码会变成：

prompt_text = prompt.format(topic="猫")  # 方法1
model_out = model.generate(prompt_text) # 方法2
result = parser.parse(model_out)        # 方法3

痛点：每个组件调用方式不同，组合时需要手动适配（或者单独记忆）。

LCEL 解决方案

通过 Runnable 协议统一：

# (分步调用)
prompt_text = prompt.invoke({"topic": "猫"}) # 方法1
model_out = model.invoke(prompt_text)       # 方法2
result = parser.invoke(model_out)           # 方法3

# (LCEL管道式)
chain = prompt | model | parser  # 用管道符组合
result = chain.invoke({"topic": "猫"}) # 所有组件统一用invoke

⚠️ 注意：chain 的调用顺序不能改变，否则会出错。

• 一致性：无论组件的功能多复杂（模型/提示词/工具），调用方式完全相同
• 组合性：管道操作符 | 背后自动处理类型匹配和中间结果传递

传统 Chain 的使用

基础链 LLMChain

LCEL 之前，最基础也最常见的链类型是 LLMChain。

这部分就是在讲述管道符之前，老版本是如何处理 chain 的

这个链至少包括一个提示词模板（PromptTemplate），一个语言模型（LLM 或聊天模型）。

特点：

• 用于 单次问答，输入一个 Prompt，输出 LLM 的响应。
• 适合 无上下文 的简单任务（如翻译、摘要、分类等）。
• 无记忆：无法自动维护聊天历史

主要步骤：

1. 配置任务链

使用 LLMChain 类将任务与提示词结合，形成完整的任务链。

chain = LLMChain(llm = llm, prompt = prompt_template)

2. 执行任务链

使用invoke()等方法执行任务链，并获取生成结果。可以根据需要对输出进行处理和展示。

result = chain.invoke(...)
print(result)

顺序链

顺序链（SequentialChain）允许将多个链顺序连接起来，每个 Chain 的输出作为下一个 Chain 的输入，形成特定场景的流水线（Pipeline）。

顺序链有两种类型：

• 单个输入/输出：对应着 SimpleSequentialChain
• 多个输入/输出：对应着：SequentialChain

SimpleSequentialChain

SimpleSequentialChain：最简单的顺序链，多个链 串联执行，每个步骤都有 单一 的输入和输出，一个步骤的输出就是下一个步骤的输入，无需手动映射。

from langchain.chains.llm import LLMChain
from langchain.chains.sequential import SimpleSequentialChain
from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import ChatOllama

chat_model = ChatOllama(
    model="qwen3:0.6b"
)

chainA_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一位精通各领域知识的知名教授"),
        ("human", "请你尽可能详细的解释一下：{knowledge}"),
    ]
)
chainA_chains = LLMChain(llm=chat_model, prompt=chainA_template, verbose=True)

chainB_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你非常善于提取文本中的重要信息，并做出简短的总结"),
        ("human", "这是针对一个提问的完整的解释说明内容：{description}"),
        ("human", "请你根据上述说明，尽可能简短的输出重要的结论，请控制在20个字以内"),
    ]
)
chainB_chains = LLMChain(llm=chat_model, prompt=chainB_template, verbose=True)

full_chain = SimpleSequentialChain(
    chains=[chainA_chains, chainB_chains],
    verbose=True
)
full_chain.invoke({"input": "什么是langChain？"})
"""
整个流程为：
full_chain.invoke({"input": "什么是langChain？"})：
    会把input的内容：什么是langChain？，传递给chainA，然后返回的结果，作为chainB入口的输入，最后输出结果

chainA：将问题什么是langChain？，组合成prompt：
1. 先生成chainA的prompt：
> Entering new LLMChain chain...
Prompt after formatting:
System: 你是一位精通各领域知识的知名教授
Human: 请你尽可能详细的解释一下：什么是langChain？
2. 调用llm，生成结果：
> Finished chain.
LangChain 是由 LangChain 团队开发的一套自然语言处理（Natural Language Processing, NLP）库，旨在简化和扩展自然语言处理任务的开发流程。它通过模块化的设计和丰富的功能，为开发者提供了高效、灵活的 NLP 工具，广泛应用于对话系统、知识问答、模型集成等多个场景。
3. 然后就会将内容作为chainB的输入，进行下一个步骤

chainB：
1. prompt结果：
System: 你非常善于提取文本中的重要信息，并做出简短的总结
Human: 这是针对一个提问的完整的解释说明内容：LangChain 是由 LangChain 团队开发的一套自然语言处理（Natural Language Processing, NLP）库，旨在简化和扩展自然语言处理任务的开发流程。它通过模块化的设计和丰富的功能，为开发者提供了高效、灵活的 NLP 工具，广泛应用于对话系统、知识问答、模型集成等多个场景。
Human: 请你根据上述说明，尽可能简短的输出重要的结论，请控制在20个字以内
2. 调用llm：
> Finished chain.
LangChain模块化设计，支持多场景应用，跨平台扩展。
"""

SequentialChain

SequentialChain：更通用的顺序链，具体来说：

• 多变量支持：允许不同子链有独立的输入/输出变量。
• 灵活映射：需显式定义变量如何从一个链传递到下一个链。即精准地命名输入关键字和输出关键字，来明确链之间的关系。需要确保变量名前后一一对应。
• 复杂流程控制：支持分支、条件逻辑（分别通过 input_variables 和 output_variables 配置输入和输出）。

这里的多个输出不是指大模型可以多个输出，而是指多个链可以多个输出（每个链都可以输出）。

from langchain.chains.llm import LLMChain
from langchain.chains.sequential import SequentialChain
from langchain_core.prompts import ChatPromptTemplate
from langchain_ollama import ChatOllama

chat_model = ChatOllama(
    model="qwen3:0.6b"
)

chainA_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一位精通各领域知识的知名教授"),
        ("human", "请你先尽可能详细的解释一下：{knowledge}，并且{action}")
    ]
)
chainA_chains = LLMChain(
    llm=chat_model,
    prompt=chainA_template,
    output_key="chainA_chains_key",
    verbose=True
)

chainB_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你非常善于提取文本中的重要信息，并做出简短的总结"),
        ("human", "这是针对一个提问完整的解释说明内容：{chainA_chains_key}"),
        ("human", "请你根据上述说明，尽可能简短的输出重要的结论，请控制在100个字以内"),
    ]
)
chainB_chains = LLMChain(
    llm=chat_model,
    prompt=chainB_template,
    output_key="chainB_chains_key",
    verbose=True
)

# 一定要有input_variables、output_variables参数
ful_chain = SequentialChain(
    chains=[chainA_chains, chainB_chains],
    input_variables=["knowledge", "action"],
    output_variables=["chainA_chains_key", "chainB_chains_key"],
    verbose=True
)
response = ful_chain.invoke({
    "knowledge": "中国足球为什么踢得烂",
    "action": "举一个实际的例子"
})
print(response)
# {'knowledge': '中国足球为什么踢得烂', 'action': '举一个实际的例子', 'chainA_chains_key': 'xxx', 'chainB_chains_key': 'xxx'}

顺序链的使用场景

场景：多数据源处理

举例：根据产品名

1. 查询数据库获取价格
2. 生成促销文案

使用 SimpleSequentialChain（会失败）

# 假设链1返回 {"price": 100}, 链2需要 {product: "xx", price: xx}
# 结构不匹配，无法自动传递！

使用 SequentialChain（正确方式）

from langchain.chains.llm import LLMChain
from langchain.chains.sequential import SequentialChain
from langchain_core.prompts import PromptTemplate
from langchain_ollama import ChatOllama

llm = ChatOllama(
    model="qwen3:0.6b"
)

# 第1环节：
query_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate.from_template(
        template="请模拟查询{product}的市场价格，直接返回一个合理的价格数字（如6999），不要包含任何其他文字或代码"),
    verbose=True,
    output_key="price"
)
# 第2环节：
promo_chain = LLMChain(
    llm=llm,
    prompt=PromptTemplate.from_template(
        template="为{product}（售价：{price}元）创作一篇50字以内的促销文案，要求突出产品卖点"),
    verbose=True,
    output_key="promo_text"
)

ful_chain = SequentialChain(
    chains=[query_chain, promo_chain],
    verbose=True,
    input_variables=["product"],  # 初始输入
    output_variables=["price", "promo_text"],  # 输出价格和文案
)
result = ful_chain.invoke({"product": "iPhone16"})
print("#########\n")
print(result)
# print(result["price"])
# print(result["promo_text"])

在完整 chain(ful_chain) 中的输入（例如product参数，也可以直接给 chainA 和 chainB 使用，不用output_variables输出

基于 LCEL 构建的 Chains 的类型

前面讲解的都是传统的 Chains，下面看最新的基于 LCEL 构建的 Chains（也就是通过管道符|使用）。

1. create_sql_query_chain
2. create_stuff_documents_chain
3. create_openai_fn_runnable
4. create_structured_output_runnable
5. load_query_constructor_runnable
6. create_history_aware_retriever
7. create_retrieval_chain

create_sql_query_chain

create_sql_query_chain，SQL 查询链，是创建生成 SQL 查询的链，用于将 自然语言 转换成 数据库的 SQL 查询

举例：

这里使用 MySQL 数据库，需要安装 pymysql

uv pip install pymysql

from langchain.chains.sql_database.query import create_sql_query_chain
from langchain_community.utilities import SQLDatabase
from langchain_ollama import ChatOllama

chat_model = ChatOllama(
    model="qwen3:0.6b"
)

db_host = "127.0.0.1"
db_port = "3306"
db_name = "java-spring"
db_user = "root"
db_password = "12345678"

# mysql+pymysql://用户名:密码@ip地址:端口号/数据库名
db = SQLDatabase.from_uri(f"mysql+pymysql://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}")

"""
print("哪种数据库：", db.dialect)
print("获取数据表：", db.get_usable_table_names())
# 执行查询
res = db.run("SELECT * FROM tb_user;")
print("查询结果：", res)
"""

chain = create_sql_query_chain(chat_model, db)
response = chain.invoke({
    "question": "一共有多少条数据？", # key必须是question
    "table_names_to_use": ["tb_user"] # 可选
})
# 会根据自然语言，去生成sql语句
print(response) # SQLQuery: SELECT COUNT(*) FROM tb_user;

LCEL 分支与并行

from langchain_core.runnables import RunnableBranch, RunnableParallel

summ = ChatPromptTemplate.from_template("用中文总结：{text}")
bullets = ChatPromptTemplate.from_template("列出三条要点：{text}")

parallel = RunnableParallel(
    summary=summ | llm | StrOutputParser(),
    points=bullets | llm | StrOutputParser(),
)
print(parallel.invoke({"text": "LangChain 是一个模块化的 LLM 应用框架，用于构建复杂对话、RAG、Agent 等系统。"}))

结果：

{
  "summary": "LangChain 是一个模块化的 LLM 应用框架，旨在构建复杂的对话系统、检索增强生成（RAG）和智能代理（Agent）等应用。",
  "points": "1. **模块化设计**：LangChain 提供了灵活的模块化架构，使开发者能够根据需求组合不同的组件，构建复杂的对话系统、检索增强生成（RAG）和智能代理（Agent）等应用。\n\n2. **支持多种功能**：该框架支持多种功能，包括自然语言处理、信息检索、上下文管理和任务执行，帮助开发者快速实现复杂的应用场景。\n\n3. **易于集成**：LangChain 可以与多种外部数据源和 API 集成，增强系统的能力和灵活性，适用于各种行业和应用需求。"
}

容错与回退机制（Fallback）

"""
当主模型超时 / 报错 / 不稳定 / 价格太贵时，
系统可以自动「切换」到另一个备用模型继续回答，
而不会让用户看到 “请求失败” 的尴尬场面。
"""

fallback_llm = ChatOpenAI(
    model="gpt-4o-mini-legacy",
    base_url="xxx",
    api_key="xxx", 
)

robust_chain = (
    ChatPromptTemplate.from_template("回答：{q}")
    | llm | StrOutputParser()
).with_fallbacks([
    ChatPromptTemplate.from_template("回答精简：{q}") | fallback_llm | StrOutputParser()
])

print(robust_chain.invoke({"q": "简述 LangSmith 的作用"}))