摘要：本文介绍如何让大模型记住与用户的交互并从中学习偏好，从而在复杂、多轮交互任务中提升效率与满意度。文章从内存存储过渡到使用 Redis 持久化，说明 LangGraph 目前支持的 Postgres、MongoDB、Redis、Oracle 四种数据库，以及通过自定义检查点接入其他数据库的思路；随后演示安装 langgraph-checkpoint-redis、创建 Redis 长连接与 store、checkpointer，为 LangGraph 图配置 Redis，并实现切换会话（线程）、基于运行时 store 的检索与写入长期记忆，最后给出完整可运行的 Python 示例与效果图。

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

我们需要让大模型记住与它之间的交互，然后让大模型从中进行学习找到用户的喜爱偏好，在处理复杂的任务、需要大量和用户交互的操作时这种学习的能力会提高效率和用户的满意程度

之前都是通过内存存储的，接下来使用Redis存储（如果没有redis可以使用Docker下载redis）

如下图红框LangGraph支持的数据库，Postgres、MongoDB、Redis、Oracle，现在只支持这4个

如果想使用其它的数据库，可以参考下图红框的自定义检查点来实现，一般使用

地址：Memory - Docs by LangChain

redis中如果想使用向量检索，需要安装redis-stack

安装redis库

uv add langgraph-checkpoint-redis

代码说明

如下图红框，创建Redis的长连接，创建store和checkpointer，with下方的代码执行完成后，它会自动关闭后面的资源，也就是自动关闭Redis的长连接

然后给LangGraph图设置redis

如下图红框它的提问，代码中有实现切换会话的功能，也就是切换线程

Redis的查询，直接通过获取运行时里面的store，然后.search就可以了

它的存储如下图红框，也是通过运行时获取store，然后.put

效果图：

代码

import uuid
import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.redis import RedisSaver
from langgraph.store.redis import RedisStore
from langgraph.runtime import Runtime
from dataclasses import dataclass

load_dotenv()

# --- 1. 初始化模型 ---
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'
)


@dataclass
class Context:
    user_id: str


# --- 定义节点逻辑 ---
def call_model(state: MessagesState, runtime: Runtime[Context]):
    user_id = runtime.context.user_id
    namespace = ("memories", user_id)

    # 检索长期记忆
    last_user_msg = state["messages"][-1].content
    memories = runtime.store.search(namespace, query=str(last_user_msg))
    info = "\n".join([d.value["data"] for d in memories])

    system_msg = f"你是一个有帮助的助手。已知用户信息: {info}"

    # 逻辑存储：如果用户说“记住...”，则存入 Store
    if "记住" in last_user_msg:
        # 简单提取“记住”后面的内容（实际生产可用LLM提取）
        memory_content = last_user_msg.replace("记住", "").strip("：: ")
        runtime.store.put(namespace, str(uuid.uuid4()), {"data": memory_content})
        print(f"--- [系统日志] 已存入长期记忆: {memory_content} ---")

    response = llm.invoke(
        [{"role": "system", "content": system_msg}] + state["messages"]
    )
    return {"messages": response}


# --- 3. 构建图 ---
DB_URI = "redis://127.0.0.1:6379"

# 使用 context manager 保持连接
with RedisStore.from_conn_string(DB_URI) as store, \
        RedisSaver.from_conn_string(DB_URI) as checkpointer:
    # 第一次初始化的时候需要
    checkpointer.setup()
    store.setup()

    builder = StateGraph(MessagesState)
    builder.add_node("call_model", call_model)
    builder.add_edge(START, "call_model")
    graph = builder.compile(checkpointer=checkpointer, store=store)

    # --- 4. 交互循环 ---
    current_thread_id = "1"
    current_user_id = "user_v1"

    print("=== LangGraph 交互系统 ===")
    print("指令说明: 输入 'switch' 切换会话, 'exit' 退出程序")

    while True:
        prompt = f"\n[当前线程: {current_thread_id}] 用户: "
        user_input = input(prompt).strip()

        if user_input.lower() == 'exit':
            break

        if user_input.lower() == 'switch':
            new_id = input("请输入新的 Thread ID (例如 1, 2, 3): ")
            current_thread_id = new_id
            print(f"--- 已切换到线程 {current_thread_id} ---")
            continue

        if not user_input:
            continue

        # 构建配置
        config = {
            "configurable": {
                "thread_id": current_thread_id,
                "user_id": current_user_id,
            }
        }

        # 执行流式输出（使用 values 模式）
        # 注意：由于我们要手动输入，每次流只传入当前这一条消息
        for chunk in graph.stream(
                {"messages": [{"role": "user", "content": user_input}]},
                config,
                context=Context(user_id=current_user_id),
                stream_mode="messages",
                version="v2"
        ):
            if chunk["type"] == "messages":
                result, metadata = chunk["data"]
                print(result.content, end="", flush=True)

---