《讲透 LangGraph 记忆管理：30 分钟让 AI 从“金鱼脑”变“记忆大师”》

from typing import Annotated, List from langchain_core.messages import HumanMessage, AIMessage from langgraph.graph import StateGraph, START from langgraph.store.memory import InMemoryStore # 定义状态（这里直接用消息列表） class ChatState(TypedDict): messages: Annotated[List, lambda x, y: x + y] def chat_node(state: ChatState, *, store): """聊天节点，可以从 store 读取/写入长期记忆""" # 假设当前用户ID（正常应该从 config 里取，这里演示写固定值） user_id = "user_888" # ---------- 读取长期记忆 ---------- # store.get((命名空间,), key) -> 返回一个 item，没有则 None user_item = store.get(("users",), user_id) if user_item: user_info = user_item.value print(f"[记忆] 读取到历史信息：{user_info}") else: user_info = {} print("[记忆] 没有找到历史信息，这是第一次对话。") # 获取用户最新消息 last_msg = state["messages"][-1].content if state["messages"] else "" if "我叫" in last_msg: # 提取简单名字（这里示意，生产请用正则） name = last_msg.split("我叫")[-1].strip()[:10] user_info["name"] = name print(f"[记忆] 新记住名字：{name}") # 保存回长期记忆 store.put(("users",), user_id, user_info) # 生成回复 known_name = user_info.get("name", "朋友") reply = f"你好，{known_name}！有什么想聊的？" return {"messages": [AIMessage(content=reply)]} # ---------- 创建记忆存储 ---------- memory_store = InMemoryStore() builder = StateGraph(ChatState) builder.add_node("chat", chat_node) builder.add_edge(START, "chat") # 关键：编译时传入 store=memory_store graph = builder.compile(store=memory_store) # 第一次对话：用户说名字 print("===== 第一次对话（thread_1）=====") resp1 = graph.invoke({"messages": [HumanMessage(content="你好，我叫李雷")]}) print("AI:", resp1["messages"][-1].content) # 第二次对话：换一个全新的 thread_id（但长期记忆依然有效） # 注意：store 是全局的，不依赖于 config print("\n===== 第二次对话，新 thread，但我们能通过 store 找回记忆 =====") resp2 = graph.invoke({"messages": [HumanMessage(content="你还记得我叫什么吗？")]}) print("AI:", resp2["messages"][-1].content) # 查看 store 里保存的内容 user_data = memory_store.get(("users",), "user_888") print(f"\n[验证] 长期记忆存储：{user_data.value if user_data else '无'}")