本文详细介绍了 LangChain 中多种 Chain 链的应用场景与实现方式。涵盖 LLMChain 基础用法、文档处理链、信息提取链、数学计算链、URL 请求链、SQL 查询链、API 调用链、顺序链及路由链。通过具体代码示例展示了如何配置环境、构建提示模板、连接数据库及外部 API，并演示了多链组合与动态路由逻辑。内容旨在帮助开发者快速掌握 LangChain 核心组件，构建复杂的大语言模型应用流程，并提供环境配置、错误处理及成本优化等最佳实践建议。
LangChain 根据功能、用途的不同，提供了大量的 Chain（链）组件。这些链将语言模型推理功能与提示模板、输出解析器等功能整合在一起，使得整个处理过程更加高效和便捷。本文将详细介绍各种 Chain 的使用示例及最佳实践。
在使用 LangChain 之前，需要配置 OpenAI 的环境变量信息，指定 URL 和 API Key。
import os
# 设置环境变量
os.environ["OPENAI_BASE_URL"] = "https://api.openai.com/v1"
os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"
注意：请勿在代码中硬编码真实的 API Key，建议使用环境变量管理敏感信息。
LLMChain 是最基础也是最常见的链。它结合了语言模型推理功能，并添加了 PromptTemplate 和 Output Parser 等功能，将模型输入输出整合在一个链中操作。
from langchain.chains import LLMChain
from langchain_core.prompts import PromptTemplate
from langchain_openai import OpenAI
# 原始字符串模板
template = "猪八戒吃{fruit}?"
# 创建模型实例
llm = OpenAI(temperature=0)
# 创建 LLMChain
llm_chain = LLMChain(
llm=llm,
prompt=PromptTemplate.from_template(template))
# 调用 LLMChain，返回结果
result = llm_chain.invoke({"fruit": "人参果"})
print(result)
create_stuff_documents_chain 链将获取文档列表并将它们全部格式化为提示（文档列表），然后将该提示传递给 LLM。
from langchain_openai import ChatOpenAI
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain.chains.combine_documents import create_stuff_documents_chain
# 创建提示模板
prompt = ChatPromptTemplate.from_messages(
[("system", "根据提供的上下文：{context} \n\n 回答问题：{input}")])
# 初始化大模型
llm = ChatOpenAI(model="gpt-3.5-turbo")
# 构建链
chain = create_stuff_documents_chain(llm, prompt)
# 定义文档内容
docs = [
Document(page_content="杰西喜欢红色，但不喜欢黄色"),
Document(page_content="贾马尔喜欢绿色，有一点喜欢红色"),
Document(page_content="玛丽喜欢粉色和红色")
]
# 执行链
res = chain.invoke({"input": "大家喜欢什么颜色？", "context": docs})
print(res)
create_extraction_chain 是一个从文章、段落中提取信息的链。使用 OpenAI 函数调用从文本中提取信息，定义一个模式，指定从 LLM 输出中提取的属性。
from langchain.chains import create_extraction_chain
from langchain_openai import ChatOpenAI
# 模式定义
schema = {
"properties": {
"name": {"type": "string"},
"height": {"type": "integer"},
"hair_color": {"type": "string"},
},
"required": ["name", "height"],
}
# 输入文本
inp = """亚历克斯身高 5 英尺。克劳迪娅比亚历克斯高 1 英尺，并且跳得比他更高。克劳迪娅是黑发女郎，亚历克斯是金发女郎。"""
# 初始化大模型
llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo")
# 构建链
chain = create_extraction_chain(schema, llm)
# 执行
res = chain.invoke(inp)
print(res)
LLMMathChain 将用户问题转换为数学问题，然后将数学问题转换为可以使用 Python 的 numexpr 库执行的表达式。使用运行此代码的输出来回答问题。
使用 LLMMathChain，需要安装 numexpr 库：
pip install numexpr
from langchain_openai import OpenAI
from langchain.chains import LLMMathChain
# 初始化大模型
llm = OpenAI()
# 创建链
llm_math = LLMMathChain.from_llm(llm)
# 执行链
res = llm_math.invoke("100 * 20 + 100 的结果是多少？")
print(res)
根据用户输入构造一个 URL，获取该 URL 上的数据，然后总结响应。
安装 BeautifulSoup：
pip install bs4
from langchain_openai import OpenAI
from langchain.chains import LLMRequestsChain, LLMChain
from langchain.prompts import PromptTemplate
# 初始化大模型
llm = OpenAI(model="gpt-3.5-turbo", temperature=0)
# 定义使用的提示
template = """
来自 google 的原始搜索结果文本：{requests_result}
根据搜索结果回答问题：{query}，如果没有相关答案则回复：'搜索查询失败'
回答格式:
question：问题在这里
answer：答案在这里
"""
prompt = PromptTemplate(
input_variables=["query", "requests_result"],
template=template,
)
# 实例化 LLMRequestsChain
chain = LLMRequestsChain(llm_chain=LLMChain(llm=OpenAI(temperature=0), prompt=prompt))
# 定义输入
question = "成都今天天气情况?"
inputs = {
"query": question,
"url": "https://www.google.com/search?q=" + question
}
# 运行 Chain
res = chain.invoke(inputs)
print(res)
create_sql_query_chain 是创建生成 SQL 查询的链，用于将自然语言转换成数据库的 SQL 查询。
SQLDatabaseChain 可与 SQLAlchemy 支持的任何 SQL 方言一起使用，例如 MS SQL、MySQL、MariaDB、PostgreSQL、Oracle SQL、Databricks 和 SQLite。
这里以 MySQL 为例，需要安装 pymysql：
pip install pymysql
from langchain_community.utilities import SQLDatabase
# 连接 MySQL 数据库
db_user = "root"
db_password = "12345678"
db_host = "IP"
db_port = "3306"
db_name = "demo"
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(*) FROM tb_users;")
print("查询结果：", res)
# 初始化语言模型并构建链
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
chain = create_sql_query_chain(llm=llm, db=db)
response = chain.invoke({"question": "数据表 tb_users 中有多少用户？"})
print(response)
APIChain 使用 LLM 将查询转换为 API 请求，然后执行该请求，获取响应，然后将该请求传递给 LLM 以进行响应。
from langchain.chains import APIChain
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
# API 文档内容
OPEN_WEATHER_DOCS = """BASE URL: https://apis.tianapi.com/tianqi/index
API 文档内容如下：
APIKEY：your_api_key_here
1. 接口信息
根据城市名称查询实时或七天内天气预报
接口地址：https://apis.tianapi.com/tianqi/index
请求方式：get/post
返回格式：utf-8 json
2. 请求参数
key: string (必需) - API 密钥
city: string (必需) - 城市 ID
type: int (必需) - 查询类型
"""
# 初始化大模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# 创建链
chain = APIChain.from_llm_and_api_docs(
llm,
OPEN_WEATHER_DOCS,
limit_to_domains=["https://apis.tianapi.com/tianqi/index"],
verbose=True
)
def ask(msg):
messages = [
SystemMessage(content="你是一位天气查询助手，回答使用中文语言。"),
HumanMessage(content=msg)
]
res = chain.invoke(messages)
print(res)
if __name__ == '__main__':
ask("成都今天天气情况？")
一条链的输出直接馈送到下一条链的链。适用于多步骤处理流程。
from langchain.chains.sequential import SequentialChain
from langchain.chains.llm import LLMChain
from langchain_core.prompts import PromptTemplate
from langchain_openai import OpenAI
llm = OpenAI()
# 第一个链：生成概述
template1 = """你是一名小说文学家，请你根据小说名称与人物，为小说人物写一个 100 字左右的概述。
小说：{book}
人物：{name}
小说人物的概述:"""
prompt1 = PromptTemplate(input_variables=["book", "name"], template=template1)
description_chain = LLMChain(llm=llm, prompt=prompt1, output_key="description")
# 第二个链：写出评论
template2 = """你是一位文学评论家，请你根据小说人物的概述，写一篇 100 字左右的评论。
小说人物概述：{description}
小说人物的评论:"""
prompt2 = PromptTemplate(input_variables=["description"], template=template2)
comments_chain = LLMChain(llm=llm, prompt=prompt2, output_key="comments")
# 第三个链：写宣传广告
template3 = """你是一名广告策划师，请根据小说人物的概述与评论写一篇宣传广告，字数 50 字左右。
小说人物概述：{description}
小说人物的评论：{comments}
宣传广告:"""
prompt3 = PromptTemplate(input_variables=["description", "comments"], template=template3)
promotionalAd_chain = LLMChain(llm=llm, prompt=prompt3, output_key="promotionalAd")
# 合并链
overall_chain = SequentialChain(
chains=[description_chain, comments_chain, promotionalAd_chain],
input_variables=["book", "name"],
output_variables=["description", "comments", "promotionalAd"],
verbose=True
)
# 运行
result = overall_chain.invoke({"book": "西游记", "name": "猪八戒"})
print(result)
针对场景，可以构建不同的目标链。LangChain 通过 LLMRouterChain 来自动引导大语言模型选择不同的模板，以应对不同类型的问题。
from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser
from langchain.chains.router.multi_prompt_prompt import MULTI_PROMPT_ROUTER_TEMPLATE as RounterTemplate
from langchain.chains.conversation.base import ConversationChain
from langchain.chains.router import MultiPromptChain
from langchain.chains.llm import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import OpenAI
llm = OpenAI()
# 1. 构建提示模板
hobby_template = """你是一名小说文学家，非常熟悉小说西游记，知晓小说人物的爱好信息。
问题：{input}"""
nature_template = """你是一名小说文学家，非常熟悉小说西游记，知晓小说人物的性格特征。
问题：{input}"""
prompt_list = [
{"id": "hobby", "description": "回答关于小说人物的爱好信息问题", "template": hobby_template},
{"id": "nature", "description": "回答关于小说人物的性格特征问题", "template": nature_template},
]
# 2. 构建目标链
chain_map = {}
for template in prompt_list:
prompt = PromptTemplate(template=template['template'], input_variables=["input"])
chain = LLMChain(llm=llm, prompt=prompt, verbose=True)
chain_map[template["id"]] = chain
# 3. 构建路由链
destinations = []
for p in prompt_list:
destination = f"{p['id']}: {p['description']}"
destinations.append(destination)
router_template = RounterTemplate.format(destinations="\n".join(destinations))
router_prompt = PromptTemplate(
template=router_template,
input_variables=["input"],
output_parser=RouterOutputParser(),
)
router_chain = LLMRouterChain.from_llm(llm, router_prompt, verbose=True)
# 4. 构建默认链
default_chain = ConversationChain(llm=llm, output_key="text", verbose=True)
# 5. 构建多提示链
chain = MultiPromptChain(
router_chain=router_chain,
destination_chains=chain_map,
default_chain=default_chain,
verbose=True
)
# 测试
print(chain.invoke({"input": "猪八戒的性格是怎样的？"}))
本文详细介绍了 LangChain 中多种 Chain 的应用场景与实现方式。为了在实际项目中更好地使用这些组件，建议遵循以下最佳实践：
verbose=True 可以查看链的执行日志，有助于定位中间步骤的错误。通过灵活组合上述 Chain 类型，开发者可以快速构建出复杂的大语言模型应用流程，满足多样化的业务需求。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online