先从最简单的内存存储开始。我们创建一个 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_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() # 新增代码 conn sqlite3.connect(checkpoints.db, check_same_threadFalse) checkpointer SqliteSaver(conn) store InMemoryStore() agent create_agent( modelmodel, tools[calculate, write_file, read_file, list_dir], system_prompt你是一个助手会用工具计算、读写文件、列出目录。, debugTrue, checkpointercheckpointer, storestore ) # 添加配置 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*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}) # 使用完毕后关闭连接 conn.close()不过 InMemoryStore 有个明显的短板——数据全在内存里程序一关就什么都没了。拿来做 demo 没问题真正跑项目肯定不行。所以我们把数据落到 SQLite 数据库里让它真正持久化。具体做法是复用同一个数据库文件 agent.db新增一个 SqliteStore 连接专门保存业务数据。这里有个容易踩坑的地方创建连接时必须加上 isolation_levelNone 参数来禁用自动事务管理因为 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_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() # 新增代码 checkpoint_conn sqlite3.connect(agent.db, check_same_threadFalse, isolation_levelNone) checkpointer SqliteSaver(checkpoint_conn) # 新增代码 store_conn sqlite3.connect(agent.db, check_same_threadFalse, isolation_levelNone) store SqliteStore(store_conn) agent create_agent( modelmodel, tools[calculate, write_file, read_file, list_dir], system_prompt你是一个助手会用工具计算、读写文件、列出目录。, debugTrue, checkpointercheckpointer, storestore ) # 添加配置 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*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}) # 使用完毕后关闭连接 checkpoint_conn.close() store_conn.close()
光说不练假把式，我们直接上代码。
先从最简单的内存存储开始。我们创建一个 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_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() # 新增代码 conn sqlite3.connect(checkpoints.db, check_same_threadFalse) checkpointer SqliteSaver(conn) store InMemoryStore() agent create_agent( modelmodel, tools[calculate, write_file, read_file, list_dir], system_prompt你是一个助手会用工具计算、读写文件、列出目录。, debugTrue, checkpointercheckpointer, storestore ) # 添加配置 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*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}) # 使用完毕后关闭连接 conn.close()不过 InMemoryStore 有个明显的短板——数据全在内存里程序一关就什么都没了。拿来做 demo 没问题真正跑项目肯定不行。所以我们把数据落到 SQLite 数据库里让它真正持久化。具体做法是复用同一个数据库文件 agent.db新增一个 SqliteStore 连接专门保存业务数据。这里有个容易踩坑的地方创建连接时必须加上 isolation_levelNone 参数来禁用自动事务管理因为 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_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() # 新增代码 checkpoint_conn sqlite3.connect(agent.db, check_same_threadFalse, isolation_levelNone) checkpointer SqliteSaver(checkpoint_conn) # 新增代码 store_conn sqlite3.connect(agent.db, check_same_threadFalse, isolation_levelNone) store SqliteStore(store_conn) agent create_agent( modelmodel, tools[calculate, write_file, read_file, list_dir], system_prompt你是一个助手会用工具计算、读写文件、列出目录。, debugTrue, checkpointercheckpointer, storestore ) # 添加配置 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*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}) # 使用完毕后关闭连接 checkpoint_conn.close() store_conn.close()
