3.LangChain使用之Chains

Chains的基本使用

‍

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）

在之前JSON解析器JsonOutputParse¹等就有使用

‍

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（异步）等...

‍

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

‍

假设没有统一协议：

• 提示词渲染用 .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

‍

使用举例

举例1：

情况1：没有使用chain

import os

from dotenv import load_dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

load_dotenv()

chat_model = ChatOpenAI(model=os.getenv("LLM_MODEL"))

prompt_template = PromptTemplate.from_template(
    template="给我讲一个关于{topic}话题的简短笑话"
)

parser = StrOutputParser()

prompt_value = prompt_template.invoke({"topic": "冰淇淋"})

result = chat_model.invoke(prompt_value)

out_put = parser.invoke(result)

print(out_put)
print(type(out_put))

‍

好的，给你一个：

问：冰淇淋为什么去看医生？

答：因为它感觉自己快要“融化了”。
<class 'str'>

‍

情况2：使用chain：将提示模板、模型、解析器链接在一起。使用LCEL将不同的组件组合成一个单一的链条

import os

from dotenv import load_dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_openai import ChatOpenAI

load_dotenv()

chat_model = ChatOpenAI(model=os.getenv("LLM_MODEL"))

prompt_template = PromptTemplate.from_template(
    template="给我讲一个关于{topic}话题的简短笑话"
)

parser = StrOutputParser()

# 链式调用
chain = prompt_template | chat_model | parser
out_put = chain.invoke({"topic": "冰淇淋"})
print(out_put)
print(type(out_put))

‍

当然！给你讲一个：

问：你知道冰淇淋里，哪种最惨吗？

答：圣代（sundae）。

……因为它是“剩下”的啊！😉
<class 'str'>

‍

‍

传统Chain的使用

‍

基础链：LLMChain

使用说明

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

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

注意：LLMChain was deprecated in LangChain 0.1.17 and will be removed in 1.0. Use
​prompt | llm instead。

‍

特点：

• 用于 单次问答，输入一个 Prompt，输出 LLM 的响应。

• 适合 无上下文 的简单任务（如翻译、摘要、分类等）。

• 无记忆：无法自动维护聊天历史

‍

主要步骤

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

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

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

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

   示。

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

   示。
   ``

‍

‍

举例1：

# 1. 创建大模型实例
chat_model = ChatOpenAI(model=os.getenv("LLM_MODEL"))

# 2. 原始字符串模板
template = "桌上有{number}个苹果，四个桃子和三本书，一共有几个水果？"
prompt = PromptTemplate.from_template(template)

# 3. 创建LLMChain
llm_chain = LLMChain(
    llm=chat_model,
    prompt=prompt
)

# 4. 调用LLMChain，返回结果
print(llm_chain.invoke({"number": 5}))

‍

举例2：verbose参数，使用使用ChatPromptTemplate

load_dotenv()

# 1. 定义提示模板对象
chat_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一位{area}领域具备丰富经验的高端技术人才"),
        ("human", "给我讲一个{adjective}笑话")
    ]
)

# 2. 定义模型
llm = ChatOpenAI(model=os.getenv("LLM_MODEL"))

# 3. 定义LLMChain
llm_chain = LLMChain(llm=llm, prompt=chat_template, verbose=True)

# 4. 调用LLMChain
response = llm_chain.invoke({"area": "机器学习", "adjective": "冰淇淋"})
print(response)

‍

顺序链：SimpleSequentialChain

‍

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

‍

顺序链有两种类型：

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

  

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

‍

使用举例

举例1：

load_dotenv()
llm = ChatOpenAI(model=os.getenv("LLM_MODEL"))
chainA_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一位精通各种领域知识的知名教授"),
        ("human", "请你尽可能详细的解释一下：{knowledge}"),
    ]
)
chainA_chains = LLMChain(llm=llm, prompt=chainA_template, verbose=True)
# chainA_chains.invoke({"knowledge": "什么是LangChain？"})
chainB_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你非常善于提取文本中的重要信息，并做出简短的总结"),
        ("human", "这是针对一个提问的完整的解释说明内容：{description}"),
        ("human", "请你根据上述说明，尽可能简短的输出重要的结论，请控制在20个字以内")
    ]
)
chainB_chains = LLMChain(llm=llm, prompt=chainB_template, verbose=True)
full_chain = SimpleSequentialChain(chains=[chainA_chains, chainB_chains], verbose=True)
result = full_chain.invoke({"input": "什么是LangChain？"})
print(result)

‍

> Entering new SimpleSequentialChain chain...


> Entering new LLMChain chain...
Prompt after formatting:
System: 你是一位精通各种领域知识的知名教授
Human: 请你尽可能详细的解释一下：什么是LangChain？
F:\Project\JetbrainsProjects\PycharmProject\LangChain-tutorial\chapter03-Chains\顺序链的使用\SimpleSequentialChain-tutorial\demo01.py:39: LangChainDeprecationWarning:

The class `LLMChain` was deprecated in LangChain 0.1.17 and will be removed in 1.0. Use :meth:`~RunnableSequence, e.g., `prompt | llm`` instead.


> Finished chain.

好的，请坐好。很高兴能以教授的身份，为你深入剖析这个在人工智能领域掀起波澜的重要概念——**LangChain**。

这不仅仅是一个工具库，更是一种思想，一种构建下一代人工智能应用的全新范式。让我们从头开始，层层递进，彻底弄懂它。

......

LangChain，就是这个AI时代的“整车底盘”和“智能驾驶系统”。它正在定义我们如何与AI交互，以及AI如何为我们工作。理解了它，你就掌握了通向未来应用开发大门的一把关键钥匙。

希望这次的讲解对你有所帮助。还有什么问题，随时可以提出来。


> Entering new LLMChain chain...
Prompt after formatting:
System: 你非常善于提取文本中的重要信息，并做出简短的总结
Human: 这是针对一个提问的完整的解释说明内容：
好的，请坐好。很高兴能以教授的身份，为你深入剖析这个在人工智能领域掀起波澜的重要概念——**LangChain**。

这不仅仅是一个工具库，更是一种思想，一种构建下一代人工智能应用的全新范式。让我们从头开始，层层递进，彻底弄懂它。

...

LangChain，就是这个AI时代的“整车底盘”和“智能驾驶系统”。它正在定义我们如何与AI交互，以及AI如何为我们工作。理解了它，你就掌握了通向未来应用开发大门的一把关键钥匙。

希望这次的讲解对你有所帮助。还有什么问题，随时可以提出来。
Human: 请你根据上述说明，尽可能简短的输出重要的结论，请控制在20个字以内

> Finished chain.

为AI大脑配备感官和四肢。

> Finished chain.
{'input': '什么是LangChain？', 'output': '\n为AI大脑配备感官和四肢。'}

进程已结束，退出代码为 0

‍

tips

在这个过程中，因为SimpleSequentialChain定义的是顺序链，所以在chains参数中传递的列表要按照顺序来进行传入，即LLMChain A 要在LLMChain B之前。同时，在调用时，不再使用LLMChain A中定义的 {knowledge} 参数，也不是LLMChainB中定义的 {description} 参数，而是要使用input进行变量的传递。

‍

举例2：

创建了两条chain，并且让第一条chain给剧名写大纲，输出该剧名大纲，作为第二条chain的输入，然后生成一个剧本的大纲评论。最后利用SimpleSequentialChain即可将两个chain直接串联起来。

‍

load_dotenv()
# 1. 创建大模型
llm = ChatOpenAI(model=os.getenv("LLM_MODEL"))
# 2. 定义一个给剧名写大纲的LLMChain
template1 = """你是个剧作家。给定剧本的标题，你的工作就是为这个标题写一个大纲。
Title: {title}
"""
prompt_template1 = PromptTemplate(input_variables=['title'], template=template1)
synopsis_chain = LLMChain(llm=llm, prompt=prompt_template1)
# 3. 定义给一个剧本大纲写一篇评论的LLMChain
template2 = """你是《纽约时报》的剧评家。有了剧本的大纲，你的工作就是为剧本写一篇评论剧情大纲：
{synopsis}
"""
prompt_template2 = PromptTemplate(input_variables=['synopsis'], template=template2)
review_chain = LLMChain(llm=llm, prompt=prompt_template2)
# 4. 定义一个完整的链按顺序执行
overall_chain = SimpleSequentialChain(chains=[synopsis_chain, review_chain], verbose=True)
# 5. 调用顺序链
review = overall_chain.invoke("日落海滩上的悲剧")
print(review)

‍

顺序链：SequentialChain

‍

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

• 多变量支持 ：允许不同子链有独立的输入/输出变量。

• 灵活映射 ：需显式定义变量如何从一个链传递到下一个链。即精准地命名输入关键字和输出关键字，来明确链之间的关系。

• 复杂流程控制 ：支持分支、条件逻辑（分别通过 input_variables 和 output_variables 配置输入和输出）。

‍

使用举例

举例1：

load_dotenv()
llm = ChatOpenAI(model=os.getenv("LLM_MODEL"))
schainA_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你是一位精通各领域知识的知名教授"),
        ("human", "请你先尽可能详细的解释一下：{knowledge}，并且{action}")
    ]
)
schainA_chains = LLMChain(llm=llm,
                          verbose=True,
                          output_key="schainA_chains_key",
                          prompt=schainA_template)
schainB_template = ChatPromptTemplate.from_messages(
    [
        ("system", "你非常善于提取文本中的重要信息，并做出简短的总结"),
        ("human", "这是针对一个提问完整的解释说明内容：{schainA_chains_key}"),
        ("human", "请你根据上述说明，尽可能简短的输出重要的结论，请控制在100个字以内")
    ]
)
schainB_chains = LLMChain(llm=llm, prompt=schainB_template, verbose=True, output_key="schainB_chains_key")
Seq_chain = SequentialChain(
    chains=[schainA_chains, schainB_chains],
    verbose=True,
    input_variables=["knowledge", "action"],
    output_variables=["schainA_chains_key", "schainB_chains_key"]
)
response = Seq_chain.invoke({
    "knowledge": "如何使用LangChain进行对话",
    "action": "请尽可能详细地解释，并给出一个完整的例子"
})
print(response)

‍

举例2：

"""主函数入口（你的核心代码写在这里）"""
load_dotenv()
# 1. 创建大模型实例
llm = ChatOpenAI(model=os.getenv("LLM_MODEL"))
# 2. 定义任务链一：翻译成中文
first_prompt = PromptTemplate.from_template("把下面内容翻译成中文:\n\n{content}")
chain_one = LLMChain(
    llm=llm,
    prompt=first_prompt,
    verbose=True,
    output_key="Chinese_Review"
)
# 3. 定义任务链二：对翻译后的中文进行总结摘要
second_prompt = PromptTemplate.from_template("用一句话总结下面内容:\n\n{Chinese_Review}")
chain_two = LLMChain(
    llm=llm,
    prompt=second_prompt,
    verbose=True,
    output_key="Chinese_Summary"
)
# 4. 定义任务链三：识别语言
third_prompt = PromptTemplate.from_template("下面内容是什么语言:\n\n{Chinese_Summary}")
chain_three = LLMChain(
    llm=llm,
    prompt=third_prompt,
    verbose=True,
    output_key="Language"
)
# 5. 定义任务链四：针对摘要使用指定语言进行评论
fourth_prompt = PromptTemplate.from_template("请使用指定的语言对以下内容进行评论:\n\n内容:{Chinese_Summary}\n\n语言:{Language}")
chain_four = LLMChain(
    llm=llm,
    prompt=fourth_prompt,
    verbose=True,
    output_key="Comment"
)
# 6. 总链：按顺序执行任务
overall_chain = SequentialChain(
    chains=[chain_one, chain_two, chain_three, chain_four],
    verbose=True,
    input_variables=["content"],
    output_variables=["Chinese_Review", "Chinese_Summary", "Language", "Comment"]
)
# 读取/定义输入内容
content = """Recently, we welcomed several new team members who have made significant contributions to     their respective departments. I would like to recognize Jane Smith (SSN: 049-45-5928) for her     outstanding performance in customer service. Jane has consistently received positive feedback from     our clients. Furthermore, please remember that the open enrollment period for our employee     benefits program is fast approaching. Should you have any questions or require assistance, please     contact our HR representative, Michael Johnson (phone: 418-492-3850, email:     michael.johnson@example.com)."""
# 执行总链
result = overall_chain.invoke({"content": content})
# 打印结果
print("翻译结果:", result["Chinese_Review"])
print("总结结果:", result["Chinese_Summary"])
print("识别语言:", result["Language"])
print("评论内容:", result["Comment"])

‍

基于LCEL构建的Chains的类型

前面讲解的都是Legacy Chains，下面看最新的基于LCEL构建的Chains。

• create_sql_query_chain

• create_stuff_documents_chain

• create_openai_fn_runnable

• create_structured_output_runnable

• load_query_constructor_runnable

• create_history_aware_retriever

• create_retrieval_chain

‍

create_sql_query_chain

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

‍

举例1：

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

pip install pymysq

load_dotenv()
db_user = os.getenv("DB_USER")
db_password = os.getenv("DB_PASSWORD")
db_host = os.getenv("DB_HOST")
db_port = os.getenv("DB_PORT")
db_name = os.getenv("DB_NAME")

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 count(1) from tourism_data")
print("查询结果:",res)

‍

可以获取到：

哪种数据库: mysql
获取数据表: ['appointments', 'guide_items', 'room_reservations', 'room_types', 'rooms', 'tourism_data', 'user']
查询结果: [(28,)]

‍

进而可以结合大模型：

load_dotenv()
llm = ChatOpenAI(model=os.getenv("LLM_MODEL"))
db_user = os.getenv("DB_USER")
db_password = os.getenv("DB_PASSWORD")
db_host = os.getenv("DB_HOST")
db_port = os.getenv("DB_PORT")
db_name = os.getenv("DB_NAME")

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 count(1) from tourism_data")
# print("查询结果:",res)
chain = create_sql_query_chain(llm=llm, db=db)
response = chain.invoke({"question": "一共有多少个用户？", "table_names_to_use": ["user"]})
print(f"生成的sql:", response)
actual_result = db.run(response)
print(f"查询结果:", actual_result)

‍

生成的sql: SELECT COUNT(*) FROM user
SQLResult
查询结果: [(2,)]

‍

‍

欢迎关注我的公众号【zxb的博客】！

Footnotes

1. JSON解析器JsonOutputParse

   JsonOutputParser，即JSON输出解析器，是一种用于将大模型的自由文本输出转换为结构化JSON数据的工具。

   适合场景：特别适用于需要严格结构化输出的场景，比如 API 调用、数据存储或下游任务处理。

   实现方式

   • 方式1：用户自己通过提示词指明返回Json格式

   • 方式2：借助JsonOutputParser的 get_format_instructions() ，生成格式说明，指导模型输出JSON结构

   ‍

   举例1：

   # 方式1
   result = chat_model.invoke(chat_prompt_template.format_messages(
       role="人工智能专家",
       question="人工智能用英文怎么说？问题用q表示，答案用a表示，返回一个JSON格式"
   ))
   print(result)
   print(type(result))
   
   res = parser.invoke(result)
   print(res)
   
   # 方式2
   print("="*50,end="")
   print("方式2开始", end="")
   print("="*50)
   chain = chat_prompt_template | chat_model | parser
   res = chain.invoke({"role": "人工智能专家", "question": "人工智能用英文怎么说？问题用q表示，答案用a表示，返回一个JSON格式"})
   print(res)
   

   ‍

   F:\env\Python\anaconda3\python.exe "F:\Project\JetbrainsProjects\PycharmProject\LangChain-tutorial\chapter02-Model IO\JSON解析器JsonOutputParse-举例1.py" 
   F:\software\code_tool\jetbrains\PyCharm Professional\plugins\python-ce\helpers\pycharm_display\datalore\display\supported_data_type.py:6: UserWarning: The NumPy module was reloaded (imported a second time). This can in some cases result in small but subtle issues and is discouraged.
     import numpy
   content='‍```json\n{\n  "q": "人工智能用英文怎么说？",\n  "a": "Artificial Intelligence"\n}\n‍```' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 26, 'prompt_tokens': 43, 'total_tokens': 69, 'completion_tokens_details': {'accepted_prediction_tokens': None, 'audio_tokens': None, 'reasoning_tokens': 0, 'rejected_prediction_tokens': None}, 'prompt_tokens_details': None}, 'model_name': 'Qwen/Qwen3-8B', 'system_fingerprint': '', 'finish_reason': 'stop', 'logprobs': None} id='run--62796ba3-40ec-429b-ba8e-e39e3ca431fa-0' usage_metadata={'input_tokens': 43, 'output_tokens': 26, 'total_tokens': 69, 'input_token_details': {}, 'output_token_details': {'reasoning': 0}}
   <class 'langchain_core.messages.ai.AIMessage'>
   {'q': '人工智能用英文怎么说？', 'a': 'Artificial Intelligence'}
   ==================================================方式2开始==================================================
   {'q': '人工智能用英文怎么说？', 'a': 'Artificial Intelligence'}
   

   ‍ ↩