LangChain 链式应用实战：多种 Chain 类型详解与案例

create_extraction_chain：提取信息链

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：数学链

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)

LLMRequestsChain：获取 URL 数据链

根据用户输入构造一个 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 查询链

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：API 链

APIChain 使用 LLM 将查询转换为 API 请求，然后执行该请求，获取响应，然后将该请求传递给 LLM 以进行响应。

实现步骤

1. 准备 API 文档描述（Prompt）。
2. 初始化 LLM。
3. 构建 APIChain。

代码示例

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("成都今天天气情况？")

Sequential Chain：顺序链

一条链的输出直接馈送到下一条链的链。适用于多步骤处理流程。

代码示例

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)

LLMRouterChain：路由链

针对场景，可以构建不同的目标链。LangChain 通过 LLMRouterChain 来自动引导大语言模型选择不同的模板，以应对不同类型的问题。

实现步骤

1. 构建多个场景的提示模板。
2. 构建目标链。
3. 构建路由链。
4. 构建默认链。
5. 整合为 MultiPromptChain。

代码示例

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 的应用场景与实现方式。为了在实际项目中更好地使用这些组件，建议遵循以下最佳实践：

1. 上下文窗口管理：在使用文档链时，务必检查文档总长度是否超出模型限制。对于长文档，考虑使用 Map-Reduce 策略或检索增强生成（RAG）技术。
2. 错误处理：在调用外部 API 或数据库时，应增加异常捕获机制，防止因网络波动或权限问题导致程序崩溃。
3. 成本优化：LLM 调用通常按 Token 计费。尽量精简 Prompt，避免不必要的重复计算，并在生产环境中缓存常用结果。
4. 安全性：严禁将 API Key 硬编码在代码中。使用环境变量或密钥管理服务（如 AWS Secrets Manager）。
5. 调试技巧：开启 verbose=True 可以查看链的执行日志，有助于定位中间步骤的错误。

通过灵活组合上述 Chain 类型，开发者可以快速构建出复杂的大语言模型应用流程，满足多样化的业务需求。