一、chains链的基础使用1. 介绍chain链尤其是 | 管道链的核心工作原理将组件串联上一个组件的输出作为下一个组件的输入链式调用的核心价值实现数据的自动化流转与组件的协同工作chain prompt_template | model核心前提Runnable子类对象才能入链以及Callable、Mapping接口子类对象也可以加入目前我们学到的组件均是Runnable接口的子类继承关系如下2. 原始操作①ChatPromptTemplate.from_messages包含占位符的提示词模板②history_data占位符内容③chat_prompt_template.invoke({history: history_data}).to_string()根据占位符拿到完整的提示词④ChatOllama(modelgemma4:12b)拿到模型对象⑤model.invoke(prompt_template)把完整的提示词给模型拿到模型的结果res通过.content输出fromlangchain_ollamaimportChatOllamafromlangchain_core.runnables.baseimportRunnableSerializable chat_prompt_templateChatPromptTemplate.from_messages([(system,你是一个民谣歌手),MessagesPlaceholder(history),(human,再来一首民谣)])history_data[(human,来首伤感民谣),(ai,我知道那些夏天就像青春一样回不来我也不会再对谁满怀期待),(human,很好听再来一首),(ai,青春又醉倒在岌岌无名的坏靠嬉笑来度过这破碎的年代)]prompt_templatechat_prompt_template.invoke({history:history_data}).to_string()#print(prompt_template)modelChatOllama(modelgemma4:12b)resmodel.invoke(prompt_template)print(res.content)3. chain链操作chain: RunnableSerializable chat_prompt_template | model代码流程fromlangchain_core.promptsimportMessagesPlaceholder,ChatPromptTemplatefromlangchain_ollamaimportChatOllamafromlangchain_core.runnables.baseimportRunnableSerializable chat_prompt_templateChatPromptTemplate.from_messages([(system,你是一个民谣歌手),MessagesPlaceholder(history),(human,再来一首民谣)])history_data[(human,来首伤感民谣),(ai,我知道那些夏天就像青春一样回不来我也不会再对谁满怀期待),(human,很好听再来一首),(ai,青春又醉倒在岌岌无名的坏靠嬉笑来度过这破碎的年代)]modelChatOllama(modelgemma4:12b)chain:RunnableSerializablechat_prompt_template|model# Runnable接口invoke执行reschain.invoke({history:history_data})print(res.content)# Runnable接口stream执行# for chunk in chain.stream({history: history_data}):# print(chunk.content, end, flushTrue)4. 小结LangChain中链是一种将各个组件串联在一起按顺序执行前一个组件的输出作为下一个组件的输入可以通过|来让各个组件形成链成链的各个组件需要是Runnable接口的子类形成的链是RunnableSerializable对象Runnable接口子类可通过链调用invoke或stream触发整个链条的执行chain chat_prompt_template | model二、或运算符的重写chain chat_prompt_template | model在语法上使用了|运算符的重写Python中的运算符如、|、等的行为由类的魔法方法决定a b本质上调用的是a.__add__(b)a | b本质上调用的是a.__or__(b)只需要实现类的__or__方法即可对|符号的功能进行重写任务让a | b | c的代码得到一个自定义的类对象类似列表[a, b, c]调用run方法依次输出a、b、c重写|即__or__方法classTest(object):def__init__(self,name):# 创建对象时自动跑这里self.namename# 把传进来的字符串存成自己的名字def__or__(self,other):# 当写 a | b 时自动跑这里returnMySequence(self,other)# 返回一个“新容器”def__str__(self):# 当你 print(对象) 时自动跑这里returnself.name# 直接返回名字classMySequence(object):def__init__(self,*args):# *args 把所有传进来的参数打包成元组self.sequence[]# 准备一个空列表forarginargs:# 把每个参数依次放进列表self.sequence.append(arg)def__or__(self,other):# 当写 容器 | c 时自动跑这里self.sequence.append(other)# 把 c 追加到同一个列表里returnself# 把“自己”返回方便继续链式调用defrun(self):# 手动调用时跑这里foriinself.sequence:print(i)if__name____main__:aTest(a)bTest(b)cTest(c)dTest(d)# 第 1 步a | b# - a.__or__(b)# - 生成 MySequence(a, b) # 列表里现在有 [a, b]# - 叫它 tmp1# 第 2 步tmp1 | c# - tmp1.__or__(c)# - tmp1 列表变成 [a, b, c]# - 返回 tmp1 自己# - 叫它 tmp2其实还是同一个对象# 第 3 步tmp2 | d# - tmp2.__or__(d)# - 列表变成 [a, b, c, d]# - 返回 tmp2 自己# - 最终赋值给 f# 所以f 里装着[a, b, c, d] 四个Test对象。fa|b|c|d f.run()print(type(f))运行结果三、Runnable接口LangChain中的绝大多数核心组件都继承了Runnable抽象基类langchain_core.runnables.basechain prompt | modelchain变量是RunnableSequenceRunnableSerializable子类类型而得到这个类型的原因是Runnable基类内部对__or__魔术方法的改写同时在后面继续使用|添加新的组件依旧会得到RunnableSequence这就是链的基础架构fromlangchain_core.promptsimportPromptTemplatefromlangchain_ollamaimportOllamaLLM promptPromptTemplate.from_template(你是一个Beyond乐队的粉丝)modelOllamaLLM(modelgemma4:12)chainprompt|modelprint(type(chain))# class langchain_core.runnables.base.RunnableSequencefrom langchain_core.prompts import PromptTemplate进入PromptTemplate类中其继承class PromptTemplate(StringPromptTemplate):进入StringPromptTemplate类中其继承class StringPromptTemplate(BasePromptTemplate[str], ABC):进入BasePromptTemplate类中其继承RunnableSerializable[dict[str, Any], PromptValue], ABC, Generic[FormatOutputType]进入RunnableSerializable类中其继承class RunnableSerializable(Serializable, Runnable[Input, Output]):进入Runnable类中有个__or__抽象方法其返回类型为RunnableSequence故无论链式怎么追加最终的返回值类型都是RunnableSequence