首页 > 基础资料博客日记

26. Agent 记不住业务数据？用 Store 给它加个“笔记本“！

2026-06-13 16:00:02基础资料围观2次

这篇文章介绍了26. Agent 记不住业务数据？用 Store 给它加个“笔记本“！，分享给大家做个参考，收藏极客资料网收获更多编程知识

Agent 通过 Checkpointer 记住对话上下文不是什么难事，但要是想让它存点“业务数据”——比如用户偏好、任务进度、历史操作记录这类东西——光靠 Checkpointer 就有点力不从心了。对话线程之间彼此隔离，换个线程就像失忆一样，之前积累的信息全用不上。Store 机制就是来解决这个问题的。你可以把它理解成 Agent 随身携带的一个“笔记本”：想记什么就记什么，以键值对的形式存进去，什么时候想查都能翻出来。更关键的是，这个笔记本不绑定某个具体的对话线程——不管 Agent 在哪个线程里工作，都能访问同一份数据，真正做到跨对话、跨线程的信息共享。

动画视频在《26. Agent记不住业务数据？用 Store 给它加个"笔记本"！》。

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

先从最简单的内存存储开始。我们创建一个 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()

文章来源:https://www.cnblogs.com/halfcode/p/20505331
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：jacktools123@163.com进行投诉反馈，一经查实，立即删除！

标签：

上一篇：[免费试用] iNeuOS_Vision视觉分析在SOP标准工作操作规程的应用
下一篇：没有了

首页 > 基础资料博客日记

26. Agent 记不住业务数据？用 Store 给它加个“笔记本“！

相关文章

最新发布

点击排行

本站推荐

标签云

首页 > 基础资料 博客日记

26. Agent 记不住业务数据？用 Store 给它加个“笔记本“！

相关文章

最新发布

点击排行

本站推荐

标签云

首页 > 基础资料博客日记