AI大模型探索之路-应用篇3：Langchain框架Chain模块—集成化策略与实践_中文大模型langchain

目录

前言

一、概述

二、LLM（简单的链式操作）

三、Sequential（链路顺序操作）

四、Transformation（数据转换处理）

五、Route（链路路由处理）

总结

---

前言

Langchain框架的Chain模块应运而生，旨在简化复杂AI任务流程管理，提高模块化组件复用性，降低维护难度。随着AI技术的不断发展，处理语言模型所需的组件变得越来越复杂，开发者面临着如何有效地组织和管理这些组件的挑战。为了解决这个问题，Langchain框架引入了Chain模块，它通过模块化的设计思想，将多个功能组件组合成有序的执行流程，使得每一步骤都可以独立开发和测试，同时也便于整个流程的跟踪和调试。这种设计不仅提高了代码的复用性，还使得维护和扩展变得更加容易。

---

一、概述

Chain模块的主要功能是将多个步骤链接起来，处理复杂的业务逻辑。这些步骤可以包括数据处理、模型调用、结果分析等等。将这些步骤组织成链式结构，可以使得代码更加易于理解和管理，同时也能更好地反映业务流程。

主要包括以下几个方面：

1. 模块化组合：支持将不同的功能模块灵活组合，形成一个完整的业务流程；方便开发调试。

2. 低代码抽象：提供了一系列预设的组件和工具，如Prompt模板、数据连接等，简化了开发。

3. 强大的集成能力：支持与多种LLM进行交互，并提供了多种预设的Prompt模板，以便快速构建AI应用。

4. 灵活性和可扩展性：允许开发者根据具体需求定制、添加新的组件和工具，以满足不断变化的需求。

二、LLM（简单的链式操作）

使用Langchain表达式语言来实现最常见的链式操作类型。在所有的LLM应用中，最常见的链接方式是将提示模板与LLM和可选的输出解析器组合起来。

示例代码：

from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
 
from langchain.schema import StrOutputParser
 
prompt = PromptTemplate.from_template(
    "What is a good name for a company that makes {product}?"
)
 
chat_model = ChatOpenAI(model_name="gpt-3.5-turbo-0125")
## LCEL chain表达式语言， 利用管道符“|” 实现 链式调用
runnable = prompt | chat_model | StrOutputParser()
 
result = runnable.invoke({"product": "colorful socks"})
print(result)

三、Sequential（链路顺序操作）

Sequential展现了如何将多个语言模型串联起来，其中每个模型的输出成为下一个模型的输入。这种方法特别适用于那些需要逐步精炼或分层抽象的任务，例如文本摘要、情感分析等。

第一次调用语言模型（promt1|model|parse1)

第二次调用语言模型（promt2|model|parse2)

希望将第一次的parse1，作为第二次的promt2输入内容

示例代码：

from langchain.prompts import PromptTemplate
 
synopsis_prompt = PromptTemplate.from_template(
    """你是一位剧作家。给定一个剧目的标题，你的任务是为这个标题写一个剧情简介。
标题: {title}
剧作家: 这是上述剧目的剧情简介:"""
)
 
review_prompt = PromptTemplate.from_template(
    """您是《纽约时报》的戏剧评论家。根据剧情简介，您的工作是为该剧撰写一篇评论。.
剧情简介:
{synopsis}
上述剧目的《纽约时报》剧评家的评论:"""
)
 
 
 
#%%
from langchain_openai import ChatOpenAI
from langchain.schema import StrOutputParser
 
llm = ChatOpenAI(model_name="gpt-3.5-turbo-0125")
 
chain = (
    # 将第一次调用的结果放到参数synopsis中，再作为第二次review_prompt的输入
    {"synopsis": synopsis_prompt | llm | StrOutputParser()}
    | review_prompt
    | llm
    | StrOutputParser()
)
result = chain.invoke({"title": "日落时的海滩悲剧"})
print(result)
 
#%%
#如果我们还想要回到概要，我们可以这样做
from langchain.schema.runnable import RunnablePassthrough
 
synopsis_chain = synopsis_prompt | llm | StrOutputParser()
review_chain = review_prompt | llm | StrOutputParser()
# 定义一个字典{"synopsis": synopsis_chain}；
# 再给到RunnablePassthrough.assign() 方法，将其与 review_chain 合并；
# RunnablePassthrough.assign()返回一个新的对象给chain
chain = {"synopsis": synopsis_chain} | RunnablePassthrough.assign(review=review_chain)
# 执行invoke时会自动将上一条链的输出，作为下一条链的输入
result = chain.invoke({"title": "日落时的海滩悲剧"})
print(result)

RunnablePassthrough.assign 方法说明：
接受一个或多个键值对作为参数，其中键表示链路的名称，值表示链路本身。在这个例子中，我们传入了两个键值对："synopsis": synopsis_chain 和 "review": review_chain。当调用 RunnablePassthrough.assign() 方法时，它会创建一个新的 RunnablePassthrough 对象，并将传入的链路添加到该对象中。这个新对象可以像字典一样使用，通过链路名称来访问和执行相应的链路

四、Transformation（数据转换处理）

在Transformation中，我们关注于数据的中间处理。通过定义转换函数，可以在数据传递至下一模块前对其进行清洗、截断或其他形式的修改，从而优化模型的输入质量。

示例代码：

from langchain.prompts import PromptTemplate
 
prompt = PromptTemplate.from_template(
    """对下面内容进行总结摘要:
{output_text}
摘要:"""
)
#%%
with open("消失的她.txt",'r', encoding='utf-8') as f:
    state_of_the_union = f.read()
#%%
from langchain_openai import ChatOpenAI
from langchain.schema import StrOutputParser
 
llm = ChatOpenAI(model_name="gpt-3.5-turbo-0125")
 
runnable = (
    {"output_text": lambda text: "\n\n".join(text.split("\n\n")[:3])}
    | prompt
    | llm
    | StrOutputParser()
)
result = runnable.invoke(state_of_the_union)
print(result)

五、Route（链路路由处理）

Route引入了条件分支的概念，允许根据先前步骤的输出动态选择后续的处理路径。这增加了Chain的灵活性，使其能够应对更加复杂的业务场景。

例如：假设我们有两个针对不同类型问题进行优化的模板，并且我们希望根据用户输入选择模板。

示例代码：

from langchain.prompts import PromptTemplate
 
## 物理学家的模板
physics_template = """You are a very smart physics professor. \
You are great at answering questions about physics in a concise and easy to understand manner. \
When you don't know the answer to a question you admit that you don't know.
Here is a question:
{input}"""
physics_prompt = PromptTemplate.from_template(physics_template)
 
## 数学家的模板
math_template = """You are a very good mathematician. You are great at answering math questions. \
You are so good because you are able to break down hard problems into their component parts, \
answer the component parts, and then put them together to answer the broader question.
Here is a question:
{input}"""
math_prompt = PromptTemplate.from_template(math_template)
#%%
from langchain_openai import ChatOpenAI
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnableBranch
 
## 其他通用问题的模块
general_prompt = PromptTemplate.from_template(
    "You are a helpful assistant. Answer the question as accurately as you can.\n\n{input}"
)
 
##路由分支
prompt_branch = RunnableBranch(
    (lambda x: x["topic"] == "math", math_prompt),
    (lambda x: x["topic"] == "physics", physics_prompt),
    general_prompt,
)
#%%
from typing import Literal
 
from langchain.pydantic_v1 import BaseModel
from langchain.output_parsers.openai_functions import PydanticAttrOutputFunctionsParser
from langchain.utils.openai_functions import convert_pydantic_to_openai_function
 
## 对用户问题分类，定义函数
class TopicClassifier(BaseModel):
    "Classify the topic of the user question"
 
    topic: Literal["math", "physics", "general"]
    "The topic of the user question. One of 'math', 'physics' or 'general'."
 
## 转化为OpenAI function函数
classifier_function = convert_pydantic_to_openai_function(TopicClassifier)
 
## 创建大模型
llm = ChatOpenAI(model_name="gpt-3.5-turbo").bind(
    functions=[classifier_function], function_call={"name": "TopicClassifier"}
)
## 定义对输出进行解析
## 输出的对象TopicClassifier的属性值topic
parser = PydanticAttrOutputFunctionsParser(
    pydantic_schema=TopicClassifier, attr_name="topic"
)
## 基于大模型对输出进行解析
classifier_chain = llm | parser
#%%
from operator import itemgetter
 
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnablePassthrough
 
 
final_chain = (
    RunnablePassthrough.assign(topic=itemgetter("input") | classifier_chain)
    | prompt_branch
    | ChatOpenAI(model_name="gpt-3.5-turbo")
    | StrOutputParser()
)
#%%
## 什么是大于40的第一个质数，使得这个质数加一可被3整除？
result = final_chain.invoke(
    {
        "input": "What is the first prime number greater than 40 such that one plus the prime number is divisible by 3?"
    }
)
 
print(result)

---

总结

Langchain框架的Chain模块集为构建和部署大型语言模型提供了一种高度模块化和可扩展的方法。通过组合不同的模块，开发者可以构建出既强大又灵活的应用，满足各种复杂场景的需求。无论是通过简单的链式操作连接基本组件，还是通过顺序、转换和路由操作处理更加复杂的工作流，Chain模块集都表现出了其卓越的适应性和实用性。

探索未知，分享所知；点击关注，码路同行，寻道人生！