LangGraph基础知识(Store )（四）

米落枫 · 发表于昨天 22:59

作者：CSDN博客
背景

仅使用`checkpointer`，我们无法做到跨线程共享信息。LangGraph 提出了Store（仓库）概念。
Store介绍

`LangGraph`通过`BaseStore`接口提供内置文档存储。与通过线程 ID 保存状态的`checkpointer`不同，存储使用自定义命名空间来组织数据。常见用例包括存储用户配置文件、构建知识库以及管理所有线程的全局首选项。具体的**实现形式是：`LangGraph` 将长期记忆作为 `JSON` 文档存储在`Store`中，每个`memory`都组织在自定义`namespace`（类似于文件夹）和不同的`key` （例如文件名）下。命名空间通常包含用户或组织 ID 或其他标签，以便更轻松地组织信息。这种结构可以实现存储器的分层组织。然后通过内容过滤器支持跨命名空间搜索。
整体而言，`LangGraph` 中的长期记忆允许系统保留不同对话或会话中的信息。与线程范围的短期内存不同，长期内存保存在自定义“命名空间”中。
举个例子

from langgraph.store.memory import InMemoryStore
import uuid
in_memory_store = InMemoryStore()
user_id = "1"
namespace_for_memory = (user_id, "memories")
memory_id = str(uuid.uuid4())
memory = {"user" : "你好，我叫BUG"}
in_memory_store.put(namespace_for_memory, memory_id, memory)
memories = in_memory_store.search(namespace_for_memory)
memories[-1].dict()
#运行结果
{'namespace': ['1', 'memories'],
'key': 'd792a784-0f52-4f91-b265-b5c80f794b86',
'value': {'user': '你好，我叫BUG'},
'created_at': '2025-06-14T07:22:41.758736+00:00',
'updated_at': '2025-06-14T07:22:41.758736+00:00',
'score': None}

复制代码

理解了上述过程后，就可以使用 `LangGraph` 中的`in_memory_store`方法了，当我们在编译图表时传递 `store` 对象，就会允许图中的每个节点访问 `store`，定义节点函数时的时候，就可以定义`store`关键字参数，`LangGraph` 会自动传递编译图时使用的 `store` 对象，代码如下所示：

import getpass
import os
from langchain_openai import ChatOpenAI
from typing import Annotated
from typing_extensions import TypedDict
from IPython.display import Image, display
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
from langgraph.graph.message import add_messages
from langchain_core.runnables import RunnableConfig
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langgraph.checkpoint.memory import MemorySaver
from dotenv import load_dotenv
load_dotenv()
in_memory_store = InMemoryStore()
memory = MemorySaver()
# if not os.environ.get("OPENAI_API_KEY"):
# os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")
# 定义大模型实例
llm = ChatOpenAI(model="deepseek-chat")
# 定义状态模式
class State(TypedDict):
messages: Annotated[list, add_messages]
# 定义对话节点，访问记忆并在模型调用中使用它们。
def call_model(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
# 获取用户id
user_id = config["configurable"]["user_id"]
# 定义命名空间
namespace = ("memories", user_id)
# 根据用户id检索记忆
memories = store.search(namespace)
print(f"memories: {memories}")
info = "\n".join([d.value["data"] for d in memories])
print(f"info: {info}")
# # 存储记忆
last_message = state["messages"][-1]
store.put(namespace, str(uuid.uuid4()), {"data": last_message.content})
system_msg = f"Answer the user's question in context: {info}"
response = llm.invoke(
[{"type": "system", "content": system_msg}] + state["messages"]
)
# 存储记忆
store.put(namespace, str(uuid.uuid4()), {"data": response.content})
return {"messages": response}
# 构建状态图
builder = StateGraph(State)
# 向图中添加节点
builder.add_node("call_model", call_model)
# 构建边
builder.add_edge(START, "call_model")
builder.add_edge("call_model", END)
# 编译图
graph = builder.compile(checkpointer=memory, store=in_memory_store)
# 可视化
display(Image(graph.get_graph().draw_mermaid_png()))

复制代码

进行测试：

config = {"configurable": {"thread_id": "10"}, "user_id": "6"}
async for chunk in graph.astream({"messages": ["你好，我是BUG"]}, config, stream_mode="values"):
chunk["messages"][-1].pretty_print()

复制代码

================================ Human Message ================================= 你好，我是BUG memories: [] info: ================================== Ai Message ================================== 你好！看来你可能想表达自己是“BUG”（程序中的错误或漏洞），或者是想开个有趣的玩笑？如果是技术问题，比如遇到了代码bug或者系统故障，可以告诉我具体情况，我会尽力帮你解决！如果是其他含义——比如昵称、游戏ID或者单纯想聊聊——也欢迎随时补充说明～

25位大厂高管转战生成式AI创业！吸金猛，不

LangGraph基础知识(Store )（四）

浏览过的版块

LangGraph——Multi-agent

关于我们

产品与服务

解决方案

产品与服务