摘要：本文围绕 LangGraph 中的长期记忆机制展开，先说明如何通过运行时获取 store，再介绍在调用大模型前查询相关记忆并拼接到提示词中的典型做法，最后给出一个完整的 Python 示例，演示如何结合检查点与内存存储实现持久化聊天机器人，并支持跨线程共享用户记忆。

内容参考于：图灵AI大模型全栈

在LangGraph中是通过运行时来获取的store，如下图红框

在LangGraph中使用长期记忆，一般就是在调用大模型之前根据问题去查询相关的记忆，然后获取到记忆后拼接到提示词中，再调用大模型

效果图：

代码

from dataclasses import dataclass
from typing import Annotated, List
from typing_extensions import TypedDict
from operator import add
import uuid
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langchain_core.messages import HumanMessage, AIMessage, BaseMessage
from langgraph.runtime import Runtime
from langchain.chat_models import init_chat_model
import os
from dotenv import load_dotenv
load_dotenv()
初始化大模型
llm = init_chat_model(api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
model_provider="openai",
model='qwen3.8-flash')
创建自定义状态
class MessagesState(TypedDict):
messages: Annotated[List[BaseMessage], add]
上下文
@dataclass
class MyContext:
user_id: str
创建检查点长期记忆
checkpointer = InMemorySaver()
创建内存长期记忆
in_memory_store = InMemoryStore()
创建聊天机器人节点
state是自定义状态
runtime是运行时，通过它获取工作流中配置的信息（状态）
def chatbot(state: MessagesState, runtime: Runtime[MyContext]):
"""主聊天机器人节点，处理用户消息并生成回复"""
# 获取用户ID和最新消息
print(runtime)
# 取出用户id
user_id = runtime.context.user_id
# 从状态中获取消息
last_message = state["messages"][-1]

# 定义内存命名空间，user_id是用户的id
namespace = (user_id, "memories")

# 获取用户输入的内容
user_input = last_message.content.lower()
# 通过命名空间获取对应的长期记忆，这里获取的所有长期记忆
memories = runtime.store.search(namespace)
# 这里是把取出来的长期记忆按照换行进行拼接
memory_text = "\n".join(m.value["memory"] for m in memories)
# 这里把记忆和问题拼接成提示词
prompt = f"请参考聊天记录：{memory_text}\n\nHuman: {user_input}\nAI:"
# 调用大模型
response = llm.invoke(prompt).content

# 拼接问题和答案，后面要存储到长期记忆中
# 实际场景不可能每次都要存储，一般要调用llm大模型去判断是否要存储
# 就是说给大模型一个系统提示词告诉它什么类型数据才可以被存储，比如用户的爱好
memory = f"问题：{user_input} --- 答案:{response}"
memory_id = str(uuid.uuid4())
# 存储问题和答案到长期记忆
runtime.store.put(namespace, memory_id, {"memory": memory})
# 返回AI消息
return {"messages": [AIMessage(content=response)]}
创建图
def create_persistent_graph():
"""创建持久化的聊天机器人图"""
# 创建状态图
workflow = StateGraph(
    # 设置自定义的状态
    MessagesState,
    # 设置上下文的类型，配置的类型，用来传递固定的信息，这里传递的是用户id
    context_schema=MyContext
)
# 添加节点
workflow.add_node("chatbot", chatbot)
# 添加边
workflow.add_edge(START, "chatbot")
workflow.add_edge("chatbot", END)

# 编译图，使用检查点保存器和存储
graph = workflow.compile(
    # 设置检查点
    checkpointer=checkpointer,
    # 设置存储
    store=in_memory_store
)

return graph
工具函数：显示状态历史
def show_state_history(graph, config):
"""显示状态历史"""
print("\n=== 状态历史 ===")
history = graph.get_state_history(config)
for i, snapshot in enumerate(history):
print(f"\n步骤 {i}:")
print(f"  配置: {snapshot.config}")
print(f"  值: {snapshot.values}")
print(f"  下一步: {snapshot.next}")
print(f"  元数据: {snapshot.metadata}")
工具函数：显示存储的记忆
def show_memories(store, user_id):
"""显示用户的所有记忆"""
print(f"\n=== 用户 {user_id} 的记忆 ===")
namespace = (user_id, "memories")
memories = store.search(namespace)
if memories:
    for memory in memories:
        print(f"记忆ID: {memory.key}")
        print(f"内容: {memory.value}")
        print(f"创建时间: {memory.created_at}")
        print(f"更新时间: {memory.updated_at}")
        print("---")
else:
    print("没有找到记忆")
主程序
def main():
# 创建图
graph = create_persistent_graph()
# 用户配置
user_id = "user_123"
thread_id = "conversation_1"

config = {
    "configurable": {
        "thread_id": thread_id,
        "user_id": user_id
    }
}

print("=== LangGraph 持久化聊天机器人 ===")
print("输入 'quit' 退出，'history' 查看状态历史，'memories' 查看记忆")
# 无限循环
while True:
    # 获取控制台里输出的内容
    user_input = input("\n用户: ").strip()

    if user_input.lower() == 'quit':
        break
    elif user_input.lower() == 'history':
        # 打印所有的状态
        show_state_history(graph, config)
        continue
    elif user_input.lower() == 'memories':
        # 打印所有的记忆
        show_memories(in_memory_store, user_id)
        continue

    # 创建用户消息
    initial_state = {
        "messages": [HumanMessage(content=user_input)]
    }

    # 运行图
    try:
        result = graph.invoke(
            # 设置用户的问题
            initial_state,
            # 设置配置信息
            config,
            # 设置上下文
            context=MyContext(user_id)
        )

        # 显示AI回复
        ai_message = result["messages"][-1]
        print(f"AI: {ai_message.content}")

    except Exception as e:
        print(f"错误: {e}")

print("\n=== 最终状态 ===")
final_state = graph.get_state(config)
print(f"最终状态: {final_state.values}")

print("\n=== 所有记忆 ===")
show_memories(in_memory_store, user_id)
演示不同线程间的记忆共享
def demo_cross_thread_memory():
"""演示跨线程记忆共享"""
print("\n=== 跨线程记忆共享演示 ===")
graph = create_persistent_graph()
user_id = "user_456"

# 第一个对话线程
config1 = {
    "configurable": {
        "thread_id": "thread_1",
        "user_id": user_id
    }
}

print("线程1 - 建立记忆:")
result1 = graph.invoke({
    "messages": [HumanMessage(content="我叫计算机王，我喜欢音乐")]
}, config1,context=MyContext(user_id))
print(f"AI: {result1['messages'][-1].content}")

# 第二个对话线程（相同用户）
config2 = {
    "configurable": {
        "thread_id": "thread_2",
        "user_id": user_id
    }
}

print("\n线程2 - 访问记忆:")
result2 = graph.invoke({
    "messages": [HumanMessage(content="你还记得我吗？")]
}, config2,context=MyContext(user_id))
print(f"AI: {result2['messages'][-1].content}")

# 显示共享的记忆
show_memories(in_memory_store, user_id)
if name == "main":
# 运行主程序
main()
# 演示跨线程记忆共享
demo_cross_thread_memory()