AI的学习之路_6_chain链
什么是链
将组件串联，上一个组件的输出作为下一个组件的输入 是LangChain链（尤其是 | 管道链）的核心工作原理
Runnable的派生类才能入链
拼接一下
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi
chart_prompt_template = ChatPromptTemplate.from_messages(
[
("system", "你是一个小学算术老师"), # 系统设定
MessagesPlaceholder("history"), # 历史对话，等待动态注入
("human", "所以加法结果是多少啊") # 最后问的结果
]
)
# 历史聊天记录，这里可以从其他地方获取，数据库，文件什么的，都可以
history_data = [
("human", "给出一个随机数"),
("ai", "17"),
("human", "再给一个"),
("ai", "23"),
]
model = ChatTongyi(
model="qwen-max"
)
# 使用链进行拼接起来
chain = chart_prompt_template | model
# 用链调用invoke
res = chain.invoke({"history": history_data})
print(res.content)
在执行invoke时，数据流向是这样的：
- 输入字典 {"history": history_data} 先被传给 chart_prompt_template，模板会把 history_data 填充到预设的提示词结构中，生成一段完整的提示词文本。
- 生成的提示词文本会自动传递给后面的 model，模型接收这段文本并开始推理。
- 模型推理完成后，返回一个包含回答内容的对象（通常是 AIMessage 类型），最后通过 res.content 提取出纯文本回答。
其中，|已经将前一部分组件拼接成了一个可以执行的整体，invoke触发了执行整体的开关
再看一下流式输出
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi
chart_prompt_template = ChatPromptTemplate.from_messages(
[
("system", "你是一个小学算术老师"), # 系统设定
MessagesPlaceholder("history"), # 历史对话，等待动态注入
("human", "所以加法结果是多少啊") # 最后问的结果
]
)
# 历史聊天记录，这里可以从其他地方获取，数据库，文件什么的，都可以
history_data = [
("human", "给出一个随机数"),
("ai", "17"),
("human", "再给一个"),
("ai", "23"),
]
model = ChatTongyi(
model="qwen-max"
)
# 使用链进行拼接起来
chain = chart_prompt_template | model
# 用链调用invoke
# res = chain.invoke({"history": history_data})
#
# print(res.content)
# stream流式输出
for chunk in chain.stream({"history": history_data}):
print(chunk.content, end="", flush=True)
| 运算符
|运算符的本质是__or__运算符
这里重写一下__or__运算符帮助理解一下
class Test(object):
def __init__(self, name):
self.name = name
def __str__(self):
return f"Text:{self.name}"
def __or__(self, other):
return Test2(self, other)
class Test2(object):
def __init__(self, *args):
self.sequence = []
for arg in args:
self.sequence.append(arg)
def __or__(self, other):
self.sequence.append(other)
return self
def run(self):
for arg in self.sequence:
print(arg)
if __name__ == '__main__':
a = Test("a")
b = Test("b")
c = Test("c")
d = a | b | c
d.run()
print(type(d))
a b c都是Text类型的对象，其中
调用的是a的|（也就是__or__）运算符
运算之后得到一个Text2类型的变量，这里可以假定为O，实际上可以理解为匿名对象
之后，就是O | c了，这里调用的是O的|运算符，也就是说实际调用的是Text2里面重写的__or__函数
之后run函数无疑调用的是Text2里面的，进行打印则会调用abc的实际类型里面的__str__函数
所以最终会得到这个结果
如果不重写__str__函数的话，就会直接打印内存地址了
Runnable接口
看一下Runnable的__or__源码
def __or__(
self,
other: Runnable[Any, Other]
| Callable[[Iterator[Any]], Iterator[Other]]
| Callable[[AsyncIterator[Any]], AsyncIterator[Other]]
| Callable[[Any], Other]
| Mapping[str, Runnable[Any, Other] | Callable[[Any], Other] | Any],
) -> RunnableSerializable[Input, Other]:
"""Runnable "or" operator.
Compose this `Runnable` with another object to create a
`RunnableSequence`.
Args:
other: Another `Runnable` or a `Runnable`-like object.
Returns:
A new `Runnable`.
"""
return RunnableSequence(self, coerce_to_runnable(other))
返回的是RunnableSerializable对象，这是Runnable的序列化对象
观察如下代码
from langchain_core.prompts import PromptTemplate
from langchain_community.llms.tongyi import Tongyi
prompt = PromptTemplate.from_template("你是一个AI助手")
model = Tongyi(
model = "qwen-max"
)
chain = prompt | model # 这里的后面无论追加再多的链，返回的都是RunnableSerializable类型的对象
print(type(chain))
StrOutputParser解析器
我们尝试将模型输出作为输入
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi
chart_prompt_template = ChatPromptTemplate.from_messages(
[
("system", "你是一个小学算术老师"), # 系统设定
MessagesPlaceholder("history"), # 历史对话，等待动态注入
("human", "所以加法结果是多少啊") # 最后问的结果
]
)
# 历史聊天记录，这里可以从其他地方获取，数据库，文件什么的，都可以
history_data = [
("human", "给出一个随机数"),
("ai", "17"),
("human", "再给一个"),
("ai", "23"),
]
model = ChatTongyi(
model="qwen-max"
)
# 使用链进行拼接起来
chain = chart_prompt_template | model | model # 这里将模型的输出作为输入
# 用链调用invoke
res = chain.invoke({"history": history_data})
但是看到结果会发现
AIMessage无法作为链的输入，因为他不是Runnable的派生类
此时字符串输出解析器StrOutputParser就起到了作用
可以吧AIMessage可以解析为简单的字符串
StrOutputParser同时也是Runnable的派生类
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_community.chat_models.tongyi import ChatTongyi
chart_prompt_template = ChatPromptTemplate.from_messages(
[
("system", "你是一个小学算术老师"), # 系统设定
MessagesPlaceholder("history"), # 历史对话，等待动态注入
("human", "所以加法结果是多少啊") # 最后问的结果
]
)
# 历史聊天记录，这里可以从其他地方获取，数据库，文件什么的，都可以
history_data = [
("human", "给出一个随机数"),
("ai", "17"),
("human", "再给一个"),
("ai", "23"),
]
model = ChatTongyi(
model="qwen-max"
)
parser = StrOutputParser()
# 使用链进行拼接起来
chain = chart_prompt_template | model | parser | model
# 这里也可以写成chain = chart_prompt_template | model | parser | model | parser
# 下面就不用content了，因为StrOutputParser返回的是str
# 用链调用invoke
res = chain.invoke({"history": history_data})
print(res.content)
这样就成功将AI的回答作为对AI的提问了
此时的res是第二个model回复了
而且res也是AIMessage类型的
JsonOutputParser&多模型执行链
AIMessage->Dict(JSON)
看一下代码
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate
str_parser = StrOutputParser()
json_parser = JsonOutputParser()
model = ChatTongyi(
model = "qwen-max"
)
first_prompt = PromptTemplate.from_template(
"{city}属于哪个国家，以json的格式返回给我，要求key是cou， value是所属的国家，严格按照格式进行输出"
)
second_prompt = PromptTemplate.from_template(
"{cou}属于那个大洲，仅回复我大洲名称即可，无需额外的任何内容"
)
chain = first_prompt | model | json_parser | second_prompt | model | str_parser
res = chain.invoke({"city": "长春"}) # stream({"city": "长春"})
print(res)
其程序流向如图所示
RunnambeLambda
处理利用JsonOutputParser这种解析器外，我们也可以自己实现函数来实现解析
RunnambeLambda就是这个作用
看代码
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableLambda
str_parser = StrOutputParser()
my_parser = RunnableLambda(lambda ai_message : { # ai_message是传入的参数
"cou": ai_message.content # 返回字典{"cou": "..."}
})
model = ChatTongyi(
model = "qwen-max"
)
first_prompt = PromptTemplate.from_template(
"{city}属于哪个国家，以json的格式返回给我，要求key是cou， value是所属的国家，严格按照格式进行输出"
)
second_prompt = PromptTemplate.from_template(
"{cou}属于那个大洲，仅回复我大洲名称即可，无需额外的任何内容"
)
chain = first_prompt | model | my_parser | second_prompt | model | str_parser
# chain = first_prompt | model | (lambda ai_message : {"cou": ai_message.content}) | second_prompt | model | str_parser 也是可以的
res = chain.invoke({"city": "长春"})
print(res)
浙公网安备 33010602011771号