AI--LangChain的链

LangChain的链

1.[扩展]或运算符的重写

  在前文中，我们有提到一串使用了链的代码：

chain = chat_template | model

  其本质是在语法上使用了丨运算符的重写，在python中，运算符（如 +、|）的行为由类的魔法方法决定的（什么是魔法方法可以查看Python中的总结），例如图：

• a + b本质是调用的 a.add(b)
• a | b本质是调用的 a.or(b)

  而我们只需要自行实现的__or__方法，即可对|符号的功能进行重写

class Test(object):
    def __init__(self, name):
        self.name = name
    def __or__(self, other):
        return MyStream(self, other)
    
    def __str__(self):
        return self.name


class MyStream(object):
    def __init__(self, *args):
        self.seqence = []
        for item in args:
            self.seqence.append(item)
    
    def __or__(self, other):
        self.seqence.append(other)
        return self
    
    def run(self):
        for item in self.seqence:
            print(item)

if __name__ == "__main__":
    t1 = Test("t1")
    t2 = Test("t2")
    t3 = Test("t3")

    t4 = t1 | t2 | t3
    t4.run()
    print(type(t4))

  这个功能的主要目的就是自定义运算符在我们代码类型中的行为，而不是使用默认的 “按位或 / 逻辑或”。

  

2.Runnable接口

  前面我们提到Chain变量是RunnableSequence类型，而得到这个类型的原因就是Runnable基类内部对__or__魔术方法进行了改写，同时，在后面继续使用|添加新的组件，依旧会得到RunnableSequence，这就是链的基础架构

from langchain_core.prompts import PromptTemplate
from langchain_community.llms.tongyi import Tongyi

prompt_template = PromptTemplate.from_template("你是一个AI助手")
model = Tongyi(model="qwen-max")

chain = prompt_template | model
print(type(chain))

res = chain.invoke(input={})
print(res)

  

3.StrOutputParser字符串输出解析器

  当我们在使用Chain链时，可能会遇到输入类型的问题，例如：

chain = prompt | model |model

  这个指令执行是会报错的，错误的主要原因便是prompt输出的结果是PromptValue类型，输入给了model，而model的输出类型是AIMessage，这里就需要做到一个类型转换，而LangChain有没有内置的转换解析器呢，有的，兄弟有的，StrOutputParser是LangChain内置的简单字符串解析器

  他可以将AImessage解析为简单的字符串，符合了模型invoke方法的要求（可传入字符串，不接受AIMessage类型），同时也是Runnable接口的子类，因此也可以接入到链中，最终就可以修改为：

parser = StrOutputParser()
chain = prompt | model | parser |model

  示例代码：

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi

prompt = PromptTemplate.from_template("我的邻居姓{name}，刚生了个{age}岁的{gender}，请起名，只需要告诉我名字即可无需其他内容")
model = ChatTongyi(model="qwen-max")

parser = StrOutputParser()
chain = prompt | model | parser | model | parser

res = chain.invoke(input={"name": "张", "age": 1, "gender": "男"})
print(res)

  

4.JsonOutputParser字典输出解析器

在前面的StrOutputParser小结中，我们完成了字符串的转换去构建多模型链，但是这种做法并不规范，因为上一个模型的输出只是简单进行了字符的处理，并没有进行额外的处理就输入到了下一个模型，正常情况下，我们的处理路径应该如下：

  前面提到提示词模版是需要字典类型的输入，所以我们需要将模型输出的AIMessage类型转化为字典后，注入到第二个提示词模版中，形成新的提示词（PromptValue对象），这里就需要使用到我们的JsonOutputParser

from os import name
from langchain_core.output_parsers import StrOutputParser
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi

model = ChatTongyi(model="qwen-max")

str_parser = StrOutputParser()
json_parser = JsonOutputParser()

first_prompt = PromptTemplate.from_template(
    "我的邻居姓{lastname}，刚生了个{age}岁的{gender}，请起名，并封装成Jsom格式返回"
    "要求key是name，value是新名字，请严格遵守格式要求"  
)

second_prompt = PromptTemplate.from_template(
    "姓名{name},请帮我解析含义"
)

chain1 = first_prompt | model | json_parser

res1 = chain1.invoke({"lastname": "张", "age": 1, "gender": "女"})
print(res1)

chain = chain1 | second_prompt | model | str_parser

res2 = chain.invoke({"lastname": "张", "age": 1, "gender": "女"})
print(res2)

 

5.RunnambeLambda 函数对象

  前一小节，我们学习了JsonOutputParser，完成了多模型执行链条的构建，但是除了JsonOutputParser这类固定功能的解析器之外，我们也可以通过Lambda匿名函数来完成自定义的数据转换，从而更加自由，需要基于RunnambeLambda进行开发，该类是LangChain内置的，将普通函数等转换为Runnable接口实例，从而方便自定义函数加入到chain中

from os import name
from langchain_core.output_parsers import StrOutputParser
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.runnables import RunnableLambda
from langchain_core.prompts import PromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi

model = ChatTongyi(model="qwen-max")

str_parser = StrOutputParser()
my_func = RunnableLambda(lambda ai_msg: {"name": ai_msg.content})

first_prompt = PromptTemplate.from_template(
    "我的邻居姓{lastname}，刚生了个{age}岁的{gender}，请起名，只需要告知我名字，不需要额外的内容"
)

second_prompt = PromptTemplate.from_template(
    "姓名{name},请帮我解析含义"
)

chain1 = first_prompt | model | my_func
res1 = chain1.invoke({"lastname": "张", "age": 1, "gender": "女"})
print(res1)

chain = chain1 | second_prompt | model | str_parser

res = chain.invoke({"lastname": "张", "age": 1, "gender": "女"})
print(res)

  除此之外，函数也是可以直接入链的，这是因为在链中实现__or__的时候，支持Callable接口的实例

from os import name
from langchain_core.output_parsers import StrOutputParser
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.runnables import RunnableLambda
from langchain_core.prompts import PromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi

model = ChatTongyi(model="qwen-max")

str_parser = StrOutputParser()
#my_func = RunnableLambda(lambda ai_msg: {"name": ai_msg.content})

first_prompt = PromptTemplate.from_template(
    "我的邻居姓{lastname}，刚生了个{age}岁的{gender}，请起名，只需要告知我名字，不需要额外的内容"
)

second_prompt = PromptTemplate.from_template(
    "姓名{name},请帮我解析含义"
)

chain1 = first_prompt | model | (lambda ai_msg: {"name": ai_msg.content})
res1 = chain1.invoke({"lastname": "张", "age": 1, "gender": "女"})
print(res1)