一、链是什么「将组件串联上一个组件的输出作为下一个组件的输入」是 LangChain 链尤其是|管道链的核心工作原理也是链式调用的核心价值实现数据的自动化流转与组件的协同工作。chain prompt_template | model1.1 谁能入链核心前提即 Runnable 子类对象才能入链以及 Callable、Mapping 接口子类对象也可加入。目前学到的组件均是 Runnable 接口的子类。1.2 链执行起来什么样通过|链接提示词模板对象和模型对象返回值 chain 是RunnableSerializable对象它是 Runnable 接口的直接子类也是绝大多数组件的父类通过invoke或stream进行阻塞执行或流式执行组成的链上一个组件的输出作为下一个组件的输入from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_community.chat_models.tongyi import ChatTongyi from langchain_core.runnables.base import RunnableSerializable chat_prompt_template ChatPromptTemplate.from_messages( [ (system, 你是一个边塞诗人可以作诗。), MessagesPlaceholder(history), (human, 请再来一首唐诗无需额外输出), ] ) history_data [ (human, 你来写一个唐诗), (ai, 床前明月光疑是地上霜举头望明月低头思故乡), (human, 好诗再来一个), (ai, 锄禾日当午汗滴禾下锄谁知盘中餐粒粒皆辛苦), ] model ChatTongyi(modelqwen3-max) chain: RunnableSerializable chat_prompt_template | model print(type(chain)) # Runnable接口invoke执行 res chain.invoke({history: history_data}) print(res.content) # Runnable接口stream执行 for chunk in chain.stream({history: history_data}): print(chunk.content, end, flushTrue)1.3 链小结链是将各个组件串联在一起按顺序执行前一个组件的输出作为下一个组件的输入通过|符号让各个组件形成链成链的各个组件需是 Runnable 接口的子类形成的链是 RunnableSerializable 对象可通过链调用invoke或stream触发整个链条的执行二、|运算符为什么能组链2.1 本质是__or__chain chat_prompt_template | model在语法上使用了|运算符的重写。在 Python 中运算符行为由类的魔法方法决定a b本质调用a.__add__(b)a | b本质调用a.__or__(b)自行实现__or__即可重写|。2.2 课件示例a | b | cclass Test(object): def __init__(self, name): self.name name def __str__(self): return fTest({self.name}) def __or__(self, other): return MySequence(self, other) class MySequence(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))2.3 落到 LangChain 上chain prompt | model得到的是RunnableSequenceRunnableSerializable 子类原因就是 Runnable 基类内部对__or__的改写。后面继续用|加组件依旧得到 RunnableSequence——这就是链的基础架构。三、StrOutputParser为什么 prompt | model | model 会报错3.1 复现需求第一次模型的输出再拿去第二次询问模型。from langchain_core.prompts import PromptTemplate from langchain_community.chat_models.tongyi import ChatTongyi model ChatTongyi(modelqwen3-max) prompt PromptTemplate.from_template( 我邻居姓{lastname}, 刚生了{gender}请起名仅告知名字无需其它内容 ) chain prompt | model | model res chain.invoke({lastname: 张, gender: 女儿}) print(res.content)运行报错ValueError: Invalid input type class langchain_core.messages.ai.AIMessage. Must be a PromptValue, str, or list of BaseMessages.3.2 原因prompt 的结果是PromptValue输入给了 model —— 这一段是合法的model 的输出是AIMessage模型invoke的 input 类型是LanguageModelInput PromptValue | str | Sequence[MessageLikeRepresentation]不接收 AIMessage3.3 用 StrOutputParser 做类型转换LangChain 内置StrOutputParser字符串输出解析器把 AIMessage 解析为简单字符串且它是 Runnable 子类可以加入链。parser StrOutputParser() chain prompt | model | parser | model小结StrOutputParser 是内置的简单字符串解析器可以将 AIMessage 转换为基础字符串可以加入 chain。四、JsonOutputParser 与标准多模型链4.1 更标准的处理逻辑prompt | model | parser | model并不标准上一个模型的输出没有被处理成「下一个提示词模板」所需的输入。正常逻辑invoke / stream 初始输入 → 提示词模板 → 模型 → 数据处理 → 提示词模板 → 模型 → 解析器 → 结果即上一个模型的输出应作为提示词模板的输入构建下一个提示词用来二次调用模型。模型输出AIMessage提示词模板invoke要求输入dict输出PromptValueStrOutputParserAIMessage → str不够JsonOutputParserAIMessage → Dict(JSON)4.2 完整代码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 str_parser StrOutputParser() json_parser JsonOutputParser() model ChatTongyi(modelqwen3-max) first_prompt PromptTemplate.from_template( 我邻居姓{lastname}刚生了{gender}请起名并封装到JSON格式返回给我 要求key是namevalue就是起的名字。请严格遵守格式要求 ) second_prompt PromptTemplate.from_template( 姓名{name}请帮我解析含义。 ) chain first_prompt | model | json_parser | second_prompt | model | str_parser res: str chain.invoke({lastname: 张, gender: 女儿}) print(res) print(type(res))4.3 输入输出必须对齐组件输入输出模型PromptValue 或字符串或序列BaseMessage、list、tuple、str、dictAIMessage提示词模板字典PromptValueStrOutputParserAIMessagestrJsonOutputParserAIMessagedict标准链类型流字典 → first_prompt → PromptValue → model → AIMessage → json_parser → 字典 → second_prompt → PromptValue → model → AIMessage → str_parser → 字符串五、RunnableLambda自定义函数入链5.1 语法除了固定功能的解析器也可以自己编写 Lambda 完成自定义逻辑。RunnableLambda把普通函数转换为 Runnable 实例方便自定义函数加入 chain。语法RunnableLambda(函数对象或 lambda 匿名函数)from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableLambda from langchain_core.prompts import PromptTemplate from langchain_community.chat_models.tongyi import ChatTongyi str_parser StrOutputParser() my_func RunnableLambda(lambda ai_msg: {name: ai_msg.content}) model ChatTongyi(modelqwen3-max) first_prompt PromptTemplate.from_template( 我邻居姓{lastname}刚生了{gender}请起名仅告知我名字不要额外信息 ) second_prompt PromptTemplate.from_template( 姓名{name}请帮我解析含义。 ) chain first_prompt | model | my_func | second_prompt | model | str_parser res: str chain.invoke({lastname: 张, gender: 女儿}) print(res) print(type(res))5.2 函数也可以直接入链chain first_prompt | model | (lambda ai_msg: {name: ai_msg.content}) | second_prompt | model | str_parser因为 Runnable 在实现__or__时支持 Callable函数就是 Callable 实例本质是将函数自动转换为 RunnableLambda。小结将函数封装入RunnableLambda它是 Runnable 接口实例可以直接入链直接将函数入链函数会自动转换为 RunnableLambda 对象六、临时记忆InMemoryChatMessageHistory如果想要封装历史记录除了自行维护历史消息外也可以借助 LangChain 内置的历史记录功能帮助模型在有历史记忆的情况下回答。6.1 两个关键类基于RunnableWithMessageHistory在原有链的基础上创建带有历史记录功能的新链新 Runnable 实例基于InMemoryChatMessageHistory为历史记录提供内存存储临时用from langchain_core.runnables.history import RunnableWithMessageHistory conversation_chain RunnableWithMessageHistory( some_chain, # 被附加历史消息的 Runnable通常是 chain None, # 获取指定会话 ID 的历史会话的函数 input_messages_keyinput, # 用户输入在模板中的占位符 history_messages_keychat_history # 历史消息在模板中的占位符 ) chat_history_store {} # 存放多个会话 ID 所对应的历史会话记录 def get_history(session_id): if session_id not in chat_history_store: chat_history_store[session_id] InMemoryChatMessageHistory() return chat_history_store[session_id]6.2 完整代码from langchain_community.chat_models.tongyi import ChatTongyi from langchain_core.chat_history import InMemoryChatMessageHistory from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import PromptTemplate from langchain_core.runnables.history import RunnableWithMessageHistory def print_prompt(full_prompt): print( * 20, full_prompt.to_string(), * 20) return full_prompt model ChatTongyi(modelqwen3-max) prompt PromptTemplate.from_template( 你需要根据对话历史回应用户问题。对话历史{chat_history}。用户当前输入{input} 请给出回应 ) base_chain prompt | print_prompt | model | StrOutputParser() chat_history_store {} def get_history(session_id): if session_id not in chat_history_store: chat_history_store[session_id] InMemoryChatMessageHistory() return chat_history_store[session_id] conversation_chain RunnableWithMessageHistory( base_chain, get_history, input_messages_keyinput, history_messages_keychat_history ) if __name__ __main__: session_config {configurable: {session_id: user_001}} print(conversation_chain.invoke({input: 小明有一只猫}, session_config)) print(conversation_chain.invoke({input: 小刚有两只狗}, session_config)) print(conversation_chain.invoke({input: 共有几只宠物}, session_config))注意若想在执行链的同时把提示词 print 出来可在链中加入自定义函数函数的输入应原封不动返回出去避免破坏原有业务仅在 return 之前 print 所需信息即可。6.3 小结RunnableWithMessageHistory用于创建一个带有历史记忆功能的 Runnable 实例链创建时需要提供BaseChatMessageHistory的具体实现InMemoryChatMessageHistory实现在内存中存储历史七、长期记忆自实现 FileChatMessageHistory7.1 为什么内存不够InMemoryChatMessageHistory仅在内存中临时存储程序退出则记忆丢失。它继承自BaseChatMessageHistory。官方注释给出了实现指南并给出基于文件的历史消息存储示例。可以自行实现基于 JSON 和本地文件的会话数据保存。7.2 核心思路基于文件存储会话记录以session_id为文件名不同 session 不同文件继承BaseChatMessageHistory实现 3 个方法add_messages同步添加消息messages同步获取消息clear同步清除消息import json, os from langchain_core.messages import messages_from_dict, message_to_dict # message_to_dict单个消息对象BaseMessage类实例 - 字典 # messages_from_dict[字典、字典...] - [消息、消息...] # AIMessage、HumanMessage、SystemMessage 都是BaseMessage的子类 class FileChatMessageHistory(BaseChatMessageHistory): def __init__(self, session_id, storage_path): self.session_id session_id # 会话id self.storage_path storage_path # 不同会话id的存储文件所在的文件夹路径 # 完整的文件路径 self.file_path os.path.join(self.storage_path, self.session_id) # 确保文件夹是存在的 os.makedirs(os.path.dirname(self.file_path), exist_okTrue) def add_messages(self, messages: Sequence[BaseMessage]) - None: # Sequence序列 类似list、tuple all_messages list(self.messages) # 已有的消息列表 all_messages.extend(messages) # 新的和已有的融合成一个list # 将数据同步写入到本地文件中 # 类对象写入文件 - 一堆二进制 # 为了方便可以将BaseMessage消息转为字典借助json模块以json字符串写入文件 # 官方message_to_dict单个消息对象BaseMessage类实例 - 字典 # new_messages [] # for message in all_messages: # d message_to_dict(message) # new_messages.append(d) new_messages [message_to_dict(message) for message in all_messages] # 将数据写入文件 with open(self.file_path, w, encodingutf-8) as f: json.dump(new_messages, f) property # property装饰器将messages方法变成成员属性用 def messages(self) - list[BaseMessage]: # 当前文件内 list[字典] try: with open(self.file_path, r, encodingutf-8) as f: messages_data json.load(f) # 返回值就是list[字典] return messages_from_dict(messages_data) except FileNotFoundError: return [] def clear(self) - None: with open(self.file_path, w, encodingutf-8) as f: json.dump([], f)7.3 业务链部分from langchain_core.prompts import PromptTemplate from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_core.chat_history import BaseChatMessageHistory, BaseMessage from langchain_core.output_parsers import StrOutputParser from langchain_core.messages import messages_from_dict, message_to_dict from langchain_community.chat_models.tongyi import ChatTongyi from typing import Sequence, List import json llm ChatTongyi(modelqwen3-max) prompt PromptTemplate.from_template(你是一个贴心的助手需要根据对话历史回应用户的问题。 对话历史{chat_history} 用户当前输入{input} 你的回应) base_chain prompt | llm | StrOutputParser() def get_message_history(session_id: str) - BaseChatMessageHistory: 根据会话 ID 获取对应的对话历史存储实例 return FileChatMessageHistory(session_idsession_id, storage_path./chat_history) conversation_chain RunnableWithMessageHistory( runnablebase_chain, get_session_historyget_message_history, input_messages_keyinput, history_messages_keychat_history, ) if __name__ __main__: session_config {configurable: {session_id: user_001}} response1 conversation_chain.invoke({input: 小明有1只猫}, configsession_config) print(第一轮, response1) response2 conversation_chain.invoke({input: 小刚有2只狗}, configsession_config) print(\n第二轮, response2) response3 conversation_chain.invoke( {input: 小明和小刚一共有几只宠物?}, configsession_config ) print(\n第三轮, response3) # 测试程序重启后读取历史注释上面的代码单独运行下面的代码仍能获取历史 # response4 conversation_chain.invoke( # {input: 分别是什么宠物}, # configsession_config # ) # print(\n重启后第四轮, response4)
[LangChain RAG] 05 LCEL 链与 Memory：从竖线组链到多轮对话
一、链是什么「将组件串联上一个组件的输出作为下一个组件的输入」是 LangChain 链尤其是|管道链的核心工作原理也是链式调用的核心价值实现数据的自动化流转与组件的协同工作。chain prompt_template | model1.1 谁能入链核心前提即 Runnable 子类对象才能入链以及 Callable、Mapping 接口子类对象也可加入。目前学到的组件均是 Runnable 接口的子类。1.2 链执行起来什么样通过|链接提示词模板对象和模型对象返回值 chain 是RunnableSerializable对象它是 Runnable 接口的直接子类也是绝大多数组件的父类通过invoke或stream进行阻塞执行或流式执行组成的链上一个组件的输出作为下一个组件的输入from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_community.chat_models.tongyi import ChatTongyi from langchain_core.runnables.base import RunnableSerializable chat_prompt_template ChatPromptTemplate.from_messages( [ (system, 你是一个边塞诗人可以作诗。), MessagesPlaceholder(history), (human, 请再来一首唐诗无需额外输出), ] ) history_data [ (human, 你来写一个唐诗), (ai, 床前明月光疑是地上霜举头望明月低头思故乡), (human, 好诗再来一个), (ai, 锄禾日当午汗滴禾下锄谁知盘中餐粒粒皆辛苦), ] model ChatTongyi(modelqwen3-max) chain: RunnableSerializable chat_prompt_template | model print(type(chain)) # Runnable接口invoke执行 res chain.invoke({history: history_data}) print(res.content) # Runnable接口stream执行 for chunk in chain.stream({history: history_data}): print(chunk.content, end, flushTrue)1.3 链小结链是将各个组件串联在一起按顺序执行前一个组件的输出作为下一个组件的输入通过|符号让各个组件形成链成链的各个组件需是 Runnable 接口的子类形成的链是 RunnableSerializable 对象可通过链调用invoke或stream触发整个链条的执行二、|运算符为什么能组链2.1 本质是__or__chain chat_prompt_template | model在语法上使用了|运算符的重写。在 Python 中运算符行为由类的魔法方法决定a b本质调用a.__add__(b)a | b本质调用a.__or__(b)自行实现__or__即可重写|。2.2 课件示例a | b | cclass Test(object): def __init__(self, name): self.name name def __str__(self): return fTest({self.name}) def __or__(self, other): return MySequence(self, other) class MySequence(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))2.3 落到 LangChain 上chain prompt | model得到的是RunnableSequenceRunnableSerializable 子类原因就是 Runnable 基类内部对__or__的改写。后面继续用|加组件依旧得到 RunnableSequence——这就是链的基础架构。三、StrOutputParser为什么 prompt | model | model 会报错3.1 复现需求第一次模型的输出再拿去第二次询问模型。from langchain_core.prompts import PromptTemplate from langchain_community.chat_models.tongyi import ChatTongyi model ChatTongyi(modelqwen3-max) prompt PromptTemplate.from_template( 我邻居姓{lastname}, 刚生了{gender}请起名仅告知名字无需其它内容 ) chain prompt | model | model res chain.invoke({lastname: 张, gender: 女儿}) print(res.content)运行报错ValueError: Invalid input type class langchain_core.messages.ai.AIMessage. Must be a PromptValue, str, or list of BaseMessages.3.2 原因prompt 的结果是PromptValue输入给了 model —— 这一段是合法的model 的输出是AIMessage模型invoke的 input 类型是LanguageModelInput PromptValue | str | Sequence[MessageLikeRepresentation]不接收 AIMessage3.3 用 StrOutputParser 做类型转换LangChain 内置StrOutputParser字符串输出解析器把 AIMessage 解析为简单字符串且它是 Runnable 子类可以加入链。parser StrOutputParser() chain prompt | model | parser | model小结StrOutputParser 是内置的简单字符串解析器可以将 AIMessage 转换为基础字符串可以加入 chain。四、JsonOutputParser 与标准多模型链4.1 更标准的处理逻辑prompt | model | parser | model并不标准上一个模型的输出没有被处理成「下一个提示词模板」所需的输入。正常逻辑invoke / stream 初始输入 → 提示词模板 → 模型 → 数据处理 → 提示词模板 → 模型 → 解析器 → 结果即上一个模型的输出应作为提示词模板的输入构建下一个提示词用来二次调用模型。模型输出AIMessage提示词模板invoke要求输入dict输出PromptValueStrOutputParserAIMessage → str不够JsonOutputParserAIMessage → Dict(JSON)4.2 完整代码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 str_parser StrOutputParser() json_parser JsonOutputParser() model ChatTongyi(modelqwen3-max) first_prompt PromptTemplate.from_template( 我邻居姓{lastname}刚生了{gender}请起名并封装到JSON格式返回给我 要求key是namevalue就是起的名字。请严格遵守格式要求 ) second_prompt PromptTemplate.from_template( 姓名{name}请帮我解析含义。 ) chain first_prompt | model | json_parser | second_prompt | model | str_parser res: str chain.invoke({lastname: 张, gender: 女儿}) print(res) print(type(res))4.3 输入输出必须对齐组件输入输出模型PromptValue 或字符串或序列BaseMessage、list、tuple、str、dictAIMessage提示词模板字典PromptValueStrOutputParserAIMessagestrJsonOutputParserAIMessagedict标准链类型流字典 → first_prompt → PromptValue → model → AIMessage → json_parser → 字典 → second_prompt → PromptValue → model → AIMessage → str_parser → 字符串五、RunnableLambda自定义函数入链5.1 语法除了固定功能的解析器也可以自己编写 Lambda 完成自定义逻辑。RunnableLambda把普通函数转换为 Runnable 实例方便自定义函数加入 chain。语法RunnableLambda(函数对象或 lambda 匿名函数)from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnableLambda from langchain_core.prompts import PromptTemplate from langchain_community.chat_models.tongyi import ChatTongyi str_parser StrOutputParser() my_func RunnableLambda(lambda ai_msg: {name: ai_msg.content}) model ChatTongyi(modelqwen3-max) first_prompt PromptTemplate.from_template( 我邻居姓{lastname}刚生了{gender}请起名仅告知我名字不要额外信息 ) second_prompt PromptTemplate.from_template( 姓名{name}请帮我解析含义。 ) chain first_prompt | model | my_func | second_prompt | model | str_parser res: str chain.invoke({lastname: 张, gender: 女儿}) print(res) print(type(res))5.2 函数也可以直接入链chain first_prompt | model | (lambda ai_msg: {name: ai_msg.content}) | second_prompt | model | str_parser因为 Runnable 在实现__or__时支持 Callable函数就是 Callable 实例本质是将函数自动转换为 RunnableLambda。小结将函数封装入RunnableLambda它是 Runnable 接口实例可以直接入链直接将函数入链函数会自动转换为 RunnableLambda 对象六、临时记忆InMemoryChatMessageHistory如果想要封装历史记录除了自行维护历史消息外也可以借助 LangChain 内置的历史记录功能帮助模型在有历史记忆的情况下回答。6.1 两个关键类基于RunnableWithMessageHistory在原有链的基础上创建带有历史记录功能的新链新 Runnable 实例基于InMemoryChatMessageHistory为历史记录提供内存存储临时用from langchain_core.runnables.history import RunnableWithMessageHistory conversation_chain RunnableWithMessageHistory( some_chain, # 被附加历史消息的 Runnable通常是 chain None, # 获取指定会话 ID 的历史会话的函数 input_messages_keyinput, # 用户输入在模板中的占位符 history_messages_keychat_history # 历史消息在模板中的占位符 ) chat_history_store {} # 存放多个会话 ID 所对应的历史会话记录 def get_history(session_id): if session_id not in chat_history_store: chat_history_store[session_id] InMemoryChatMessageHistory() return chat_history_store[session_id]6.2 完整代码from langchain_community.chat_models.tongyi import ChatTongyi from langchain_core.chat_history import InMemoryChatMessageHistory from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import PromptTemplate from langchain_core.runnables.history import RunnableWithMessageHistory def print_prompt(full_prompt): print( * 20, full_prompt.to_string(), * 20) return full_prompt model ChatTongyi(modelqwen3-max) prompt PromptTemplate.from_template( 你需要根据对话历史回应用户问题。对话历史{chat_history}。用户当前输入{input} 请给出回应 ) base_chain prompt | print_prompt | model | StrOutputParser() chat_history_store {} def get_history(session_id): if session_id not in chat_history_store: chat_history_store[session_id] InMemoryChatMessageHistory() return chat_history_store[session_id] conversation_chain RunnableWithMessageHistory( base_chain, get_history, input_messages_keyinput, history_messages_keychat_history ) if __name__ __main__: session_config {configurable: {session_id: user_001}} print(conversation_chain.invoke({input: 小明有一只猫}, session_config)) print(conversation_chain.invoke({input: 小刚有两只狗}, session_config)) print(conversation_chain.invoke({input: 共有几只宠物}, session_config))注意若想在执行链的同时把提示词 print 出来可在链中加入自定义函数函数的输入应原封不动返回出去避免破坏原有业务仅在 return 之前 print 所需信息即可。6.3 小结RunnableWithMessageHistory用于创建一个带有历史记忆功能的 Runnable 实例链创建时需要提供BaseChatMessageHistory的具体实现InMemoryChatMessageHistory实现在内存中存储历史七、长期记忆自实现 FileChatMessageHistory7.1 为什么内存不够InMemoryChatMessageHistory仅在内存中临时存储程序退出则记忆丢失。它继承自BaseChatMessageHistory。官方注释给出了实现指南并给出基于文件的历史消息存储示例。可以自行实现基于 JSON 和本地文件的会话数据保存。7.2 核心思路基于文件存储会话记录以session_id为文件名不同 session 不同文件继承BaseChatMessageHistory实现 3 个方法add_messages同步添加消息messages同步获取消息clear同步清除消息import json, os from langchain_core.messages import messages_from_dict, message_to_dict # message_to_dict单个消息对象BaseMessage类实例 - 字典 # messages_from_dict[字典、字典...] - [消息、消息...] # AIMessage、HumanMessage、SystemMessage 都是BaseMessage的子类 class FileChatMessageHistory(BaseChatMessageHistory): def __init__(self, session_id, storage_path): self.session_id session_id # 会话id self.storage_path storage_path # 不同会话id的存储文件所在的文件夹路径 # 完整的文件路径 self.file_path os.path.join(self.storage_path, self.session_id) # 确保文件夹是存在的 os.makedirs(os.path.dirname(self.file_path), exist_okTrue) def add_messages(self, messages: Sequence[BaseMessage]) - None: # Sequence序列 类似list、tuple all_messages list(self.messages) # 已有的消息列表 all_messages.extend(messages) # 新的和已有的融合成一个list # 将数据同步写入到本地文件中 # 类对象写入文件 - 一堆二进制 # 为了方便可以将BaseMessage消息转为字典借助json模块以json字符串写入文件 # 官方message_to_dict单个消息对象BaseMessage类实例 - 字典 # new_messages [] # for message in all_messages: # d message_to_dict(message) # new_messages.append(d) new_messages [message_to_dict(message) for message in all_messages] # 将数据写入文件 with open(self.file_path, w, encodingutf-8) as f: json.dump(new_messages, f) property # property装饰器将messages方法变成成员属性用 def messages(self) - list[BaseMessage]: # 当前文件内 list[字典] try: with open(self.file_path, r, encodingutf-8) as f: messages_data json.load(f) # 返回值就是list[字典] return messages_from_dict(messages_data) except FileNotFoundError: return [] def clear(self) - None: with open(self.file_path, w, encodingutf-8) as f: json.dump([], f)7.3 业务链部分from langchain_core.prompts import PromptTemplate from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_core.chat_history import BaseChatMessageHistory, BaseMessage from langchain_core.output_parsers import StrOutputParser from langchain_core.messages import messages_from_dict, message_to_dict from langchain_community.chat_models.tongyi import ChatTongyi from typing import Sequence, List import json llm ChatTongyi(modelqwen3-max) prompt PromptTemplate.from_template(你是一个贴心的助手需要根据对话历史回应用户的问题。 对话历史{chat_history} 用户当前输入{input} 你的回应) base_chain prompt | llm | StrOutputParser() def get_message_history(session_id: str) - BaseChatMessageHistory: 根据会话 ID 获取对应的对话历史存储实例 return FileChatMessageHistory(session_idsession_id, storage_path./chat_history) conversation_chain RunnableWithMessageHistory( runnablebase_chain, get_session_historyget_message_history, input_messages_keyinput, history_messages_keychat_history, ) if __name__ __main__: session_config {configurable: {session_id: user_001}} response1 conversation_chain.invoke({input: 小明有1只猫}, configsession_config) print(第一轮, response1) response2 conversation_chain.invoke({input: 小刚有2只狗}, configsession_config) print(\n第二轮, response2) response3 conversation_chain.invoke( {input: 小明和小刚一共有几只宠物?}, configsession_config ) print(\n第三轮, response3) # 测试程序重启后读取历史注释上面的代码单独运行下面的代码仍能获取历史 # response4 conversation_chain.invoke( # {input: 分别是什么宠物}, # configsession_config # ) # print(\n重启后第四轮, response4)
