LangChain教程:FileChatMessageHistory长期记忆 作者:马育民 • 2026-02-27 15:56 • 阅读:10003 # 介绍 本地文件版聊天历史管理器,适合**单实例、低并发、需要长期保存对话**的场景。 ### 存储格式 以 **JSON 数组** 格式写入文件,每条消息为 `BaseMessage` 对象的字典序列化形式。 ### 核心逻辑 - 初始化时创建/读取指定文件,初始化为空列表。 - 添加消息时:读取文件 → 反序列化为消息列表 → 追加新消息 → 序列化 → 写回文件。 - 读取消息时:读取文件 → 反序列化为 `BaseMessage` 列表。 ### 适用场景 - 个人/本地聊天机器人、命令行对话工具。 - 低流量、单用户、需要跨会话保留记忆的应用。 - 快速原型开发,无需部署数据库/缓存。 # 用法 | 方法/属性 | 功能 | 示例 | |---|---|---| | `__init__(file_path: str)` | 初始化,指定存储文件路径 | `history = FileChatMessageHistory("chat.json")` | | `add_user_message(content: str)` | 添加用户消息 | `history.add_user_message("你好")` | | `add_ai_message(content: str)` | 添加 AI 回复 | `history.add_ai_message("你好!")` | | `add_message(message: BaseMessage)` | 底层添加消息(支持自定义消息类型) | `history.add_message(HumanMessage(content="Hi"))` | | `messages` | 获取所有历史消息(`List[BaseMessage]`) | `print(history.messages)` | | `clear()` | 清空文件中的所有历史 | `history.clear()` | # 例子 简单使用 ```python # 1. 安装依赖(如未安装) # pip install langchain langchain-community # 2. 导入模块 from langchain_community.chat_message_histories import FileChatMessageHistory from langchain_core.messages import HumanMessage, AIMessage # 3. 初始化文件历史(文件不存在会自动创建) chat_history = FileChatMessageHistory(file_path="my_chat_history.json") # 4. 添加对话历史 chat_history.add_user_message("你好,我是小明") chat_history.add_ai_message("你好小明!我是AI助手") chat_history.add_user_message("今天天气怎么样?") chat_history.add_ai_message("今天天气晴朗,适合出门") # 5. 读取所有历史 print("当前聊天历史:") for msg in chat_history.messages: print(f"{msg.type}: {msg.content}") # 6. 清空历史(可选) # chat_history.clear() ``` ### 执行结果 ``` 当前聊天历史: human: 你好,我是小明 ai: 你好小明!我是AI助手 human: 今天天气怎么样? ai: 今天天气晴朗,适合出门 ``` # 案例 ``` from dotenv import load_dotenv from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables.history import RunnableWithMessageHistory from langchain_community.chat_models.tongyi import ChatTongyi from langchain_community.chat_message_histories import FileChatMessageHistory # 加载环境变量(需配置 OPENAI_API_KEY) load_dotenv() # 1. 初始化 LLM 和 Prompt(关键:添加 MessagesPlaceholder 接收对话历史) model = ChatTongyi(model="qwen3-max") # Prompt 中通过 占位符接收对话历史 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个温柔的小学数学老师,回答问题时用“亲”、“小朋友”,并且简短回答问题,回答问题不要用markdown格式"), ("placeholder", "{chat_history}"), ("human", "计算过程:{msg}") ]) # 2. 构建基础链(Prompt + LLM) chain = prompt | model # 3. 关键:文件聊天记忆(重启程序也不会丢) chat_history = FileChatMessageHistory(file_path="chat_history.json") # 4. 用 RunnableWithMessageHistory 包装基础链 # 核心:关联「对话历史获取函数」和「历史变量名」 chain_his = RunnableWithMessageHistory( chain, lambda session_id: chat_history, # 告诉链如何获取对话历史 history_messages_key="chat_history", # 对话历史的变量名(对应Prompt中的{chat_history}) input_messages_key="msg", # 用户当前输入的变量名(对应Prompt中的{input}) ) # 用户对话ID(同一个 session_id 共享历史) SESSION_ID = "user_小明" # 可理解为「用户ID」或「对话ID」 while True: """ 第一次提问:3x+5=20,x 等于多少? 第二次提问:5x-8=72,x 等于多少? 第三次提问:把x=10代入上面公式,等于多少? """ question = input("请输入:") # 调用stream向模型提问,返回 generator类型 res = chain_his.stream({ "msg": question, # 传入问题 }, # 关键:传入 config 指定 session_id,关联对话历史 config={ "configurable": { "session_id": SESSION_ID } } ) print("\n----AI回答-------\n") # 流式输出 for chunk in res: print(chunk.content, end="", flush=True) print("\n----AI回答结束-------\n") # 查看内存中的对话历史(验证存储) print("\n===== 查看对话历史 =====") for msg in chat_history.messages: print(f"{msg.type}: {msg.content}") print("\n===== 查看对话历史结束 =====\n\n\n") ``` # 优缺点 | 优点 | 缺点 | |---|---| | 实现简单,零依赖,开箱即用 | 读写性能一般,不适合高并发 | | 数据永久保存,程序重启不丢失 | 无内置并发锁,多进程写入易冲突 | | 可直接查看/编辑 JSON 文件,便于调试 | 单文件存储,不适合海量会话 | # 注意事项 1. **文件路径**:确保程序对目标路径有读写权限。 2. **并发安全**:多进程/多线程写入时,需自行加文件锁(如 `fcntl`/`portalocker`)。 3. **数据备份**:重要对话历史建议定期备份 JSON 文件。 4. **性能优化**:频繁读写可考虑批量添加消息,减少 IO 次数。 原文出处:http://malaoshi.top/show_1GW2r3H2PmCN.html