AI大模型开发工程师

008 LangChain之Chains模块

1 Chain模块核心知识

组合常用的模块

• LLM：最常见的链式操作类型
• SequentialChain：串联式调用语言模型链
• RouterChain：实现条件判断的大模型调用
• Transformation：数据传递过程中进行数据处理

2 Chain模块代码实战

LLM

• 在任何LLM应用中，最常见的链接方式是将提示模板与LLM和可选的输出解析器组合起来。
• 推荐的方法是使用LangChain表达式语言来实现。我们也继续支持传统的LLMChain，这是一个用于组合这三个组件的单个类。

from langchain.prompts import PromptTemplate
from langchain.chat_models 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="gpt-3.5-turbo")
## 链式调用
runnable = prompt | chat_model | StrOutputParser()

result = runnable.invoke({"product": "colorful socks"})
print(result)

以前的 API（已不推荐使用）

from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langchain.chains import LLMChain

prompt_template = "What is a good name for a company that makes {product}?"

llm = ChatOpenAI(model_name="gpt-3.5-turbo")

llm_chain = LLMChain(llm=llm, prompt=PromptTemplate.from_template(prompt_template))

result = llm_chain("colorful socks")
print(result)

Sequential

• 在调用语言模型之后，下一步是对语言模型进行一系列的调用。当您希望将一次调用的输出作为另一次调用的输入时，这是特别有用的。
• 推荐的方法是使用LangChain表达语言。遗留的方法是使用SequentialChain，为了向后兼容性，我们在此继续记录它。

示例：假设我们想创建一个链，首先创建剧情简介，然后根据简介生成一篇剧评。

from langchain.prompts import PromptTemplate

synopsis_prompt = PromptTemplate.from_template(
    """你是一位剧作家。给定一个剧目的标题，你的任务是为这个标题写一个剧情简介。

标题: {title}
剧作家: 这是上述剧目的剧情简介:"""
)

review_prompt = PromptTemplate.from_template(
    """您是《纽约时报》的戏剧评论家。根据剧情简介，您的工作是为该剧撰写一篇评论。.

剧情简介:
{synopsis}
上述剧目的《纽约时报》剧评家的评论:"""
)

from langchain.chat_models import ChatOpenAI
from langchain.schema import StrOutputParser

llm = ChatOpenAI(model="gpt-4")

chain = (
    {"synopsis": synopsis_prompt | llm | StrOutputParser()}
    | review_prompt
    | llm
    | StrOutputParser()
)
result = chain.invoke({"title": "日落时的海滩悲剧"})
print(result)

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.chat_models import ChatOpenAI
from langchain.schema import StrOutputParser

llm = ChatOpenAI(model_name="gpt-3.5-turbo")

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)

Router

• 路由允许您创建非确定性链，其中前一步的输出定义下一步。路由有助于在与LLMs的交互中提供结构和一致性。

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

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.chat_models 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_core.utils.function_calling 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="gpt-3.5-turbo").bind(
    functions=[classifier_function], function_call={"name": "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)