LangChain教程——Chain链

原创于 2025-07-02 19:33:24 发布 · 636 阅读

18 ·

CC 4.0 BY-SA版权

文章标签：

#langchain #网络 #前端 #知识图谱 #语言模型 #人工智能 #架构

Chain链

chain链可以将多个组件组合在一起，以创建一个单一、连贯的应用程序，例如：创建一个链接受用户输入，使用提示词模板对其进行格式化并传递给LLM。

链在内部把一系列的功能进行封装，而链的外部又可以组合串联。

通过“ | ”分隔，其格式如下：

chain = 提示词模板 | 模型 | 输出解析器 | ....

# 或
from langchain.chains.llm import LLMChain
chain = LLMChain(
    llm=模型,
    prompt=提示词模板,
    output_parser=输出解析器,
    ....
)

注意：LLMChain方式准备被移除了。

LangChain中主要有以下几种链：

基础链LLMChain
路由链RouterChain
组合链SequentialChain
转换链TransformChain
文档链DocumentsChain
数字链LLMMathChain
SQL查询链create_sql_query_chain

基础链

LLMChain是最基础、最常见的链，它将提示词模板、模型和输出解析器整合在一个链中操作。

它利用提示词模板格式化输入，将格式化后的字符串传递给LLM模型，LLM模型根据输出解析器输出结果，这样使得整个处理过程更加高效便捷。

示例代码如下：

from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 加载model
model = ChatOpenAI(openai_api_key="google/gemma-3-12b", openai_api_base='http://127.0.0.1:1234/v1')

# 定义提示词模板
prompt = PromptTemplate.from_template("请列出5个流行的{subject}品牌，不需要介绍。")

这里我们定义了列表输出解析器和提示词模板，接下来通过“ | ”链接起来，示例代码如下：

# 使用LLMChain创建Chain链
chain = LLMChain(llm=model, prompt=prompt)

content = chain.invoke({"subject" : "手机"})

这里使用了invoke方法执行整个Chain流程，处理输入并返回最终结果。

运行结果如下：

predict方法

predict可以获取模型的预测结果，可能会跳过Chain中的某些步骤。例如：输入预处理或后处理。

其专注于模型的预测部分，而不是整个Chain的流程。

示例代码如下：

from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
from langchain.chains.llm import LLMChain

# 加载model
model = ChatOpenAI(openai_api_key="google/gemma-3-12b", openai_api_base='http://127.0.0.1:1234/v1')

# 定义提示词模板
prompt = PromptTemplate.from_template("请列出5个流行的{subject}品牌，不需要介绍。")

# 使用LLMChain创建Chain链
chain = LLMChain(llm=model, prompt=prompt)

content = chain.predict(subject="手机")
print(content)

注意：predict方法需要LLMChain创建的Chain链才能使用。

运行结果如下：

1. 苹果 (Apple)
2. 三星 (Samsung)
3. 小米 (Xiaomi)
4. 华为 (Huawei)
5. OPPO

与invoke方法相比，predict方法输出更简洁，所以速度也会相对比较快。

batch方法

batch方法允许输入列表运行链，一次处理多个输入，示例代码如下：

from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 加载model
model = ChatOpenAI(openai_api_key="google/gemma-3-12b", openai_api_base='http://127.0.0.1:1234/v1')

# 定义提示词模板
prompt = PromptTemplate.from_template("{subject}的创始人是谁，不需要介绍。")

# Chain链
chain = prompt | model

input_list=[{"subject":"小米"},{"subject":"华为"}]
content = chain.batch(input_list)
for i in content:
    print(i)

运行结果如下：

路由链

路由链（RouterChain）是一种决策链，可以在多个LLMChain之间的路由选择与切换，动态选择最合适的处理该问题的链。

假设，我们现在有两个LLMChain，vegetables和animals链，当问蔬菜相关的问题时，调用vegetables链，当问动物相关问题时，调用animals链。

首先导入相关的模块、加载模型、设置提示词模板，示例代码如下：

from langchain.chains.llm import LLMChain
from langchain.chains.router import MultiPromptChain
from langchain.chains.router.llm_router import RouterOutputParser, LLMRouterChain
from langchain.chains.router.multi_prompt_prompt import MULTI_PROMPT_ROUTER_TEMPLATE
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from langchain_openai import ChatOpenAI

# 加载model
model = ChatOpenAI(openai_api_key="google/gemma-3-12b", openai_api_base='http://127.0.0.1:1234/v1')

# 动物的提示词模板
animals_template = """You are an expert on animals. Please answer the below query:{input}"""
# 蔬菜的提示词模板
vegetables_template = """You are an expert on vegetables. Please answer the below query:{input}"""

接下来将提示词模板封装在列表中，方便我们创建链，示例代码如下：

prompt_infos = [
    {
        "name": "animals",
        "description": "prompt for an animal expert",
        "prompt_template": animals_template,
    },
    {
        "name": "vegetables",
        "description": "prompt for a vegetable expert",
        "prompt_template": vegetables_template,
    },
]

有了提示词模板后，开始封装链，示例代码如下：

# 创建模板链
destination_chains = {}
for p_info in prompt_infos:
    name = p_info["name"]
    # 定义提示词模板
    prompt = ChatPromptTemplate.from_template(template=p_info["prompt_template"])
    # 定义链
    chain = LLMChain(llm=model, prompt=prompt)
    # 将链放在字典中
    destination_chains[name] = chain

为了避免用户问了其他问题无法匹配对应的链，我们还需要创建一个默认链，示例代码如下：

# 默认提示词模板
default_prompt = ChatPromptTemplate.from_template("{input}")
default_chain = LLMChain(prompt=default_prompt, llm=model)

接下来我们开始编写路由模板和路由链，示例代码如下：

# 将animals、vegetables加入到提示词模板描述中
router_template = MULTI_PROMPT_ROUTER_TEMPLATE.format(
    destinations=", ".join(["animals", "vegetables"])
)
# 定义路由提示词模板
router_prompt = PromptTemplate(
    template=router_template,
    input_variables=["input"],
    # 路由输出解析器
    output_parser=RouterOutputParser()
)
# 创建路由链
router_chain = LLMRouterChain.from_llm(prompt=router_prompt,llm=model)

最后通过MultiPromptChain方法来实现路由切换，示例代码如下：

# 实现路由切换
chain = MultiPromptChain(
# 路由链
 router_chain=router_chain,
# 默认链
 default_chain=default_chain,
# 链的描述
 destination_chains=destination_chains,
# 启用详细日志
 verbose=True)

content = chain.invoke({"input": "玉米如何烹饪?"})
print(content)

运行结果如下：

在输出结果中，我们调用了vegetables链。

组合链

组合链（SequentialChain）将多个Chain链连接起来，前一个Chain链的输出作为下一个Chain链的输入，其语法格式如下：

SequentialChain(
    chains=[Chain1, Chain2...],    # 组合链
    input_variables=[第一个链的输入变量],     # 输入变量
    output_variables=[输出变量],     # 输出变量
    verbose=True  # 启用详细日志
 )

在使用LLMChain创建链时，需要使用output_key的指定输出变量的名称。示例代码如下：

from langchain.chains import LLMChain,SequentialChain
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

# 加载model
model = ChatOpenAI(openai_api_key="google/gemma-3-12b", openai_api_base='http://127.0.0.1:1234/v1')

# 第一个链的提示词模板
first_prompt=PromptTemplate.from_template("""只返回{phone}的创始人""")
# 创建链，并指定该输出变量名为Name
first_chain=LLMChain(llm=model,prompt=first_prompt,output_key="Name")
# 第二个链的提示词模板
second_prompt=PromptTemplate.from_template("""请用50字简单介绍该创始人{Name}""")
# 创建链，并指定该输出变量名为slogan
second_chain=LLMChain(llm=model,prompt=second_prompt,output_key="slogan")

接下来使用SequentialChain方法，示例代码如下：

# 组合两个LLMChain
a_v_chains=SequentialChain(
    # 组合链
    chains=[first_chain, second_chain],
    # 输入变量
    input_variables=["phone"],
    # 输出变量
    output_variables=["Name","slogan"],
    verbose=True)
content=a_v_chains.invoke({"phone":"小米"})
print(content)

运行结果如下：

SimpleSequentialChain与SequentialChain相似，SimpleSequentialChain使用于每个链只有一个输入和输出、只需要传递单一值的情况。

示例代码如下：

# 组合两个LLMChain
simple_chain = SimpleSequentialChain(chains=[first_chain, second_chain], verbose=True)
content=simple_chain.invoke("小米")
print(content)

运行结果如下：

转换链

转换链（TransformChain）是一种无模型链，可以通过自定义函数对输入数据进行处理。其不需要调用LLM，所以执行速度快。

示例代码如下：

from langchain.chains import TransformChain

# 定义转换函数，截取输入文档的前10个字符
def transform_func(inputs:dict)->dict:
    text=inputs["text"]
    shortened_text=text[:10]
    return {"output_text":shortened_text}

transform_chain=TransformChain(
    input_variables=["text"],   # 输入变量
    output_variables=["output_text"], # 输出变量
    transform=transform_func,  # 转换函数
    verbose=True,
)
print(transform_chain.invoke({'text':"LangChain 是一个帮助在应用程序中使用大型语言模型（LLM）的编程框架"}))

运行结果如下：

文档链

文档链（create_stuff_documents_chain）将获取文档，将它们全部格式化为文档列表，并将其内容整合到提示词中传递给LLM。

首先进行文档的获取和加载，示例代码如下：

import bs4
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_community.document_loaders import WebBaseLoader
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 加载model
model = ChatOpenAI(openai_api_key="google/gemma-3-12b", openai_api_base='http://127.0.0.1:1234/v1')

# 获取网络上的文章
loader = WebBaseLoader(
    web_path="https://www.ruankao.org.cn/article/content/2506261533205108520700001.html",
    # 获取指定区域的内容
    bs_kwargs=dict(parse_only=bs4.SoupStrainer(id="contentTxt"))
)
docs = loader.load()

这里我们使用WebBaseLoader方法获取网络上的文章，并指定了获取区域的内容。

接下来使用RecursiveCharacterTextSplitter方法将获取的内容分割，示例代码如下：

# 满足任何一个条件就进行分割
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,  # 根据字符长度
    chunk_overlap=200  # 根据行数
)
split_docs = text_splitter.split_documents(docs)

将内容分割后，接下来就可以创建文档链，示例代码如下：

# 定义提示模板
prompt = ChatPromptTemplate.from_template("""根据以下上下文回答问题：{context}问题：{input}""")
# 创建文档链
document_chain = create_stuff_documents_chain(llm=model,prompt=prompt)
result = document_chain.invoke({
    "input": "这篇文章主要是说什么？",
     # 传入分割后的文档
    "context": split_docs
})
print(result)

运行结果如下：

数字链

数字链（LLMMathChain）将用户问题转换为数学问题，然后将数学问题转换为可以使用Python的numexpr库执行的表达式，使用该表达式的输出来回答问题。

在使用前先安装numexpr

pip install numexpr

示例代码如下：

from langchain.chains import LLMMathChain
from langchain_openai import ChatOpenAI
# 加载model
model = ChatOpenAI(openai_api_key="google/gemma-3-12b", openai_api_base='http://127.0.0.1:1234/v1')

# 创建链
llm_math=LLMMathChain.from_llm(model)

# 执行链
res=llm_math.invoke("10**3 +100的结果是多少？")
print(res)

运行结果为：

{'question': '10**3 +100的结果是多少？', 'answer': 'Answer: 1100'}

SQL查询链

SQL查询链（create_sql_query_chain）是创建生成和执行SQL查询的链，用于将自然语言转换为数据库的SQL查询，并与数据库交互获取结果。

示例代码如下：

from langchain_community.utilities import SQLDatabase
from langchain.chains import create_sql_query_chain
from langchain_openai import ChatOpenAI

model = ChatOpenAI(openai_api_key="google/gemma-3-12b", openai_api_base='http://127.0.0.1:1234/v1')

HOST = '127.0.0.1'
USERNAME = 'root'
PASSWORD = '123456'
PORT = 3306
DB = 'music_wechat'
# 创建数据库连接
db=SQLDatabase.from_uri(f"mysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DB}")

# 获取SQL链对象
chain=create_sql_query_chain(llm=model, db=db)

response=chain.invoke({"question":"查询一共有多少mv"})
print(response)

运行结果如下：

我们可以使用table_names_to_use参数来限制使用哪个表，示例代码如下：

# 限制使用表
response=chain.invoke({"question":"查询一共有多少数据","table_names_to_use":["mv"]})
print(response)

运行结果如下：

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

在这里插入图片描述