从 Store 基础设施到自主记忆 Agent，Module 5 构建了一套完整的
跨会话长期记忆体系，让 AI 真正记住每一个用户。
(user_id, "memories") 的元组，像文件系统路径一样组织记忆Checkpointer 只管"这次对话说了什么"，Store 管"这个用户是谁"。L1 引入了跨线程长期记忆的基础架构，让 AI 在不同会话之间真正记住用户。
(user_id, "memories") 像文件路径，隔离不同用户和不同类型记忆SearchItem 列表，每项含 .value 和相关度分数Profile（画像）是对用户的单一结构化认知：姓名、职业、偏好……每次对话后用 LLM 自动更新，始终保持最新版本，用固定 key 覆写，永远只有一份。
create_extractor(llm, tools=[Profile]) 封装 LLM，专职从对话中提取结构化 Profilewrite_memory 节点读取消息历史，调用 trustcall 更新 ProfileCollection（集合）存储的是独立、可增量扩充的事实列表：用户说过的偏好、学过的概念、记录的待办……每条记忆独立存在，用 UUID 作 key，互不干扰。
content: str 字段，LLM 决定从对话中提取哪些值得记录的独立事实str(uuid4()) 作为 key，保证新记忆追加而不是覆盖旧记忆前三课的记忆写入由图的节点硬编码触发。L4 进化到让 LLM 自主决定何时读写记忆——Agent 通过工具调用，在需要时主动存储任务、偏好和反思，无需人工干预。
compile(checkpointer=memory, store=in_memory_store) 同时挂载两种存储，节点可同时访问store: BaseStore 参数，LangGraph 自动注入当前 Store 实例save_memory / update_memory 工具，或直接回复用户，完全自主决策Checkpointer 存"这次会话的上下文"（工作记忆），Store 存"关于这个用户的持久事实"（长期记忆）。前者是线程级，后者是用户级，组合使用才完整。
Profile（固定 key + Pydantic）适合"单一且会演化的全局事实"；Collection（UUID key + 列表）适合"不断增长的独立条目"。选错 Schema 会导致记忆相互覆盖或重复爆炸。
store.search(query) 让记忆系统可以存储数千条历史记忆，但每次只检索最相关的几条注入 LLM。记忆量不再受限于 Token 窗口，系统真正可扩展。
(user_id, type) 的命名空间设计让多用户系统天然隔离。同一 Store 实例可服务数百个用户，每个用户的记忆完全独立互不干扰。
从"节点硬编码触发写记忆"进化到"LLM 判断值不值得记录"，是从规则驱动到智能驱动的跨越。Agent 能感知哪些对话内容值得长期保留，这才是真正个性化助理的核心能力。
记忆写入放在 respond 节点之后，用户拿到回复的瞬间，记忆更新在后台进行。对用户来说，记忆系统是"透明"的——感知不到延迟，但每次对话后记忆都悄然更新。
一个能在多次对话间持续了解你的学习助理——记录你的知识背景（Profile），累积你的学习笔记（Collection），并由 LLM 自主决定何时提取和检索记忆（Memory Agent）。综合运用 Module 5 全部 4 个核心知识点，呈现完整的长期记忆工程。
# ── 依赖导入 ──────────────────────────────────────────────
import uuid
from typing import Optional, Literal
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore # 【L1】跨线程 Store
from langgraph.checkpoint.memory import MemorySaver # 【L1】对话内 Checkpointer
from trustcall import create_extractor # 【L2】结构化提取库
# ════════════════════════════════════════════════
# 【L2】Profile Schema：单一结构化用户画像
# · 固定 key "user_profile" → 每次覆写同一份
# · 所有字段 Optional，新对话只更新提及的字段
# ════════════════════════════════════════════════
class LearnerProfile(BaseModel):
name: Optional[str] = Field(None, description="用户姓名")
expertise_level: Optional[str] = Field(None, description="技术水平: beginner/intermediate/advanced")
current_topics: Optional[list[str]] = Field(None, description="正在学习的技术主题列表")
preferred_style: Optional[str] = Field(None, description="偏好的解释风格: concise/detailed/example-driven")
background: Optional[str] = Field(None, description="职业背景或编程经验简述")
# ════════════════════════════════════════════════
# 【L3】Collection Schema：独立学习笔记条目
# · 每条笔记独立，UUID key → 追加不覆盖
# · LLM 决定对话中哪些知识点值得记录
# ════════════════════════════════════════════════
class LearningNote(BaseModel):
topic: str = Field(description="知识点所属主题（如 LangGraph、Python、ML）")
content: str = Field(description="具体知识点内容或用户的理解/疑问")
type: Literal["concept", "question", "insight"] = Field(
description="记忆类型：concept=新概念, question=待解答疑问, insight=用户洞察"
)
# ── 【L1】初始化双存储 ──────────────────────────────────
in_memory_store = InMemoryStore() # 跨会话长期记忆
checkpointer = MemorySaver() # 对话内短期记忆（断点/时间旅行）
llm = ChatOpenAI(model="gpt-4o-mini")
# ── 【L2】trustcall 提取器：从对话历史提取 LearnerProfile ──
profile_extractor = create_extractor(llm, tools=[LearnerProfile], tool_choice="LearnerProfile")
# ── 【L3】trustcall 提取器：从对话提取多条 LearningNote ──
notes_extractor = create_extractor(llm, tools=[LearningNote], tool_choice="LearningNote")用户不会在一次对话中说出所有信息。Optional + trustcall 的 patch 模式保证：对话提到了 name，就只更新 name；没提到 expertise_level，就保留上次的值。避免"新对话把旧 Profile 清零"的问题。
LearningNote 的 type 字段（concept/question/insight）让 Store 的语义检索更精准——可以单独 search 用户的 question 类型笔记，找出所有未解答的疑问，而不是把所有笔记混在一起搜索。
class LearnerState(MessagesState):
user_id: str # 用户唯一 ID，用于构建 namespace
profile_ctx: str # 从 Store 读出的 Profile 摘要，注入 System
notes_ctx: str # 从 Store 检索出的相关笔记，注入 System
# ════════════════════════════════════════════════
# 【L1 + L2 + L3】retrieve_memories 节点
# · 接收 store 参数（LangGraph 自动注入）
# · 从 Store 读取 Profile（L2）和相关笔记（L3）
# · 将记忆格式化后存入 State，后续节点直接使用
# ════════════════════════════════════════════════
def retrieve_memories(state: LearnerState, store: BaseStore):
user_id = state["user_id"]
last_msg = state["messages"][-1].content
# ── 【L2】读取用户画像（Profile）────────────────────────
# 用固定 key "user_profile" get 单条记忆，无则返回 None
profile_ns = (user_id, "profile")
profile_item = store.get(profile_ns, "user_profile")
profile_ctx = ""
if profile_item:
p = profile_item.value
profile_ctx = f"用户画像：姓名={p.get('name','未知')}, 水平={p.get('expertise_level','未知')}, "
f"当前主题={p.get('current_topics','')}, 偏好风格={p.get('preferred_style','')}"
# ── 【L3】语义检索最相关学习笔记（Collection）──────────
# 用当前问题作为 query，找到最相关的 3 条历史笔记
notes_ns = (user_id, "notes")
results = store.search(notes_ns, query=last_msg, limit=3)
notes_ctx = ""
if results:
notes_lines = [f"[{r.value['type']}] {r.value['topic']}: {r.value['content']}"
for r in results]
notes_ctx = "相关学习记录：\n" + "\n".join(notes_lines)
return {"profile_ctx": profile_ctx, "notes_ctx": notes_ctx}节点函数签名写 store: BaseStore，LangGraph 在 compile(store=...) 时会自动将 Store 实例注入给每个声明了该参数的节点。不需要手动传递或通过 State 中转 Store 实例。
Profile 用 store.get(ns, "user_profile")——精确 key 查找，返回单条记忆或 None。Collection 用 store.search(ns, query=...)——语义搜索，返回按相关度排序的列表。两种方式在同一节点中配合使用。
# ════════════════════════════════════════════════
# 【L4】respond 节点：Memory Agent 的核心
# · LLM 读取 Profile + 相关笔记，个性化作答
# · 可选地调用 save_memory 工具主动触发记忆保存
# · 也可以主动调用 search_memory 工具获取更多记忆
# ════════════════════════════════════════════════
# ── 【L4】供 LLM 调用的记忆工具 ──────────────────────────
from langchain_core.tools import tool
@tool
def save_memory(topic: str, content: str, memory_type: str) -> str:
"""将重要知识点或用户信息保存到长期记忆。
用于：用户分享了重要背景信息、学到了新概念、提出了关键疑问时。
memory_type: 'concept' / 'question' / 'insight'
"""
return f"已请求保存记忆：[{memory_type}] {topic}: {content}"
@tool
def search_memory(query: str) -> str:
"""主动检索用户的历史学习记录。
用于：需要回顾用户之前的问题或理解时。
"""
return f"已请求检索记忆：{query}"
agent_llm = llm.bind_tools([save_memory, search_memory])
def respond(state: LearnerState):
"""【L4】核心回复节点：根据用户画像和历史笔记个性化作答"""
profile_ctx = state.get("profile_ctx", "")
notes_ctx = state.get("notes_ctx", "")
system_parts = ["你是一个个性化学习伴侣，根据用户的知识背景和历史笔记定制回答。"]
if profile_ctx:
system_parts.append(f"\n{profile_ctx}")
if notes_ctx:
system_parts.append(f"\n{notes_ctx}")
system_parts.append(
"\n当对话中出现值得记录的新知识点、用户疑问或重要背景信息时，"
"请主动调用 save_memory 工具。这些记录将在未来对话中帮助你更好地辅导用户。"
)
response = agent_llm.invoke([
SystemMessage(content="\n".join(system_parts)),
*state["messages"],
])
return {"messages": [response]}注意 save_memory 的 docstring——这是给 LLM 看的决策指南，不是给开发者看的。LLM 读取工具描述后，自主判断对话中是否出现值得记录的信息。这比写规则硬编码触发时机聪明得多：它能感知语义而非格式。
工具本身只做"记录意图"——实际的 store.put() 操作发生在 save_notes 节点，respond 节点结束后异步执行。这样设计让 respond 节点保持轻量，用户拿到回复时记忆写入才开始，完全不阻塞响应速度。
# ════════════════════════════════════════════════
# 【L2】update_profile 节点
# · 用 trustcall 从完整对话历史提取 LearnerProfile
# · 固定 key "user_profile"：原地覆写，始终只有一份画像
# · patch 模式：只更新对话中提及的字段
# ════════════════════════════════════════════════
async def update_profile(state: LearnerState, store: BaseStore):
user_id = state["user_id"]
profile_ns = (user_id, "profile")
# 先读取已有 Profile 作为上下文，让 trustcall 做增量更新
existing = store.get(profile_ns, "user_profile")
existing_profile = existing.value if existing else {}
result = await profile_extractor.ainvoke({
"messages": state["messages"],
"existing": {"LearnerProfile": existing_profile}, # patch 模式的关键
})
if result["responses"]:
updated = result["responses"][0].model_dump(exclude_none=True)
# 【L1】写入 Store：固定 key，下次 put 覆写同一条记录
store.put(profile_ns, "user_profile", updated)
# ════════════════════════════════════════════════
# 【L3 + L4】save_notes 节点
# · 解析 respond 节点的工具调用结果
# · 也用 notes_extractor 从对话中提取知识点
# · UUID key：每条笔记独立追加，不覆盖历史
# ════════════════════════════════════════════════
async def save_notes(state: LearnerState, store: BaseStore):
user_id = state["user_id"]
notes_ns = (user_id, "notes")
# ── 方式 1：解析 LLM 工具调用（L4：Agent 主动触发）──────
last_msg = state["messages"][-1]
if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
for tc in last_msg.tool_calls:
if tc["name"] == "save_memory":
args = tc["args"]
note = {"topic": args["topic"], "content": args["content"],
"type": args.get("memory_type", "concept")}
# 【L1 + L3】UUID key：每条笔记独立存储，追加不覆盖
store.put(notes_ns, str(uuid.uuid4()), note)
# ── 方式 2：无论 LLM 是否主动触发，也用 extractor 自动提取 ──
result = await notes_extractor.ainvoke({"messages": state["messages"][-4:]})
for note_obj in result["responses"]:
store.put(notes_ns, str(uuid.uuid4()), note_obj.model_dump())传入 "existing": {"LearnerProfile": old_profile} 告诉 trustcall"这是当前版本"。trustcall 的 patch 模式会对比新旧，只生成变化字段的更新指令（JSON Patch），而不是重新生成整个 Profile。这保证了历史数据不会因为一次对话的局部信息而被清空。
方式 1（工具调用）：捕获 LLM 认为"非常重要、应该主动记录"的内容。方式 2（自动提取）：保底策略，确保即使 LLM 没有触发工具调用，本轮对话的知识点也不会丢失。两者形成互补，记忆系统更可靠。
# ════════════════════════════════════════════════
# 【L1 + L4】图构建：双 Store 挂载
# · checkpointer：对话内状态（断点/时间旅行）
# · store：跨会话长期记忆（Profile + Notes）
# ════════════════════════════════════════════════
builder = StateGraph(LearnerState)
builder.add_node("retrieve_memories", retrieve_memories)
builder.add_node("respond", respond)
builder.add_node("update_profile", update_profile) # 【L2】异步记忆节点
builder.add_node("save_notes", save_notes) # 【L3】异步记忆节点
builder.add_edge(START, "retrieve_memories")
builder.add_edge("retrieve_memories", "respond")
builder.add_edge("respond", "update_profile") # respond 后更新 Profile
builder.add_edge("respond", "save_notes") # respond 后保存笔记
builder.add_edge("update_profile", END)
builder.add_edge("save_notes", END)
# 【L1 + L4】同时挂载 checkpointer 和 store
graph = builder.compile(checkpointer=checkpointer, store=in_memory_store)
# ── 多轮对话示例 ──────────────────────────────────────────
user_id = "learner_tony"
config_1 = {"configurable": {"thread_id": "session_001"}}
# ── 第一次会话：介绍自己 ──
result = graph.invoke({
"user_id": user_id,
"messages": [HumanMessage(content="我叫 Tony，是个 Python 工程师，最近在学 LangGraph，对 Memory 机制特别感兴趣")]
}, config=config_1)
# → Profile 已更新: name=Tony, background=Python工程师, current_topics=[LangGraph]
# → Notes 新增: [concept] LangGraph/Memory 机制
# ── 第二次会话（新 thread，但 Store 记忆持久化）──
config_2 = {"configurable": {"thread_id": "session_002"}}
result2 = graph.invoke({
"user_id": user_id,
"messages": [HumanMessage(content="InMemoryStore 和 Checkpointer 有什么区别？")]
}, config=config_2)
# retrieve_memories 自动搜索到：Tony 是 Python 工程师，正在学 LangGraph + Memory
# respond 节点用这些背景个性化回答，不再是"对陌生人的通用解释"
# → LLM 可能调用 save_memory 记录这个"question"类型的笔记
# ── 验证 Profile 持久化 ──
stored_profile = in_memory_store.get((user_id, "profile"), "user_profile")
print(stored_profile.value)
# {'name': 'Tony', 'background': 'Python工程师', 'current_topics': ['LangGraph'], ...}
# ── 验证 Notes 积累 ──
all_notes = in_memory_store.search((user_id, "notes"), query="LangGraph memory")
for note in all_notes:
print(f"[{note.value['type']}] {note.value['topic']}: {note.value['content']}")
# [concept] LangGraph: Memory 机制 - Checkpointer vs Store 的区别
# [question] LangGraph: InMemoryStore 与 Checkpointer 的具体区分第二次会话用了新的 thread_id: "session_002"，Checkpointer 里没有第一次对话的消息——但 Store 里有 Tony 的 Profile 和笔记。这就是 Store 的核心价值：跨线程持久化。用户关闭 App 再打开，AI 还记得他是谁、在学什么。
respond → update_profile 和 respond → save_notes 两条边让两个记忆节点并行执行（LangGraph 的扇出语义）。用户看到回复后，Profile 更新和笔记保存同时进行，比串行快一倍。这是 L4 的"异步写入"设计在图结构上的体现。
同时维护用户画像（单一覆写）和学习笔记集合（逐条追加），两种记忆服务不同目的，不产生冲突。真实个性化助理通常需要两种模式配合。
LLM 主动调用工具的"精准记录"与 trustcall 自动提取的"兜底保障"双管齐下，记忆不遗漏，覆盖率更高。
LearningNote 的 type 字段（concept/question/insight）让未来可以过滤性地只检索"用户未解答的 question 类型笔记"，实现更精准的记忆召回。
Profile 更新采用 patch 模式，只修改本轮对话涉及的字段，历史信息安全保留。避免"每次更新都重新生成整个 Profile"带来的数据丢失风险。