原始代码importos,jsonfromtypingimportSequencefromlangchain_community.chat_modelsimportChatTongyifromlangchain_core.messagesimportmessage_to_dict,messages_from_dict,BaseMessagefromlangchain_core.chat_historyimportBaseChatMessageHistoryfromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.promptsimportChatPromptTemplate,MessagesPlaceholderfromlangchain_core.runnablesimportRunnableWithMessageHistory# message_to_dict单个消息对象BaseMessage类实例 - 字典# messages_from_dict[字典、字典...] - [消息、消息...]# AIMessage、HumanMessage、SystemMessage 都是BaseMessage的子类classFileChatMessageHistory(BaseChatMessageHistory):def__init__(self,session_id,storage_path):self.session_idsession_id# 会话idself.storage_pathstorage_path# 不同会话id的存储文件所在的文件夹路径# 完整的文件路径self.file_pathos.path.join(self.storage_path,self.session_id)# 确保文件夹是存在的os.makedirs(os.path.dirname(self.file_path),exist_okTrue)defadd_messages(self,messages:Sequence[BaseMessage])-None:# Sequence序列 类似list、tupleall_messageslist(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)formessageinall_messages]# 将数据写入文件withopen(self.file_path,w,encodingutf-8)asf:json.dump(new_messages,f)property# property装饰器将messages方法变成成员属性用defmessages(self)-list[BaseMessage]:# 当前文件内 list[字典]try:withopen(self.file_path,r,encodingutf-8)asf:messages_datajson.load(f)# 返回值就是list[字典]returnmessages_from_dict(messages_data)#将纯字典的数据转换成了messageexceptFileNotFoundError:return[]defclear(self)-None:withopen(self.file_path,w,encodingutf-8)asf:json.dump([],f)#将文件打开写一个空列表进去modelChatTongyi(modelqwen3-max)# prompt PromptTemplate.from_template(# 你需要根据会话历史回应用户问题。对话历史{chat_history}用户提问{input}请回答# )promptChatPromptTemplate.from_messages([(system,你需要根据会话历史回应用户问题。对话历史),MessagesPlaceholder(chat_history),(human,请回答如下问题{input})])str_parserStrOutputParser()defprint_prompt(full_prompt):print(*20,full_prompt.to_string(),*20)returnfull_prompt base_chainprompt|print_prompt|model|str_parserdefget_history(session_id):returnFileChatMessageHistory(session_id,./chat_history)# 创建一个新的链对原有链增强功能自动附加历史消息conversation_chainRunnableWithMessageHistory(base_chain,# 被增强的原有chainget_history,# 通过会话id获取InMemoryChatMessageHistory类对象input_messages_keyinput,# 表示用户输入在模板中的占位符history_messages_keychat_history# 表示用户输入在模板中的占位符)if__name____main__:# 固定格式添加LangChain的配置为当前程序配置所属的session_idsession_config{configurable:{session_id:user_001}}resconversation_chain.invoke({input:小明有2个猫},session_config)print(第1次执行,res)resconversation_chain.invoke({input:小刚有1只狗},session_config)print(第2次执行,res)resconversation_chain.invoke({input:总共有几个宠物},session_config)print(第3次执行,res) 代码流程梳理导入依赖langchain_community.chat_models.ChatTongyi通义千问聊天模型。langchain_core.messages中的序列化工具message_to_dict消息 → 字典、messages_from_dict字典 → 消息。BaseChatMessageHistory自定义历史存储的基类。ChatPromptTemplate、MessagesPlaceholder构建聊天提示模板。RunnableWithMessageHistory为链增加自动历史管理。自定义FileChatMessageHistory继承BaseChatMessageHistory实现基于本地 JSON 文件的持久化存储。__init__根据session_id和storage_path构造文件路径并确保目录存在。add_messages将新消息追加到现有历史中通过message_to_dict序列化为字典列表覆盖写入文件。messages属性从文件读取 JSON用messages_from_dict还原为BaseMessage列表若文件不存在返回空列表。clear清空历史写入空列表。构建聊天提示模板ChatPromptTemplate.from_messages创建模板包含系统提示固定角色设定。MessagesPlaceholder(chat_history)运行时注入完整历史消息列表。人类消息占位符{input}。构建基础链base_chainbase_chainprompt|print_prompt|model|str_parserprint_prompt中间调试节点打印生成的完整提示文本用于观察历史注入情况。model通义千问模型。str_parser提取AIMessage.content返回纯字符串。定义历史获取函数get_history返回FileChatMessageHistory实例存储路径为./chat_history。包装为RunnableWithMessageHistoryconversation_chainRunnableWithMessageHistory(base_chain,get_history,input_messages_keyinput,history_messages_keychat_history)在invoke调用时根据session_id从get_history中获取历史存储对象。从存储中读取历史消息列表。将历史消息注入模板的chat_history位置。执行base_chain获取模型回复。将本轮用户消息和模型回复HumanMessage和AIMessage追加到历史存储中。三次执行同一session_id三次调用使用相同的session_configsession_iduser_001因此共享同一份历史文件。第一次用户输入“小明有2个猫”模型回复历史文件保存该轮对话。第二次用户输入“小刚有1只狗”历史包含首轮模型能理解上下文。第三次用户问“总共有几个宠物”模型从历史中提取数据正确回答“3个”。 相关知识点知识点说明BaseChatMessageHistoryLangChain 提供的抽象基类规范了历史存储的接口add_messages、messages、clear。message_to_dict/messages_from_dict内置序列化工具用于在BaseMessage和 JSON 字典之间转换便于持久化。自定义历史存储FileChatMessageHistory是一个典型的实现展示了如何将消息序列化到本地文件。MessagesPlaceholder在ChatPromptTemplate中插入整个消息列表保留角色结构而非纯文本拼接。RunnableWithMessageHistory一个Runnable包装器自动完成“加载历史 → 调用链 → 保存历史”的闭环。**LCEL 管道**会话隔离通过session_id或thread_id区分不同会话实现多用户对话记忆隔离。 面试高频追问与回答Q1FileChatMessageHistory存在并发写入问题吗如何解决A是的多进程/多线程同时写入同一文件会导致数据损坏。解决方式包括使用文件锁如fcntl、改用支持并发的存储如 Redis、PostgreSQL或采用队列异步写入。Q2如何控制历史消息的长度避免 Token 超限A可以在add_messages中实现限制逻辑如保留最近 N 轮或使用 LangChain 的trim_messages工具在注入前裁剪。Q3RunnableWithMessageHistory的input_messages_key和history_messages_key的作用A它们分别指定invoke字典中用户输入和历史消息的键名必须与提示模板中的占位符名称一致否则无法正确注入。Q4如果想在会话之间共享部分记忆如用户偏好该怎么做A可使用 LangChain 的Store组件如PostgresStore实现长期记忆手动读写跨会话的数据与基于thread_id的短期记忆互补。 总结该代码通过自定义FileChatMessageHistory和RunnableWithMessageHistory实现了一个支持持久化多轮对话的智能体。其核心价值在于展示了 LangChain 中对话记忆的抽象与扩展方式是理解BaseChatMessageHistory、MessagesPlaceholder和RunnableWithMessageHistory协同工作的绝佳范例。掌握这些组件便能轻松构建具备上下文感知能力的生产级对话应用。