上一课用thread_id保存一段对话;这次即使换了对话编号,只要是同一个user_id,助手仍能记住他喜欢的书。
新建unit4_lesson11_user_memory.py。这课不调用 DeepSeek:
from dataclasses import dataclass from typing import Annotated, TypedDict from langchain_core.messages import AIMessage, AnyMessage, HumanMessage from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import START, END, StateGraph from langgraph.graph.message import add_messages from langgraph.runtime import Runtime from langgraph.store.memory import InMemoryStore @dataclass class Context: user_id: str class ChatState(TypedDict): messages: Annotated[list[AnyMessage], add_messages] def respond(state: ChatState, runtime: Runtime[Context]) -> dict: text = state["messages"][-1].content namespace = ("bookstore_users", runtime.context.user_id) if text.startswith("记住:我喜欢"): genre = text.removeprefix("记住:我喜欢").strip(" 。!") runtime.store.put(namespace, "favorite_genre", {"genre": genre}) reply = f"记住了,你喜欢{genre}。" else: saved = runtime.store.get(namespace, "favorite_genre") reply = ( f"你喜欢{saved.value['genre']}。" if saved else "我还不知道你喜欢什么书。" ) return {"messages": [AIMessage(content=reply)]} builder = StateGraph(ChatState, context_schema=Context) builder.add_node("respond", respond) builder.add_edge(START, "respond") builder.add_edge("respond", END) graph = builder.compile( checkpointer=InMemorySaver(), store=InMemoryStore(), ) def ask(user_id: str, thread_id: str, question: str) -> None: result = graph.invoke( {"messages": [HumanMessage(content=question)]}, config={"configurable": {"thread_id": thread_id}}, context=Context(user_id=user_id), ) print(f"\n用户 {user_id}|对话 {thread_id}") print("问题:", question) print("回答:", result["messages"][-1].content) print("本段对话的消息数:", len(result["messages"])) ask("A", "chat-1", "记住:我喜欢科幻小说。") ask("A", "chat-2", "我喜欢什么书?") ask("B", "chat-3", "我喜欢什么书?")运行python unit4_lesson11_user_memory.py。观察第二次:A 换了thread_id,本段对话只有 2 条消息,却仍能答出“科幻小说”;B 则不知道。
这里InMemorySaver管每段对话的消息,InMemoryStore管按用户保存的偏好。注意本例的 Store 也只在程序运行期间有效;“跨对话”不等于“关掉程序后仍保存”。LangGraph 长期记忆文档