这就是 AI 最让人头疼的 先天性失忆症——默认情况下所有大模型和智能体都是无状态的。每次 API 调用都是一次完全独立的交互模型不会保留任何之前的上下文信息就像每次见面都是初次相遇一样。动画视频在《25. AI 总是失忆3 行代码让它记住你》。不过别担心LangChain/LangGraph 提供了一套极其优雅的解决方案checkpointer检查点机制。只需要 3 行核心代码就能让你的智能体拥有持久化记忆不仅能记住完整的对话历史还能保存执行过程中的所有状态下次接着聊甚至程序重启后也能完美恢复之前的会话。3 行代码实现内存记忆我们先从最简单的内存记忆开始把对话状态保存在程序内存中。这种方式速度极快不需要任何额外的数据库依赖非常适合开发测试和短期会话场景。LangChain的 checkpointer 机制会自动拦截智能体的每一次状态变化包括用户输入、AI 回复、工具调用请求和工具返回结果并将这些状态保存下来。当你下次调用时它会根据唯一的会话标识自动加载之前的状态让智能体仿佛从未中断过对话。import os from dotenv import load_dotenv from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langchain_core.tools import BaseTool from langchain_community.tools import WriteFileTool, ReadFileTool, ListDirectoryTool from langgraph.checkpoint.memory import MemorySaver # 加载环境变量配置通义千问API load_dotenv() prefix QWEN model init_chat_model( model_provideropenai, configurable_fields[model, api_key, base_url], config_prefixprefix ).with_config({ configurable: { f{prefix}_model: os.getenv(f{prefix}_MODEL), f{prefix}_api_key: os.getenv(f{prefix}_API_KEY), f{prefix}_base_url: os.getenv(f{prefix}_BASE_URL) } }) # 自定义计算工具 class CalculateTool(BaseTool): name: str calculate description: str 计算数学表达式的值 def _run(self, expression: str) - str: try: return f计算结果: {eval(expression)} except Exception as e: return f计算错误: {str(e)} async def _arun(self, expression: str) - str: return self._run(expression) # 初始化工具 calculate CalculateTool() write_file WriteFileTool() read_file ReadFileTool() list_dir ListDirectoryTool() # 关键代码1创建内存检查点 checkpointer MemorySaver() # 创建带记忆功能的智能体 agent create_agent( modelmodel, tools[calculate, write_file, read_file, list_dir], system_prompt你是一个助手会用工具计算、读写文件、列出目录。, debugTrue, # 关键代码2将检查点传入智能体 checkpointercheckpointer ) # 关键代码3配置会话唯一标识 config {configurable: {thread_id: session-1}} # 测试对话包含记忆验证 queries [ 计算 2024*12500然后把结果保存到 result.txt, 读取 result.txt 的内容, 列出当前目录文件, 刚才计算的结果是多少 ] for q in queries: print(f\n问{q}) result agent.invoke({messages: [{role: user, content: q}]}, configconfig) print(f答{result[messages][-1].content})代码解释创建 MemorySaver这是内存版的检查点实现负责在程序内存中保存所有的状态数据。传入 create_agent将 checkpointer 对象传递给智能体构造函数这样智能体就会自动启用状态保存功能。设置 thread_id这是会话的唯一标识checkpointer 通过它来关联不同的对话状态。同一thread_id下的所有调用会共享同一个记忆上下文不同的thread_id则实现了多用户记忆隔离。验证记忆是否生效运行代码后你会看到智能体依次完成计算、保存文件、读取文件和列出目录的任务。当最后问到 刚才计算的结果是多少 时它会准确回答247882024×1250024788这说明它已经成功记住了之前的对话内容。数据库持久化记忆内存记忆虽然简单高效但有一个致命的缺点程序重启后所有记忆都会丢失。如果我们需要长期保存会话数据或者在多进程 / 多线程环境中共享记忆就需要把检查点保存到数据库中。LangGraph 官方提供了SqliteSaver使用轻量级的 SQLite 数据库来持久化检查点数据。SQLite 不需要单独安装服务器一个文件就是一个数据库非常适合中小型应用。首先安装必要的依赖包pip install langgraph-checkpoint-sqlite然后修改代码将MemorySaver替换为SqliteSaverimport os import sqlite3 from dotenv import load_dotenv from langchain.agents import create_agent from langchain.chat_models import init_chat_model from langchain_core.tools import BaseTool from langchain_community.tools import WriteFileTool, ReadFileTool, ListDirectoryTool from langgraph.checkpoint.sqlite import SqliteSaver # 加载环境变量配置通义千问API load_dotenv() prefix QWEN model init_chat_model( model_provideropenai, configurable_fields[model, api_key, base_url], config_prefixprefix ).with_config({ configurable: { f{prefix}_model: os.getenv(f{prefix}_MODEL), f{prefix}_api_key: os.getenv(f{prefix}_API_KEY), f{prefix}_base_url: os.getenv(f{prefix}_BASE_URL) } }) # 自定义计算工具 class CalculateTool(BaseTool): name: str calculate description: str 计算数学表达式的值 def _run(self, expression: str) - str: try: return f计算结果: {eval(expression)} except Exception as e: return f计算错误: {str(e)} async def _arun(self, expression: str) - str: return self._run(expression) # 初始化工具 calculate CalculateTool() write_file WriteFileTool() read_file ReadFileTool() list_dir ListDirectoryTool() # 关键修改创建SQLite检查点 # check_same_threadFalse 是必须的因为智能体可能会跨线程访问数据库 conn sqlite3.connect(checkpoint.db, check_same_threadFalse) checkpointer SqliteSaver(conn) # 创建带持久化记忆功能的智能体 agent create_agent( modelmodel, tools[calculate, write_file, read_file, list_dir], system_prompt你是一个助手会用工具计算、读写文件、列出目录。, debugTrue,