光说不练假把式，我们直接上代码。

先从最简单的内存存储开始。我们创建一个 InMemoryStore 对象，把它传给 create_agent，这样 Agent 就具备了存储和查询业务数据的能力。存数据用 store.put() 方法，三个参数依次是命名空间元组、键名和值，命名空间可以理解为“文件夹”，用来对数据做分类管理。取数据用 store.get() 方法，传入命名空间和键名就能把之前存的内容读出来，不管隔了多少轮对话，只要键对得上数据就在那里。

import 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 tool, BaseTool from langchain_community.tools import WriteFileTool, ReadFileTool, ListDirectoryTool from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.store.memory import InMemoryStore load_dotenv() prefix = "QWEN" model = init_chat_model( model_provider="openai", configurable_fields=["model", "api_key", "base_url"], config_prefix=prefix ).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() # 新增代码 conn = sqlite3.connect("checkpoints.db", check_same_thread=False) checkpointer = SqliteSaver(conn) store = InMemoryStore() agent = create_agent( model=model, tools=[calculate, write_file, read_file, list_dir], system_prompt="你是一个助手，会用工具计算、读写文件、列出目录。", debug=True, checkpointer=checkpointer, store=store ) # 添加配置 config = {"configurable": {"thread_id": "session-1"}} # 用户资料 store.put(("user", "user-1"), "profile", { "name": "张三", "role": "developer", "skills": ["python", "typescript", "java"] }) profile = store.get(("user", "user-1"), "profile") print(f"用户资料: {profile.value}") queries = [ "计算 2024*12+500，然后把结果保存到 result.txt", "读取 result.txt 的内容", "列出当前目录文件", "刚才计算的结果是多少？" ] for q in queries: print(f"\n问：{q}") result = agent.invoke({"messages": [{"role": "user", "content": q}]}, config=config) print(f"答：{result['messages'][-1].content}") # 使用完毕后关闭连接 conn.close()

不过 InMemoryStore 有个明显的短板——数据全在内存里，程序一关就什么都没了。拿来做 demo 没问题，真正跑项目肯定不行。所以我们把数据落到 SQLite 数据库里，让它真正持久化。具体做法是复用同一个数据库文件 agent.db，新增一个 SqliteStore 连接专门保存业务数据。这里有个容易踩坑的地方：创建连接时必须加上 isolation_level=None 参数来禁用自动事务管理，因为 agent.db 同时被 Checkpointer 和 Store 共用，如果不禁用自动事务，两者操作起来很容易出现嵌套事务冲突。改动很小，把 InMemoryStore 换成 SqliteStore 就行。最后别忘了，程序结束时关闭数据库连接，释放资源。

import 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 tool, BaseTool from langchain_community.tools import WriteFileTool, ReadFileTool, ListDirectoryTool from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.store.memory import InMemoryStore from langgraph.store.sqlite import SqliteStore load_dotenv() prefix = "QWEN" model = init_chat_model( model_provider="openai", configurable_fields=["model", "api_key", "base_url"], config_prefix=prefix ).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() # 新增代码 checkpoint_conn= sqlite3.connect("agent.db", check_same_thread=False, isolation_level=None) checkpointer = SqliteSaver(checkpoint_conn) # 新增代码 store_conn = sqlite3.connect("agent.db", check_same_thread=False, isolation_level=None) store = SqliteStore(store_conn) agent = create_agent( model=model, tools=[calculate, write_file, read_file, list_dir], system_prompt="你是一个助手，会用工具计算、读写文件、列出目录。", debug=True, checkpointer=checkpointer, store=store ) # 添加配置 config = {"configurable": {"thread_id": "session-1"}} # 用户资料 store.put(("user", "user-1"), "profile", { "name": "张三", "role": "developer", "skills": ["python", "typescript", "java"] }) profile = store.get(("user", "user-1"), "profile") print(f"用户资料: {profile.value}") queries = [ "计算 2024*12+500，然后把结果保存到 result.txt", "读取 result.txt 的内容", "列出当前目录文件", "刚才计算的结果是多少？" ] for q in queries: print(f"\n问：{q}") result = agent.invoke({"messages": [{"role": "user", "content": q}]}, config=config) print(f"答：{result['messages'][-1].content}") # 使用完毕后关闭连接 checkpoint_conn.close() store_conn.close()