人工智能的黄金链条：LangChain核心模块Chains解析

简介

Chain 作为 LangChain 的核心模块之一，重要性不言而喻，它相当于是所有复杂逻辑的基础，用来将每个零散的逻辑串联成一整个业务流程，Chain 的设计非常巧妙，可以说是大模型应用的最佳实践之一。

设计思路

Chain 的设计非常巧妙，也非常值得借鉴，也是为什么 LangChain 能火爆的主要原因之一。

Chain 的一个主要作用，就是根据需求，可以将各种能力拼接整合。比如上图，LangChain 内部包含多个模块，可以定制 LLMChain 只使用 Prompt 和 LLM 模块，也可以定制一个HogwartsChain 使用了 LLM、Parser、Agent模块。

实践演练

Chains主要包含以下几个模块，接下来的实践演练，会分别演示这几个模块的使用。

LLMChain

LLMChain是一个整合语言模型和提示模板的最简单链。

# LangChain相关模块的导入
from langchain import LLMChain
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

# 加载个人的OpenAI Token
key = 'open_ai_key'
# 创建OpenAI调用实例
# 本示例中为了让结果更具有创造性，temperature设置为0.9
llm = ChatOpenAI(temperature=0.9, openai_api_key=key)
# 根据prompt模板生成prompt实例
prompt = ChatPromptTemplate.from_template(
    "请给生产: {product} 的工厂起一个恰当的厂名，并给出一句广告语。"
)
# 组合大模型实例和prompt实例，生成LLMChain实例，将结构固定，方便复用
chain = LLMChain(
    # 大模型实例
    llm=llm,
    # prompt实例
    prompt=prompt,
    # 开启详细模式，会将大模型调用细节输出到控制台
    verbose=True
)
# 通过run方法，传入模版中需要的参数，调用大模型获取结果
product = "IPhone999 Pro Max Ultra"
res = chain.run(product)
print(res)

返回数据示例:

其中绿色的部分就是请求大模型时发送的完整数据，通过查看发送的数据，可以更有效的了解调用细节、排查问题。

SimpleSequentialChain

串联式调用语言模型链的一种，简单的串联每个步骤（Chain 实例），每个步骤都有单一的输入/输出，并且一个步骤的输入是下一个步骤的输出。

# LangChain相关模块的导入
from langchain import LLMChain
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.chains import SimpleSequentialChain
# 加载个人的OpenAI Token
key = 'open_ai_key'

# 创建OpenAI调用实例
# temperature用来设置大模型返回数据的随机性和创造性，较低的数值返回的数据就更贴近现实。
llm = ChatOpenAI(temperature=0.9, openai_api_key=key)
# 第一个LLM请求的prompt模板
first_prompt = ChatPromptTemplate.from_template(
    "请给生产 {product} 的工厂起一个恰当的厂名"
)
# 第一个Chain，接收外部输入，根据模版请求大模型获取输出，作为第二个Chain的输入
chain_one = LLMChain(llm=llm, prompt=first_prompt, verbose=True)
# 第二个大模型请求的prompt模版
second_prompt = ChatPromptTemplate.from_template(
    "为厂名写一段不少于20字的广告语: {company_name}"
)
# 第二个Chain，接收第一个Chain的输出，根据模版请求大模型获取输出
chain_two = LLMChain(llm=llm, prompt=second_prompt, verbose=True)
# 将请求拆分成两个Chain，可以针对每段请求细化相应的prompt内容，得到更准确更合理的结果，并且也可以复用其中的每个Chain实例
# 使用SimpleSequentialChain将两个Chain串联起来，其中每个Chain都只支持一个输入和一个输出，根据chains列表中的顺序，将前一个Chain的输出作为下一个Chain的输入
overall_simple_chain = SimpleSequentialChain(
    chains=[chain_one, chain_two],
    verbose=True
)
# 第一个Chain需要的输入
product = "IPhone999 Pro Max Ultra"
# 通过run方法，传入参数，逐个运行整个Chain后，获取最终的结果
res = overall_simple_chain.run(product)
print(res)

返回数据示例如图:

SequentialChain

串联式调用语言模型链的一种，序列中的每个 Chain 实例都支持多个输入和输出，最终 SequentialChain 运行时根据 Chains 参数和每个 Chain 示例中设定的参数，分析每个实例所需的参数并按需传递。代码示例:

import langchain
# LangChain相关模块的导入
from langchain import LLMChain
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.chains import SequentialChain

# 在全局范围开启详细模式，能将调用大模型时发送的数据打印到控制台，绿色文本
langchain.verbose = True

key = 'open_ai_key'

# 本示例中为了让结果更具有创造性，temperature设置为0.9
llm = ChatOpenAI(temperature=0.9, openai_api_key=key)

# Chain1 语言转换，产生英文产品名
prompt1 = ChatPromptTemplate.from_template(
    "将以下文本翻译成英文: {product_name}"
)
chain1 = LLMChain(
    # 使用的大模型实例
    llm=llm,
    # prompt模板
    prompt=prompt1,
    # 输出数据变量名
    output_key="english_product_name",
)
# Chain2 根据英文产品名，生成一段英文介绍文本
prompt2 = ChatPromptTemplate.from_template(
    "Based on the following product, give an introduction text about 100 words: {english_product_name}"
)
chain2 = LLMChain(
    llm=llm,
    prompt=prompt2,
    output_key="english_introduce"
)
# Chain3 找到产品名所属的语言
prompt3 = ChatPromptTemplate.from_template(
    "下列文本使用的语言是什么?: {product_name}"
)
chain3 = LLMChain(
    llm=llm,
    prompt=prompt3,
    output_key="language"
)
# Chain4 根据Chain2生成的英文介绍，使用产品名称原本的语言生成一段概述
prompt4 = ChatPromptTemplate.from_template(
    "使用语言类型为: {language} ，为下列文本写一段不多于50字的概述: {english_introduce}"
)
chain4 = LLMChain(
    llm=llm,
    prompt=prompt4,
    output_key="summary"
)
# 标准版的序列Chain,SequentialChain,其中每个chain都支持多个输入和输出，
# 根据chains中每个独立chain对象，和chains中的顺序，决定参数的传递，获取最终的输出结果
overall_chain = SequentialChain(
    chains=[chain1, chain2, chain3, chain4],
    input_variables=["product_name"],
    output_variables=["english_product_name", "english_introduce", "language", "summary"],
    verbose=True
)
product_name = "黄油啤酒"
res = overall_chain(product_name)
print(res)

运行过程和返回数据示例如图:

LLMRouteChain

以下是一段非常简单的 Python 代码。实现的主要是分支判断的作用。

if a == 1:
    print("我爱吃苹果")
elif b == 1:
    print("我爱吃香蕉")
else:
    print("我爱吃西瓜")

而 LLMRouteChain 的主要作用是能根据提示词的不同而选择不同的Chain进行执行。而实现这一需求，需要以下3个模块结合完成，也是MultiPromptChain的三个参数，以下内容是摘取的部分源码：
class MultiPromptChain(MultiRouteChain):
"""A multi-route chain that uses an LLM router chain to choose amongst prompts."""

    router_chain: RouterChain
"""Chain for deciding a destination chain and the input to it."""
    destination_chains: Mapping[str, LLMChain]
"""Map of name to candidate chains that inputs can be routed to."""
    default_chain: LLMChain
"""Default chain to use when router doesn't map input to one of the destinations."""

加载 Token，创建大模型实例。

# 加载个人的OpenAI Token
key = "open_ai_key"
# temperature用来设置大模型返回数据的随机性和创造性，较低的数值返回的数据就更贴近现实。
llm = ChatOpenAI(temperature=0.0, openai_api_key=key)

创建 destination_chains，目的是构造各个分支，以及其对应的Prompt。

# 2个角色prompt模板，作为请求处理的2个分支
physics_template = """你是一位非常聪明的物理学家，非常擅长回答物理相关的问题，\  
并且会以一种简洁易懂的方式对问题做出讲解。\  
当你无法回答问题的时候，就会主动承认无法回答问题。\  

以下是具体问题:  
{input}"""

math_template = """你是一位非常棒的数学家，非常擅长回答数学相关的问题。\  
你之所以这么棒，是因为你能够将难题拆解成它们的组成部分，\  
对组成部分分别作答后，再将它们组合起来最终成功的回答出最初的原始问题。\  

以下是具体问题:  
{input}"""

# 将角色prompt模板和对应的描述、名称组装成列表，方便遍历
prompt_info = [
    {
        "name": "物理学家",
        "description": "擅长回答物理方面的问题",
        "prompt_template": physics_template
    },
    {
        "name": "数学家",
        "description": "擅长回答数学方面的问题",
        "prompt_template": math_template
    },
]

# 名称和大模型Chain映射关系的字典
destination_chains = {}
# 根据prompt_info中的信息，创建对应的LLMChain实例，并放入映射字典中
for p_info in prompt_info:
    # destination_chains最终的结构为: {name: 对应的LLMChain实例}
    name = p_info["name"]
    prompt_template = p_info["prompt_template"]
    prompt = ChatPromptTemplate.from_template(template=prompt_template)
    # 给每个角色创建一个自己的大模型Chain实例
    chain = LLMChain(llm=llm, prompt=prompt)
    # 组装成字典，方便router根据逻辑选择分支之后，能够找到分支对应调用的Chain实例
    destination_chains[name] = chain

通过模板和大模型对象，生成LLMRouterChain，用于实现分支逻辑。而其能实现分支逻辑的原理，也是通过Prompt 实现，注意示例中的变量

MULTI_PROMPT_ROUTER_TEMPLATE

对应的内容以及信息：

# 生成destinations
destinations = [f"{p['name']}: {p['description']}" for p in prompt_info]
# 转换成多行字符串，每行一句对应关系
destinations_str = "\n".join(destinations)
# 真正实现 if 分支判断的地方
# 选择使用哪个人设模型进行处理的prompt模板
MULTI_PROMPT_ROUTER_TEMPLATE = """Given a raw text input to a \  
language model select the model prompt best suited for the input. \  
You will be given the names of the available prompts and a \  
description of what the prompt is best suited for. \  
You may also revise the original input if you think that revising\  
it will ultimately lead to a better response from the language model.  

<< FORMATTING >>  
Return a markdown code snippet with a JSON object formatted to look like:  
```json  {{{{    "destination": string \ name of the prompt to use or "DEFAULT"    "next_inputs": string \ a potentially modified version of the original input  }}}}  ```  
REMEMBER: "destination" MUST be one of the candidate prompt \  
names specified below OR it can be "DEFAULT" if the input is not\  
well suited for any of the candidate prompts.  
REMEMBER: "next_inputs" can just be the original input \  
if you don't think any modifications are needed.  

<< CANDIDATE PROMPTS >>  
{destinations}  

<< INPUT >>  
{{input}}  

<< OUTPUT (remember to include the ```json)>>"""

# 在router模版中先补充部分数据
router_template = MULTI_PROMPT_ROUTER_TEMPLATE.format(
    destinations=destinations_str
)
# 组装一个基础的prompt模板对象，通过更多的参数设置更多的信息
router_prompt = PromptTemplate(
    # 基础模板
    template=router_template,
    # 输入参数名称
    input_variables=["input"],
    # 输出数据解析器
    output_parser=RouterOutputParser(),
)
# 通过模板和大模型对象，生成LLMRouterChain，用于支持分支逻辑
router_chain = LLMRouterChain.from_llm(llm, router_prompt)

创建一个默认的LLMChain实例，作为上述匹配未命中时的默认调用目标，避免调用最终没有逻辑去处理的情况出现。

# 创建 else 场景
default_prompt = ChatPromptTemplate.from_template("{input}")
default_chain = LLMChain(llm=llm, prompt=default_prompt)

通过 MultiPromptChain 将以上三个模块整合，实现一个支持分支判断的Chain：

# 把 分支、 else 的情况 以及 做if判断的语句结合到一起
# 将多个chain组装成完整的chain对象，完成带有逻辑的请求链
chain = MultiPromptChain(
    router_chain=router_chain,
    destination_chains=destination_chains,
    default_chain=default_chain,
    verbose=True
)
# 提出一个物理问题
physics_res = chain.run("什么是黑体辐射?")
print(physics_res)
# 提出一个数学问题
math_res = chain.run("2的4次方是多少？")
print(math_res)

最终可以看到请求的时候，如果问题与之前给出的角色有相关，会使用到该角色的 llm 实例进行实际问题的解答。相关完整代码如下：

相关完整代码如下：

from langchain import LLMChain
from langchain.chat_models import ChatOpenAI
from langchain.chains.router import MultiPromptChain
from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser
from langchain.prompts import PromptTemplate, ChatPromptTemplate

# 加载个人的OpenAI Token
key = "open_ai_key"
# temperature用来设置大模型返回数据的随机性和创造性，较低的数值返回的数据就更贴近现实。
llm = ChatOpenAI(temperature=0.0, openai_api_key=key)

# 2个角色prompt模板，作为请求处理的2个分支
physics_template = """你是一位非常聪明的物理学家，非常擅长回答物理相关的问题，\  
并且会以一种简洁易懂的方式对问题做出讲解。\  
当你无法回答问题的时候，就会主动承认无法回答问题。\  

以下是具体问题:  
{input}"""

math_template = """你是一位非常棒的数学家，非常擅长回答数学相关的问题。\  
你之所以这么棒，是因为你能够将难题拆解成它们的组成部分，\  
对组成部分分别作答后，再将它们组合起来最终成功的回答出最初的原始问题。\  

以下是具体问题:  
{input}"""

# 将角色prompt模板和对应的描述、名称组装成列表，方便遍历
prompt_info = [
    {
        "name": "物理学家",
        "description": "擅长回答物理方面的问题",
        "prompt_template": physics_template
    },
    {
        "name": "数学家",
        "description": "擅长回答数学方面的问题",
        "prompt_template": math_template
    },
]

# 名称和大模型Chain映射关系的字典
destination_chains = {}
# 根据prompt_info中的信息，创建对应的LLMChain实例，并放入映射字典中
for p_info in prompt_info:
    # destination_chains最终的结构为: {name: 对应的LLMChain实例}
    name = p_info["name"]
    prompt_template = p_info["prompt_template"]
    prompt = ChatPromptTemplate.from_template(template=prompt_template)
    # 给每个角色创建一个自己的大模型Chain实例
    chain = LLMChain(llm=llm, prompt=prompt)
    # 组装成字典，方便router根据逻辑选择分支之后，能够找到分支对应调用的Chain实例
    destination_chains[name] = chain

# 生成destinations
destinations = [f"{p['name']}: {p['description']}" for p in prompt_info]
# 转换成多行字符串，每行一句对应关系
destinations_str = "\n".join(destinations)
# 真正实现 if 分支判断的地方
# 选择使用哪个人设模型进行处理的prompt模板
MULTI_PROMPT_ROUTER_TEMPLATE = """Given a raw text input to a \  
language model select the model prompt best suited for the input. \  
You will be given the names of the available prompts and a \  
description of what the prompt is best suited for. \  
You may also revise the original input if you think that revising\  
it will ultimately lead to a better response from the language model.  

<< FORMATTING >>  
Return a markdown code snippet with a JSON object formatted to look like:  
  ```json  {{{{    "destination": string \ name of the prompt to use or "DEFAULT"    "next_inputs": string \ a potentially modified version of the original input  }}}}  ```  
REMEMBER: "destination" MUST be one of the candidate prompt \  
names specified below OR it can be "DEFAULT" if the input is not\  
well suited for any of the candidate prompts.  
REMEMBER: "next_inputs" can just be the original input \  
if you don't think any modifications are needed.  

<< CANDIDATE PROMPTS >>  
{destinations}  

<< INPUT >>  
{{input}}  

<< OUTPUT (remember to include the ```json)>>"""

# 在router模版中先补充部分数据
router_template = MULTI_PROMPT_ROUTER_TEMPLATE.format(
    destinations=destinations_str
)
# 组装一个基础的prompt模板对象，通过更多的参数设置更多的信息
router_prompt = PromptTemplate(
    # 基础模板
    template=router_template,
    # 输入参数名称
    input_variables=["input"],
    # 输出数据解析器
    output_parser=RouterOutputParser(),
)
# 通过模板和大模型对象，生成LLMRouterChain，用于支持分支逻辑
router_chain = LLMRouterChain.from_llm(llm, router_prompt)

# 创建 else 场景
# 创建一个默认的LLMChain实例，作为上述匹配未命中时的默认调用目标，避免调用最终没有逻辑去处理的情况出现
# 上述的匹配规则可以看成一组if elif的逻辑匹配规则，default作为最后的else负责处理所有未命中的情况
default_prompt = ChatPromptTemplate.from_template("{input}")
default_chain = LLMChain(llm=llm, prompt=default_prompt)

# 把 分支、 else 的情况 以及 做if判断的语句结合到一起
# 将多个chain组装成完整的chain对象，完成带有逻辑的请求链
chain = MultiPromptChain(
    router_chain=router_chain,
    destination_chains=destination_chains,
    default_chain=default_chain,
    verbose=True
)

# 提出一个物理问题
physics_res = chain.run("什么是黑体辐射?")
print(physics_res)
# 提出一个数学问题
math_res = chain.run("2的4次方是多少？")
print(math_res)

后续在练习的过程中，尝试基于以上的示例代码再添加几条分支逻辑。

更多Python基础语法趣味学习视频，请点击！