LangGraph基础知识(Store )（四） - AI创想

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}

复制代码

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()

复制代码