一、背景

在 LLM（大语言模型）原生交互中，模型默认是 “无状态” 的 —— 每次对话都是独立请求，无法记住上一轮的提问、用户偏好或上下文信息。例如：

• 用户先问 “如何用 LangChain 实现 RAG？”，模型给出回答后；
• 再问 “那这个流程里的文档分割用什么工具？”，原生模型会因 “忘记上一轮的 RAG 上下文” 而无法精准回应。

示例如下：

def test_no_memory(modelName: str):
    llm = getChatModel(modelName)
    # 告诉AI 一些内容
    res = llm.invoke("我叫张三")
    print("模型第一次返回：", res.content)
    # 询问模型之前告诉它的内容
    resp = llm.invoke("你知道我叫什么吗")
    # 模型回答 不知道，因为没有记忆，所以每次对话都是新的
    print("询问模型之前告诉他的事情：",resp.content)

LangChain 的Memory（记忆）组件正是为解决 “无状态” 问题而生 —— 它相当于 AI 的 “大脑记忆区”，负责存储、管理和调用交互过程中的上下文信息，让多轮对话、复杂任务流程（如多步推理、个性化服务）成为可能。

在实际业务场景中，Memory 组件更是核心支撑：无论是智能客服需记住用户历史诉求、企业知识库需关联用户过往提问，还是自动化工作流需保留中间结果，都依赖 Memory 实现 “连续上下文感知”。

二、记忆组件的核心特性

LangChain 的 Memory 组件并非简单的 “数据存储”，而是具备 “标准化、可扩展、易集成” 的专业记忆管理能力。其有两种体系结构：

• BaseChatMessageHistory：对话历史存储抽象基类，定核心接口
• BaseMemory”：记忆逻辑层抽象基类，管记忆格式化与读写

langchain3.0+以后。官方更推荐用 BaseChatMessageHistory 子类配合 RunnableWithMessageHistory 实现记忆，而非直接依赖 BaseMemory 子类。

维度	BaseChatMessageHistory	BaseMemory
定位	专用于存储和管理对话历史（消息序列）	通用内存管理，支持多类型数据（如上下文、知识库等）
功能	提供消息的增删查（如add_message、get_messages）	支持复杂内存操作（如状态跟踪、上下文推理）
使用场景	对话系统、聊天机器人（需保留历史消息）	需要动态记忆的复杂任务（如多轮推理、知识调用）
实现方式	基于消息列表（序列化存储）	基于键值对或图结构（支持非线性数据）

2.2. 统一的记忆操作接口

所有 Memory 类都继承自BaseMemory、BaseChatMessageHistory抽象基类，统一实现load_memory_variables（加载记忆）、save_context（保存上下文）、clear（清空记忆）三大核心方法，开发者无需关注底层存储逻辑（如内存、Redis、数据库），只需调用标准接口即可实现记忆管理。

方法名	核心作用	参数说明
__init__() -> None	构造方法 初始化记忆组件基础配置（如键名、输出格式）	通常包含： - memory_key: str（记忆在输入中的键名，默认 "history"） - return_messages: bool（是否返回 Message 对象，默认 False）
load_memory_variables(inputs: Dict[str, Any]) -> Dict[str, Any]	抽象方法（必实现) 从记忆中加载历史上下文，格式化为 Chain/Agent 可用的输入（如字符串或 Message 列表）	inputs: Dict[str, Any]：当前对话输入数据（如用户问题）
save_context(inputs: Dict[str, Any], outputs: Dict[str, Any]) -> None	抽象方法（必实现) “输入 - 输出” 上下文保存到记忆中，更新历史记录	- inputs: Dict[str, Any]：当前输入（如用户问题）- outputs: Dict[str, Any]：当前输出（如 AI 回答）
clear() -> None	抽象方法（必实现) 清空所有记忆内容（如对话历史、静态键值对），用于场景切换或主动清除	无参数

BaseChatMessageHistory 是 LangChain 中所有对话历史存储类的抽象基类（ABC），定义了对话历史管理的核心接口规范。所有具体的对话历史存储实现

方法名	类型	作用说明	参数说明	返回值
add_message(self, message: BaseMessage)	抽象方法	向对话历史中添加一条消息（用户消息 HumanMessage 或 AI 消息 AIMessage 等）	message: 消息对象，必须是 BaseMessage 的子类（如 HumanMessage、AIMessage、SystemMessage 等）	无
clear(self)	抽象方法	清空当前会话的所有对话历史	无	无
messages	抽象属性	获取当前会话的所有消息列表（按时间顺序排列）	无	列表，元素为 BaseMessage 子类对象（如 [HumanMessage(...), AIMessage(...)]）

2.3. 灵活的记忆类型适配

针对不同业务场景，Memory 提供多种记忆策略，可按需选择：

• 短对话场景：用ConversationBufferMemory保存完整对话历史；
• 长对话场景：用ConversationBufferWindowMemory只保留最近 N 轮，避免 Token 溢出；
• 实体关联场景：用ConversationEntityMemory提取并记忆用户、产品等实体信息（如 “用户张三偏好 Python 开发”）；
• 长文本摘要场景：用ConversationSummaryMemory将历史对话压缩为摘要，节省 Token 消耗。

Memory 的 “存储层” 与 “逻辑层” 分离，默认提供内存存储（InMemoryChatMessageHistory），同时支持扩展为持久化存储：

• 分布式场景：用RedisChatMessageHistory将对话历史存到 Redis，支持多实例共享记忆；
• 长期存储场景：用SQLChatMessageHistory存到 MySQL/PostgreSQL，满足用户长期对话记录需求；
• 云存储场景：集成 AWS S3、阿里云 OSS 等，实现海量对话数据的低成本存储。

三、Memory 类体系

LangChain 的 Memory 体系以BaseMemory为顶层抽象，衍生出 “通用记忆”“对话记忆” 两大分支，核心类关系如下：

3.1. 核心类继承图

3.2. 核心类功能对比

类名	核心功能	适用场景	优缺点
SimpleMemory	存储静态键值对（如{"user_name": "张三"}）	固定信息复用（如 API 调用中的固定参数）	优点：轻量、无需更新；缺点：无法动态保存上下文
ConversationBufferMemory	完整保存对话历史（用户消息 + AI 消息）	短对话（如 5 轮内咨询）、需回溯全部交互	优点：信息无丢失；缺点：长对话 Token 消耗大
ConversationBufferWindowMemory	保留最近 k 轮对话（如 k=3：只存最近 3 轮）	中长对话（如 10 + 轮）、控制 Token 成本	优点：节省 Token；缺点：可能丢失早期关键信息
ConversationSummaryMemory	用 LLM 生成对话摘要（如 “用户问了 RAG 实现，AI 推荐用 StuffChain”）	超长对话（如 30 + 轮）、需压缩历史	优点：大幅节省 Token；缺点：依赖 LLM 摘要准确性
ConversationEntityMemory	提取实体（如 “产品：LangChain，功能：链式流程”）	实体关联场景（如销售、客户管理）	优点：精准记忆实体属性；缺点：需额外定义实体提取规则

存储层实现记忆

类名	核心功能	适用场景	优缺点
InMemoryChatMessageHistory	基于内存列表存储对话消息	适用于临时对话、测试场景	优点：速度快、无依赖；缺点：不持久化、难共享
SQLChatMessageHistory	基于 SQL 数据库存储对话消息	数据库存储对话消息，适用于用户长期对话记录、合规存档	优点：持久化强、可查询；缺点：需依赖数据库服务
RedisChatMessageHistory	分布式存储对话消息	多实例部署、跨服务共享记忆	优点：支持分布式、高可用；缺点：需依赖 Redis 服务

四、基础 Memory 使用实战

以下结合实际开发场景，演示 3 种高频 Memory 的使用方式（基于 LangChain v0.3.x 版本）。

4.1. 短对话记忆(废弃)

需求：智能客服对话中，需记住用户上一轮提到的 “产品型号”，后续回答需关联该信息。

ConversationBufferMemory使用代码实现

from langchain.chains.llm import LLMChain
from langchain.memory import ConversationBufferMemory
from langchain_core.prompts import ChatPromptTemplate

from lang_chain_base.chat_models.model_language import getChatModel

def test_buffer_memory(modelName: str):
    # 1. 获取模型
    llm = getChatModel(modelName)

    # 2. 初始化ConversationBufferMemory
    memory = ConversationBufferMemory(
        memory_key="chat_history",  # 记忆在上下文中的键名
        return_messages=False  # 返回字符串格式的历史（True则返回Message对象列表）
    )

    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个助手，会使用对话历史{chat_history}回答问题。"),
        ("human", "{input}")
    ])
    # 3. 创建带记忆的Chain（不直接用LLMChain，而是通过Chain的方式）
    # 注意：虽然仍会有LLMChain的警告，但这是兼容写法
    chat_chain = LLMChain(
        llm=llm,
        prompt=prompt,
        memory=memory,
        verbose=True
    )

    # 4. 多轮对话测试
    response1 = chat_chain.run(input="你好，我叫小红")
    print("第一轮：", response1)  # 输出：你好，小红！有什么可以帮你的吗？

    response2 = chat_chain.run(input="我刚才说我叫什么？")
    print("第二轮：", response2)  # 输出：你刚才说你叫小红。

    # 查看当前记忆内容
    print("当前对话历史：", memory.load_memory_variables({})["chat_history"])

关键说明

• memory_key="chat_history"：需与 Prompt 中的{chat_history}变量名一致，确保 Chain 能正确加载记忆；

4.2. 长对话记忆(废弃)

需求：AI 助手处理 10 + 轮技术交流时，需控制 Token 消耗，只保留最近 3 轮对话历史。

ConversationBufferWindowMemory使用代码实现

from langchain.chains.llm import LLMChain
from langchain.memory import ConversationBufferWindowMemory
from langchain_core.prompts import ChatPromptTemplate

from lang_chain_base.chat_models.model_language import getChatModel

def test_buffer_window_memory(modelName: str):
    """
     测试长对话场景记忆组件
     多轮对话，只保留N个历史记录提供给模型
    """
    # 1. 获取模型
    llm = getChatModel(modelName)

    # 2. 创建记忆
    window_memory = ConversationBufferWindowMemory(
        memory_key="chat_window_history",  # 记忆在上下文中的键名
        k=3,  # 保留最近几轮对话
        return_messages=False  # 默认返回字符串格式的历史（True则返回Message对象列表）
    )

    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个助手，会使用对话历史{chat_window_history}回答问题。"),
        ("human", "{input}")
    ])
    # 3. 创建带记忆的Chain
    chat_chain = LLMChain(
        llm=llm,
        prompt=prompt,
        memory=window_memory,
        verbose=False
    )
    #循环10次测试
    for i in range(10):
        #模拟10次对话 记忆组件保留其中最后三轮对话
        chat_chain.invoke({"input": f"我的学号是{i+1},我的名字是{i+1}小子"})

    print("当前对话历史：", window_memory.load_memory_variables({})["chat_window_history"],"\n")
    # 前七轮对话 没有保存到记忆中，所以模型应该回答 不知道
    no_memory_resp = chat_chain.invoke({"input": "学号是1的人名字叫啥"})
    print("前七轮回答：", no_memory_resp["text"])

    # 后三轮 保存到记忆中，所以模型产生正确回答 注意上面的前七轮也算一轮对话了，
    # 所以最后三轮对话是 学号9、10、和学号是1的人名字叫啥 没有学号8的信息了
    no_memory_resp = chat_chain.invoke({"input": "学号是9的人名字叫啥"})
    print("最后三轮回答：", no_memory_resp["text"])

因为只保留最新的三轮回答，所以问前七轮问题，模型回答不知道。
只有问最后三轮回答，模型才正确回答（注意每一次对话又会程序最新的一轮记忆 替换掉之前的回答数据）

关键说明

• k=3：控制窗口大小，超过 3 轮的早期对话会自动被截断；
• 适用场景：的技术支持机器人、长期咨询类服务，避免因对话过长导致 LLM 调用成本上升或响应变慢。

4.4. redis分布式记忆

需求：多实例部署 AI 助手时，需实现 “用户在实例 A 发起对话，切换到实例 B 后仍能获取历史记录”。

RedisChatMessageHistory使用代码实现

from langchain.chains.llm import LLMChain
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableWithMessageHistory
from lang_chain_base.chat_models.model_language import getChatModel

def get_redis_history() -> RedisChatMessageHistory:
    """
       返回对应的Redis对话历史实例
    """
    # 创建redis类型的Memory
    # redis_url：Redis 服务地址，格式为 redis://[用户名:密码@]主机:端口/数据库编号（默认数据库编号为 0）
    return RedisChatMessageHistory(
        session_id="test_redis_memory",
        url="redis://username:password@localhost:6379/0",
        ttl=3600
    )
    
def test_redis_memory(modelName: str):
    """
     使用redis作为记忆组件工具
    """
    llm = getChatModel(modelName)

    prompt = ChatPromptTemplate.from_messages([
        ("system", "你一个助手，会使用对话历史{persist_history}回答问题。"),
        ("human", "{input}")
    ])

    # 3. 构建带记忆的对话链（使用 RunnableWithMessageHistory 包装）
    chat_chain = RunnableWithMessageHistory(
        runnable=prompt | llm,  # 核心逻辑管道：Prompt → LLM
        get_session_history=get_redis_history,  # 传入 Redis 存储的历史获取函数
        input_messages_key="input",  # 输入中用户问题的键名
        history_messages_key="persist_history"  # 对话历史在 Prompt 中的变量名
    )
    
    chat_chain.invoke({"input": "我有一条小狗叫做贝贝，她是一个4岁的中华田园犬。"})

    resp= chat_chain.invoke({"input": "我的狗叫什么名字，什么品种，多少岁了"})
    print("模型回答：", resp.content)
    print("当前对话历史：", get_redis_history().messages)

、
关键说明

• session_id：必须唯一，否则会导致不同用户记忆混淆；
• ttl：设置记忆过期时间，避免 Redis 存储冗余数据；
• 多实例部署时，所有实例只需使用相同的session_id和 Redis 连接，即可共享对话历史。

注意：langchain 推荐使用RunnableWithMessageHistory 形式去组合记忆组件，在上面4.1、4.2使用方式已经过时

4.5 基于mysql记忆

SQLChatMessageHistory是LangChain 中用于对话历史持久化存储的组件，属于 BaseChatMessageHistory 的子类，核心作用是将对话消息（用户输入、AI 回复）存储到 SQL 数据库。我们此处以mysql数据库例

• 安装mysql依赖

# MySQL
pip install pymysql

• 建表语句

CREATE TABLE `chat_history` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '自增主键',
  `session_id` varchar(255) NOT NULL COMMENT '会话唯一标识',
  `message` longtext NOT NULL COMMENT '序列化的消息内容（包含角色、文本、时间等）',
  `create_stime`  datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '消息创建时间',
  PRIMARY KEY (`id`),
  KEY `idx_session_id` (`session_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='存储对话历史记录，适配LangChain的SQLChatMessageHistory';

• 相关代码

与redis使用方式一致 定义一个获取SQLChatMessageHistory函数(get_sql_history)，创建RunnableWithMessageHistory时候传入的历史存储函数为get_sql_history即可

def get_sql_history(session_id: str) -> SQLChatMessageHistory:
    """
      返回对应的Redis对话历史实例
      session_id 对应表chat_history的session_id字段 首次使用没有会插入
    """
    # 数据库连接字符串（根据实际数据库类型调整）
    # - SQLite: "sqlite:///chat_history.db"（文件型数据库，无需额外部署）
    # - MySQL: "mysql+pymysql://user:password@host:port/dbname"
    # - PostgreSQL: "postgresql+psycopg2://user:password@host:port/dbname"
    connection_string ="mysql+pymysql://user:password@host:port/dbname"

    return SQLChatMessageHistory(
        session_id=session_id,  # 会话唯一标识
        connection=connection_string,
        table_name="chat_history"  # 存储对话的表名（自动创建，可自定义）
    )

 # 代码片段
 # 构建带mysql记忆的对话链（使用 RunnableWithMessageHistory 包装）
  chat_chain = RunnableWithMessageHistory(
      runnable=prompt | llm,  # 核心逻辑管道：Prompt → LLM
      get_session_history=get_sql_history,  # 传入 mysql存储的历史获取函数
      input_messages_key="input",  # 输入中用户问题的键名
      history_messages_key="persist_history"  # 对话历史在 Prompt 中的变量名
  )

• 查看数据库是否存储历史记录

五、memory进阶

5.1 自定义 Memory 实现

在特殊业务场景中（如需结合企业 CRM 数据、自定义记忆清理规则），原生 Memory 可能无法满足需求，此时可通过继承BaseMemory实现自定义记忆。

需求：AI 销售助手需通过某种途径获取（比如业务接口）用户提供的历史信息，并将其作为 “记忆” 融入对话。

from typing import Dict, Any
from langchain.memory import BaseMemory
from langchain.schema import BaseMessage, HumanMessage, AIMessage

# 1. 模拟企业接口
class CRMApi:
    @staticmethod
    def get_user_purchase_history(user_id: str) -> str:
        """从CRM获取用户历史信息"""
        # 实际场景：调用相关API接口查询数据
        return "你好，我是路飞，一个要成为海贼王的男人"

# 2. 自定义Memory：结合CRM数据与对话历史
class CRMBasedMemory(BaseMemory):
    user_id: str  # 用户ID（用于查询CRM）
    chat_history: list[BaseMessage] = []  # 对话历史
    memory_key: str = "crm_chat_history"  # 记忆键名

    @property
    def memory_variables(self) -> list[str]:
        """返回记忆变量名（必须实现）"""
        return [self.memory_key]

    def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, str]:
        """加载记忆：合并数据与对话历史（必须实现）"""
        # 1. 从业务接口中获取记录 作为对话历史使用
        crm_data = CRMApi.get_user_purchase_history(self.user_id)
        # 2. 格式化对话历史（用户消息+AI消息）
        chat_history_str = "\n".join([
            f"用户：{msg.content}" if isinstance(msg, HumanMessage) 
            else f"AI：{msg.content}" for msg in self.chat_history
        ])
        # 3. 合并为完整记忆
        full_memory = f"用户对话历史：{crm_data}\n对话历史：{chat_history_str}"
        return {self.memory_key: full_memory}

    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, Any]) -> None:
        """保存上下文：记录用户消息和AI消息（必须实现）"""
        # 提取用户输入（假设输入键为"question"）
        human_msg = HumanMessage(content=inputs["question"])
        # 提取AI输出（假设输出键为"text"）
        ai_msg = AIMessage(content=outputs["text"])
        # 添加到对话历史
        self.chat_history.append(human_msg)
        self.chat_history.append(ai_msg)

    def clear(self) -> None:
        """清空记忆（必须实现）"""
        self.chat_history = []

# 3. 使用自定义Memory
if __name__ == "__main__":
    # 初始化自定义记忆（传入用户ID）
    memory = CRMBasedMemory(user_id="user_123")
    # 初始化Chain
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你一个助手，会使用对话历史{chat_memory_history}回答问题。"),
        ("human", "当前问题：{question}")
    ])
    chat_chain = LLMChain(llm=llm, prompt=prompt, memory=memory)
    # 调用Chain：AI会自动结合CRM购买记录推荐
    response = chat_chain.invoke({"question": "我是谁"})

5.2 自定义 MessageHistory 实现

我们实现了一个基于BaseChatMessageHistory的读取json格式文件作为记忆信息的示例。

用户读取，存储记忆信息放入json格式文件中

自定义记忆组件实现类

实现了BaseChatMessageHistory 三个核心方法

• message方法: 加载历史记忆信息。
• add_message方法：对话完成后，追加此次对话为历史记录存入json格式文件中
• clear方法：手动调用，清空历史对话（json格式文件内容清除）。

class SimpleFileChatHistory(BaseChatMessageHistory):
    """
    基于BaseChatMessageHistory 实现记忆组件
    """
    def __init__(self, session_id: str, dir_path: str = "simple_chat_histories"):
        self.session_id = session_id
        self.dir_path = dir_path
        self.file_path = os.path.join(dir_path, f"{session_id}.json")

        # 确保存储目录存在
        os.makedirs(dir_path, exist_ok=True)

        # 初始化空文件（如果不存在）
        if not os.path.exists(self.file_path):
            with open(self.file_path, "w", encoding="utf-8") as f:
                json.dump([], f)

    @property
    def messages(self) -> list[BaseMessage]:
        """从文件读取消息并转换为BaseMessage对象"""
        with open(self.file_path, "r", encoding="utf-8") as f:
            raw_messages = json.load(f)

        # 直接转换为消息对象（简化版：只处理常见类型）
        message_list = []
        for msg in raw_messages:
            if msg["type"] == "human":
                message_list.append(HumanMessage(content=msg["content"]))
            elif msg["type"] == "ai":
                message_list.append(AIMessage(content=msg["content"]))
            elif msg["type"] == "system":
                message_list.append(SystemMessage(content=msg["content"]))

        return message_list

    def add_message(self, message: BaseMessage) -> None:
        """添加消息到文件（直接存储类型和内容）"""
        # 读取现有消息
        current_messages = self.messages

        # 简化存储格式：只保留类型和内容
        msg_type = ""
        if isinstance(message, HumanMessage):
            msg_type = "human"
        elif isinstance(message, AIMessage):
            msg_type = "ai"
        elif isinstance(message, SystemMessage):
            msg_type = "system"

        # 构建原始消息数据
        raw_msg = {
            "type": msg_type,
            "content": message.content
        }

        # 读取文件中已有的原始数据并添加新消息
        with open(self.file_path, "r", encoding="utf-8") as f:
            all_raw = json.load(f)
        all_raw.append(raw_msg)

        # 写回文件
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump(all_raw, f, ensure_ascii=False, indent=2)

    def clear(self) -> None:
        """清空对话历史"""
        with open(self.file_path, "w", encoding="utf-8") as f:
            json.dump([], f)

json格式文件(记忆组件存储方式)

[
  {
    "type": "human",
    "content": "你好，我是路飞，一个要成为海贼王的男人"
  }
]

测试使用

def test_my_histories(modelName:str):
    llm = getChatModel(modelName)
    # 初始化Chain
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你一个助手，会使用对话历史{chat_histories_memory}回答问题。"),
        ("human", "当前问题：{question}")
    ])
    # 核心方法 get_session_history 使用自定义的从文件获取历史记录实例
    chat_chain = RunnableWithMessageHistory(
        runnable=prompt | llm,  # 核心逻辑管道：Prompt → LLM
        get_session_history=get_simple_file_chat_history,  # 传入 mysql存储的历史获取函数
        input_messages_key="question",  # 输入中用户问题的键名
        history_messages_key="chat_histories_memory"  # 对话历史在 Prompt 中的变量名
    )
    #调用时会读取指定目录/user.json文件信息作为chat_histories_memory记忆
    user_session_id = "user"
    user_session_config = {"configurable": {"session_id": user_session_id}}
     # 记忆里面有 '你好，我是路飞，一个要成为海贼王的男人' 我再问我是谁的时候，他会回答记忆里的内容
    response = chat_chain.invoke({"question": "我是谁"},config=user_session_config)
    print("模型回答：", response.content)

没有记忆时的回答：

有记忆时的回答:

六、 使用LangGraph

LangGraph 是一个基于 LangChain 构建的扩展，专注于工作流抽象和编排。它可以与 LangChain 无缝集成，直接复用 LangChain 的 Chains、Tools 和 Memory 组件等，同时提供更强大的流程控制。例如，LangGraph 通过图结构（StateGraph）组织任务流程，使开发者能够精确控制执行逻辑、状态更新和任务调度，特别适合处理需要循环、分支、持久化状态管理和人工干预的复杂场景。

后续学习LangGraph,我们再着重说明

6.1 为啥使用langGraph

​ LangChain v0.3 版本推荐使用 LangGraph 持久化功能 实现应用内存管理，其支持多用户 / 多对话、错误恢复、自定义状态等特性，更适配复杂场景；兼容RunnableWithMessageHistory 与 BaseChatMessageHistory ，废弃 LangChain 0.0.x 版本的多数内存抽象（如 ConversationBufferMemory、BaseEntityStore）

组件类型	状态	说明
RunnableWithMessageHistory	无需改动，暂不废弃	用于 LCEL 编排的内存接口，适用于简单聊天应用，现有代码可正常运行
BaseChatMessageHistory	无需改动，暂不废弃	对话消息存储的抽象基类，支持基础的消息增删查，可与 LangGraph 配合使用
LangChain 0.0.x 内存抽象	多数废弃，需迁移	如 ConversationBufferMemory、BaseEntityStore 等，功能被 LangGraph 覆盖

6.2 使用记忆组件

langGraph提供如下的记忆组件。

存储方案	核心优势	适用场景	依赖要求
MemorySaver	轻量、无依赖、速度快	本地测试、临时会话	无（内存存储，重启丢失）
RedisSaver	分布式、高并发、低延迟	生产环境、多实例部署	需 Redis 服务
SQLSaver	长期存储、事务支持、复杂查询	金融 / 医疗等需合规存档的场景	需关系型数据库 + SQLAlchemy
MongoDBSaver	灵活文档结构、动态字段	非结构化状态、快速迭代的应用	需 MongoDB 服务
FileSaver	简单易用、无需数据库	单机应用、调试场景	本地文件系统
UpstashRedisSaver	Serverless 友好、托管服务	云原生、无服务器架构（如 Vercel、AWS Lambda）

我们以MemorySaver为例学习一下如何基于langGraph使用记忆组件

1. 安装依赖

pip install -U langchain langgraph langchain-openai  # 以 OpenAI 为例，也可替换其他 LLM

#我其他依赖已经安装,只单独安装langgraph
pip install -U langgraph

2. 代码示例

from langchain.schema import HumanMessage, AIMessage
from langgraph.graph import StateGraph, MessagesState  # 核心类：StateGraph 定义图，Graph 运行图
from langgraph.checkpoint.memory import MemorySaver  # 内存持久化（进阶可换数据库）
from lang_chain_base.chat_models.model_language import getChatModel

# ---------------------- 1. 定义 State（状态）：存储对话关键数据 ----------------------
# 用 TypedDict 强类型定义状态结构，确保节点间数据格式一致
class ChatState(MessagesState):
    # 如需扩展其他状态字段可在此添加，如：
    # user_info: dict = {}
    pass

# ---------------------- 2. 定义 Node（节点）：执行具体逻辑 ----------------------
# 节点 1：调用 LLM 生成回复（核心逻辑）
def generate_response(state: ChatState) -> ChatState:
    # 1. 从状态中获取对话历史
    messages = state["messages"]
    # 2. 调用 LLM（此处用 OpenAI，可替换为其他模型）
    llm = getChatModel("DouBao")
    response = llm.invoke(messages)  # 传入历史，让 LLM 基于上下文回复
    # 3. 将 AI 回复追加到状态的对话历史中，返回更新后的状态
    return {"messages": messages + [response]}

# ---------------------- 3. 构建 Graph（图）：定义流程 ----------------------
def get_workflow() -> StateGraph :
    # 3.1 初始化状态图，指定初始状态结构
    workflow = StateGraph(ChatState)

    # 3.2 添加节点：给节点命名（"generate"），绑定逻辑（generate_response）
    workflow.add_node("generate", generate_response)

    # 3.3 定义流程起点：从 "generate" 节点开始（因是简单对话，只有一个节点）
    workflow.set_entry_point("generate")

    # 3.4 定义流程终点："generate" 节点执行完后，流程结束（无分支）
    workflow.set_finish_point("generate")

    return workflow

# ---------------------- 4. 使用记忆组件 ----------------------
def test_lang_chain_memory():
    workflow = get_workflow()
    # 4 启用持久化：用 MemorySaver 存储状态（进阶可换 Redis/SQL）
    memory = MemorySaver()
    app = workflow.compile(checkpointer=memory)  # 编译图为可运行的应用
    result1 = app.invoke(
        input={"messages": [HumanMessage(content="我是路飞，一个吃了橡皮果实")]},  # 初始状态：用户输入
        config={"configurable": {"thread_id": "user_001"}}  # 会话 ID 确保状态隔离
    )
    # 打印 AI 回复（取最后一条 AIMessage）
    # print(f"AI 回复 1：{result1['messages'][-1].content}")

    # 对话 2：用户追问（基于历史上下文）
    result2 = app.invoke(
        input={"messages": [HumanMessage(content="我是谁")]},  # 新输入
        config={"configurable": {"thread_id": "user_001"}}  # 同一会话 ID，自动加载历史
    )
    print(f"AI 回复 2：{result2['messages'][-1].content}")

    # 查看完整对话历史
    print("\n完整对话历史：")
    for msg in result2["messages"]:
        print(f"{msg.__class__.__name__}: {msg.content}")

# ---------------------- 5. 运行 Graph：与用户交互 ----------------------
if __name__ == "__main__":
    test_lang_chain_memory()

3. 回答结果

可以看到，第二次回答使用了第一次对话的记忆