第 7 节 基于 LangGraph 的 MiniChatGPT 助手#
在「从零实现 Mini ChatGPT 完整实战教程：短期记忆、长期记忆与工具调用 AI 助手开发详解」 中我们详细介绍了如何基于本地持久化的长期记忆来从零实现一个类似 ChatGPT 的个人助手。不过，从工程实现角度看这个版本更多是手工维护流程：短期记忆需要由程序员自己维护，长期记忆则保存在本地 JSON 文件中，整体更适合理解原理。
要把它推向更接近生产的形态，光手动拼接消息、手动维护状态是顶不住的——会话状态怎么持久化、多个线程怎么隔离、长期记忆怎么统一交给数据库、工具调用怎么跟消息状态形成闭环，这些都会变成新问题。
这一节，我们将使用 LangGraph 来重构整个过程，相比 JSON 版本的实现，这里的核心变化有 2 个：
-
短期记忆不再手动维护，交给 LangGraph 的 Checkpointer 自动管理；
-
长期记忆不再落本地 JSON，迁到 PostgreSQL，靠
PostgresStore统一存和查。
这样一来，我们不仅保留了原有长期记忆、短期记忆、工具写入的整体思路，还进一步获得了更清晰的状态编排能力。
7.1 整体结构设计思路#
首先，长期记忆的定位没变，保存的是用户跨会话共享的稳定信息（技术栈、长期目标、进行中的项目、回答偏好等），跟具体某次会话无关，只跟"用户"绑定。user_id 不变，换个新线程照样能调出来。
其次，短期记忆的职责也没变，把同一轮会话的上下文消息保住，让模型能接着前面那段对话继续。差别在于：JSON 版本里的短期记忆是保存在内存中手动维护的，本节里直接交给 LangGraph 的 Checkpointer 自动存——会话上下文状态正式变成图运行时的一部分。
最后，整个系统的驱动方式也从手工调用大模型并循环拼接上下文转变为让 LangGraph 统一组织状态、节点与持久化。虽然当前工程中的图结构只有一个核心节点 chat，如图7-1所示，但这个节点已经被放入了标准的 StateGraph 中运行，因此后续如果需要继续扩展出摘要节点、记忆筛选节点、回复评估节点等，也都可以沿着同一套图结构继续增加。
从图7-1 这个角度（Agent 节点编排）看，本节的 MiniChatGPT 已经不只是有记忆的聊天助手——而是一个真正基于图状态编排的聊天智能体。chat 节点内部处理流程跟「第 4 节」的图4-1 一致，工程目录跟图4-2 一致，就不再赘述。
接下来，我们将围绕 Mini-ChatGPT/mini-chatgpt-langgraph 工程中的代码，依旧按照第 4 节中的顺序，逐步介绍这个版本的实现过程。
7.2 记忆存储结构改造#
在第 4 节内容中长期记忆是 JsonMemoryStore 写本地文件；这一节改成 store.py 里的 PostgresMemoryStore，统一落到 PostgreSQL 中。
与先前一样，本工程仍然使用 MemoryRecord 和 MemorySearchResult 两个数据结构来分别表示一条记忆以及一条记忆检索结果，略微改动在于 MemoryRecord 中去掉了 embedding 字段，因为记忆向量将被存储到 store_vector 表中，语义检索能力由 PostgresStore 与外部向量索引共同承担，不再需要在业务层显式维护每条记录的向量内容。这部分内容就不再赘述，各位读者直接参见源码即可。
7.3 记忆持久化改造#
进一步，需要借助 PostgresStore 来完成持久化长期记忆，并同样实现插入记忆、更新记忆、保存记忆等功能。这里依旧主要介绍其中核心的3个方法， list_memories、upsert 和search ，分别用来返回当前用户的所有记忆、插入或更新记忆以及根据当前用户请求搜索与之相关的记忆内容，其它功能实现可直接阅读工程代码注释，对应模块为 store.py。同时，由于短期记忆的持久化 PostgresSaver 会自动完成。
（1）列出当前所有记忆
先看 PostgresMemoryStore 的初始化 + list_memories 方法：
1 class PostgresMemoryStore:
2 def __init__(self,store: PostgresStore, memory_root: str):
3 self.store = store
4 self.memory_root = memory_root
5
6 def get_namespace(self, user_id: str) -> tuple[str, str]:
7 return self.memory_root, user_id
8
9 def list_memories(self, user_id: str) -> list[MemoryRecord]:
10 namespace = self.get_namespace(user_id)
11 records, offset, page_size = [], 0, 100
12 while True:
13 items = self.store.search(namespace, limit=page_size, offset=offset)
14 if not items:
15 break
16 for item in items:
17 record = self._item_to_record(item)
18 if record:
19 records.append(record)
20 if len(items) < page_size:
21 break
22 offset += page_size
23 records.sort(key=lambda record: record.updated_at, reverse=True)
24 return records在上述代码中，第6~7行是为了实现不同用户长期记忆彼此隔离，通过命名空间对长期记忆进行组织，这样设计以后，只要 user_id 不同，即使多个用户都连接到同一个 PostgreSQL 实例，也会落在不同的命名空间下；而当同一个用户开启多个会话时，由于 user_id 没有变化，因此仍然可以共享同一份长期记忆。
第9~24行是根据传入的 user_id 来获取该用户的所有长期记忆，其中第12~22行是循环从表 store 中每次读取100条记录直到结束，第16~19行是将其转换成 MemoryRecord 的记忆格式，第23行是将最后的结果以更新时间降序排列返回。
（2）更新或插入新记忆
进一步，系统还需要具备长期记忆的插入与更新能力，核心代码如下：
1 def upsert(self, user_id: str, content: str, context: str,
2 , memory_type: MemoryType = "semantic",
3 memory_id: str | None = None,
4 source_thread_id: str | None = None) -> MemoryRecord:
5 normalized_memory_type = self._normalize_memory_type(memory_type)
6 namespace = self.get_namespace(user_id)
7 content, context = content.strip(), context.strip()
8 now = datetime.now().isoformat(timespec="seconds")
9 existing = self._find_existing(namespace, content, memory_id)
10 record_id = existing.memory_id if existing else str(uuid.uuid4())
11 created_at = existing.created_at if existing else now
12 value = {"memory_id": record_id, "memory_type": normalized_memory_type,
13 "content": content, "context": context, "source": "chat",
14 "source_thread_id": source_thread_id, "created_at": created_at,
15 "updated_at": now}
16 self.store.put(namespace, record_id, value)
17 return MemoryRecord(memory_id=record_id, content=content,
18 context=context, created_at=created_at, updated_at=now,
19 memory_type=normalized_memory_type)在上述代码中，第9行 _find_existing() 会先尝试判断当前内容是否对应已有记忆，或者是否显式传入了旧的 memory_id。如果命中已有记忆，则第10~11行会保留旧的 memory_id 与 created_at，只更新内容和更新时间；如果没有命中，则为新的记忆生成新的memory_id 与 created_at。第12~16行是将这条记忆插入到表 store 中。
相比第 4 节中的实现，此处最大的变化在第 16 行。之前我们是将整个用户记忆列表重新落盘到本地 JSON 文件；而现在则是直接通过 self.store.put() 把结构化字典写入到数据库。这样不仅使记忆存储更加稳定，也为后续分页、检索、删除和多端共享提供了基础。同时，在第14行中还加入了记忆产生的原始会话标识 source_thread_id，便于记忆溯源。
store.put() 落库以后，记忆原文会进 store 表：
prefix | key | value |created_at|updated_at|expires_at|ttl_minutes
-----------+------+-------------------------+----------+----------+----------+-----------
memory.user|e99...|{"content":"用户最..",...}|2026-06...|2026-06...| NULL | NULL
memory.user|6ad...|{"content":"2026-..",...}|2026-06...|2026-06...| NULL | NULL对应的向量会进 store_vector 表：
prefix | key |field_name| embedding |created_at|updated_at
-----------+------+----------+-----------------+----------+-----------
memory.user|e99...| content |[-0.013,0.034...]|2026-06...|2026-06...|
memory.user|e99...| context |[-0.025,0.017...]|2026-06...|2026-06...|
memory.user|6ad...| content |[-0.022,0.012...]|2026-06...|2026-06...|
memory.user|6ad...| context |[-0.073,0.014...]|2026-06...|2026-06...|因为一条记忆里有 content 和 context 两个文本字段，所以 store_vector 里它就占 2 行。
（3）检索已有记忆
检索方法：
1 def search(self, user_id: str, query: str,
2 limit: int = 12) -> list[MemorySearchResult]:
3 results: list[MemorySearchResult] = []
4 namespace = self.get_namespace(user_id)
5 items = self._search(namespace, query, limit)
6 if not items:
7 return results
8 for item in items:
9 record = self._item_to_record(item)
10 if record:
11 results.append(MemorySearchResult(record=record,
12 score=float(getattr(item, "score", 0.0) or 0.0)))
13 results.sort(key=lambda item: (item.score, item.record.updated_at),
14 reverse=True)
15 return results在上述代码中，第5行 _search() 是交由底层 PostgresStore 执行检索，具体过程可以参见上一节内容中的介绍。第8~12行是将底层检索结果重新规整为 MemorySearchResult。最后，第13~14行按照相似度优先、更新时间次之的规则对结果进行排序并返回。
有了检索结果，用户发新问题时就能先去长期记忆里捞最相关的几条，再塞进系统提示词——模型回答时就同时知道"当前问题"和"用户过往背景"两件事的关联。
（4）记忆插入工具改造
记忆插入/更新工具 tools.py 的写法跟第 4 节基本一致，这里就不再重复。
7.4 Agent 智能体节点实现#
在完成长期记忆存储与工具封装以后，接下来就需要真正把这些能力放进 LangGraph 节点中运行。当前工程的核心节点是 agent.py 中定义的 chat() 方法，示例代码如下：
1 def chat(state: MessagesState, runtime: Runtime[Context]) -> MessagesState:
2 user_text = state["messages"][-1].content
3 memory_store = PostgresMemoryStore(runtime.store, runtime.context.memory_root)
4 memory_hits = memory_store.search(runtime.context.user_id, user_text)
5 system_prompt = SYSTEM_PROMPT.format(user_info=_format_memories(memory_hits),
6 time=datetime.now().isoformat(timespec="seconds"))
7 messages: list[BaseMessage] = [
8 SystemMessage(content=system_prompt),state["messages"]]
9 tool = build_postgres_memory_tool(memory_store, runtime.context)
10 runnable = load_chat_model(runtime.context.model).bind_tools([tool])
11 for _ in range(5):
12 ai_message = runnable.invoke(messages)
13 messages.append(ai_message)
14 tool_calls = getattr(ai_message, "tool_calls", []) or []
15 if not tool_calls:
16 return {"messages": [AIMessage(content=message_text(ai_message))]}
17 for tool_call in tool_calls:
18 result = tool.invoke(tool_call["args"])
19 messages.append(ToolMessage(content=result, tool_call_id=tool_call["id"]))
20 return {"messages": [AIMessage(content="本轮记忆工具调用次数过多，已停止继续执行。")]}上述代码的组织逻辑同「第4节」 中 respond() 函数一致，都是用户输入内容与大模型多次交互的过程。具体地说，第2行获取用户当前输入。第3~4行构造长期记忆存储对象并检索与当前输入最相关的历史记忆。第5~6行将这些记忆格式化后填入系统提示词中。第7~8行再把系统提示词与当前会话中的消息状态拼接起来，形成真正发给模型的上下文内容。
进一步，第9~10行将长期记忆写入工具绑定到模型上。这样模型在回答之前，就可以根据系统提示词和当前输入自行判断是否需要调用 upsert_memory。如果模型没有触发任何工具调用，那么第15~16行直接返回最终回答；如果模型触发了工具调用，那么第17~19行就会执行工具，并把工具执行结果包装成 ToolMessage 继续追加到消息序列中，再由下一轮模型继续调用处理。
这里的 for _ in range(5) 与第4.6节中对应的部分一样，也是 ReAct 范式的实现过程，只不过现在这个循环发生在 LangGraph 节点内部，并且每一次中间结果都被组织成 LangGraph 认可的消息类型，包括 SystemMessage、AIMessage 和 ToolMessage。这意味着整个状态演化过程不再只是普通字符串拼接，而是一个具备明确语义的消息流。
7.5 Agent 工作流构建#
在有了节点函数以后，还需要把它真正组织成一个图。当前工程在 agent.py 模块中的 get_workflow() 里完成了这一步，示例代码如下：
1 def get_agent(checkpointer: PostgresSaver, store: PostgresStore):
2 agent = StateGraph(MessagesState, context_schema=Context)
3 agent.add_node("chat", chat)
4 agent.add_edge(START, "chat")
5 agent.add_edge("chat", END)
6 return agent.compile(checkpointer=checkpointer, store=store)在上述代码中，第2行表示当前这张图使用 MessagesState 作为全局状态结构，同时允许在运行时额外接收一个 Context 类型的上下文对象。第3~5行定义了一条最简单的执行路径，即从 START 进入 chat，处理结束以后直接到 END。
从图结构上看，这似乎只是一个非常简单的单节点流程，但它已经从一个普通的聊天程序升级成了一个可以持续扩展的图工作流。
7.6 运行时上下文结构定义#
在基于 LangGraph 组织整个流程之前，首先需要明确哪些信息属于图运行时的上下文。在当前工程中，这部分内容通过 context.py 中定义的 Context 类来统一表示，示例代码如下：
1 @dataclass(kw_only=True)
2 class Context:
3 user_id: str = "default"
4 model: str = field(default="qwen-plus", ...)
5 thread_id: str = field(default="demo_user_001", ...)
6 memory_root: str = field(default="demo_memory_root", ...)在上述代码中，第5行 thread_id 表示每个会话对应的会话 ID ，PostgresSaver 会根据 thread_id 将短期记忆进行持久化，因此，同一个用户可以拥有多个不同的 thread_id，这些线程之间共享长期记忆，但彼此拥有独立的短期上下文
进一步，在真正调用 Agent 时将会通过如下方式来组装输入及上下文，见 agent.py 模块，示例代码如下：
1 def invoke_agent(agent, user_id: str, thread_id: str,
2 user_text: str, model: str, memory_root: str) -> str:
3 input_data = {"messages": convert_to_messages(
4 [{"role": "user", "content": user_text}])}
5 config = RunnableConfig({"configurable": {"thread_id": thread_id}})
6 result = agent.invoke(input_data,config=config,
7 context=Context(user_id=user_id,thread_id=thread_id,
8 model=model,memory_root=memory_root))
9 return result["messages"][-1].content从上述代码可以看出，第5行是通过可运行配置 RunnableConfig 将 thread_id 传入到节点中。第6～8行构造的 Context 则是提供给业务逻辑使用的运行时上下文，后续节点函数可以通过 runtime.context.user_id、runtime.context.thread_id 等方式直接访问这些信息。
7.7 长短期记忆的协同#
理解这个工程最关键的一点，在于分清楚短期记忆和长期记忆分别由谁管理，以及它们如何共同作用。在 main.py 中，我们通过如下方式初始化运行环境：
1 def main() -> None:
2 args = build_parser().parse_args()
3 memory_index = get_memory_index()
4 with (PostgresStore.from_conn_string(DB_URI, index=memory_index) as store,
5 PostgresSaver.from_conn_string(DB_URI) as checkpointer):
6 agent = get_agent(checkpointer, store)
7 memory_store = PostgresMemoryStore(store, args.memory_root)
8 if args.cli or not STREAMLIT_AVAILABLE:
9 run_cli(agent, memory_store, args, checkpointer)
10 else:
11 run_streamlit_app(agent, memory_store, args, checkpointer)在上述代码中，第4行的 PostgresStore 负责长期记忆，第5行的 PostgresSaver 负责图运行过程中的状态检查点，即短期记忆。两者虽然都落在 PostgreSQL 中，但职责完全不同。这样设计以后，同一个用户在多个线程中的聊天内容可以共享长期记忆；而不同用户即使使用同一个系统，也不会互相污染彼此的数据。
第6行是返回定义的 Agent 工作流。第7行是得到实例化的长期记忆存储器 memory_store。第8~11行分别提供了命令行交互以及网页端交互的两种运行方式，相关介绍可以直接参见源码注释。
7.8 服务运行 CLI 与 Streamlit#
（1）命令行模式
命令行主流程在 cli.py 里。启动时会显示用户 ID、会话 ID、模型名、命名空间；运行过程支持若干管理命令。命令行模式下，进 Mini-ChatGPT/mini-chatgpt-langgraph 目录后这样启：
(rag) wangcheng@Cheng mini-chatgpt-langgraph % python main.py --cli
========== Mini ChatGPT(Postgres存储长期记忆)==========
用户ID = demo-user2
会话ID = 3869e1ed-fe45-410e-aada-dfe370990894
模型 = qwen-plus
记忆命名空间 = ('demo_memory', 'demo-user')
已存储记忆数量 = 0
Commands: /new 新会话, /thread <id> 切换会话, /memories: 查看当前用户所有长期记忆,
/clear: 清空当前用户所有长期记忆, /clearall: 清空用户所有记忆, /exit: 退出, /help: 查看帮助（2）页面模式
除命令行外，工程在 streamlit.py 里还提供了一个简单页面版。核心思路是把 thread_id、消息列表和 user_id 保存在 Streamlit 的 session_state 里——每次用户输入新消息就调一次 invoke_agent() 拿回复。侧边栏还能直接看当前用户保存的所有长期记忆。
页面模式同样进 Mini-ChatGPT/mini-chatgpt-langgraph 目录后这样启：
(rag) wangcheng@Cheng mini-chatgpt-langgraph % streamlit run main.py
You can now view your Streamlit app in your browser.
Local URL: http://localhost:8501
Network URL: http://192.168.0.109:8501然后在浏览器打开上面的链接即可。
先在会话 7c45cf2e-35…… 里告诉智能体一些基本信息，它会提取后作为长期记忆保存（图7-2）。
然后新建一个会话再问同样的问题来验证长期记忆是否真的被记住（图7-3）。
从图7-3智能体在回答问题以前的确加入了之前保存的长期记忆。作为验证，各位可以再次新建会话，同时点击左侧”删除长期记忆“按钮，此时再问智能体同样的问题便不会得到类似的答案。
7.9 小结#
到这，基于 LangGraph 的 MiniChatGPT 就跑通了。相比第 4 节的手工版本，核心提升不是"换了个 PostgreSQL 存储"这么简单——重要的是引入了 LangGraph 的状态图思想，把短期记忆、长期记忆、工具调用、消息流转统统组织进一个清晰、可扩展的运行框架。系统既保留了 ChatGPT 式的多轮会话体验，又拿到了跨会话长期记忆能力。