• ✅ 会话级短期记忆（重启不丢）
• ✅ 跨会话长期记忆（记住用户偏好）
• ✅ 会话恢复（用同一个 thread_id 继续聊）
• ✅ 比 LangChain Agent 更灵活、可定制

一、LangGraph 内存管理核心概念

LangGraph 把记忆分成两套系统：

1. 短期记忆（会话记忆）—— Checkpointer

• 作用：保存 ** 当前会话（thread_id）** 的每一步状态快照
• 内容：messages、循环次数、中间结果、工具返回等
• 生命周期：绑定 thread_id，服务重启后仍能恢复LangGraph
• 本质：每执行完一个节点 → 自动保存一次 “游戏存档”

2. 长期记忆（跨会话记忆）—— Store（BaseStore）

• 作用：保存用户级 / 全局的长期信息（跨 thread_id）
• 内容：用户姓名、偏好、历史事实、重要记录等
• 生命周期：永久持久化，任何会话都能读取LangChain
• 本质：独立于会话的 “数据库”

3. 记忆架构对比表

维度	短期记忆（Checkpointer）	长期记忆（Store）
归属	单个 thread_id	跨 thread_id / 用户
存储内容	对话历史、状态、中间结果	用户偏好、事实、长期知识
恢复方式	用相同 thread_id 加载	按用户 ID 检索
持久化介质	内存 / Redis / Postgres	内存 / Redis / 向量库

二、为什么比 LangChain Agent 更灵活？

1. 细粒度状态控制：你可以自定义 State 所有字段，想存什么就存什么（LangChain Agent 状态固定）
2. 自动快照 + 手动恢复：每步自动存档，随时可回退到任意历史状态（时间旅行）
3. 双记忆系统解耦：会话记忆和长期记忆分开管理，可独立扩展
4. 生产级持久化：原生支持 Redis/Postgres，开箱即用LangChain

三、完整实现：带长期记忆 + 会话恢复的 ReAct Agent（GLM-4.6）

1. 安装依赖

pip install langgraph langchain-zhipu langchain-core redis

2. 完整代码（可直接运行）

import os
import redis
from typing import TypedDict, Literal, Sequence

from dotenv import load_dotenv
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_core.tools import tool
from langchain_zhipu import ChatZhipuAI
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.prebuilt import ToolNode
from langgraph.checkpoint.redis import RedisSaver
from langgraph.store.redis import RedisStore

# 加载环境
load_dotenv()

# ===================== 1. 工具定义 =====================
@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    if city == "北京":
        return "北京：晴，25℃，微风"
    elif city == "上海":
        return "上海：多云，28℃，东南风"
    else:
        return f"{city}：未知（仅支持北京/上海）"

@tool
def calculator(a: float, b: float, op: str) -> float:
    """计算器：支持 + - * /"""
    if op == "+": return a + b
    elif op == "-": return a - b
    elif op == "*": return a * b
    elif op == "/": return a / b if b != 0 else "除数不能为0"
    else: return "不支持的运算符"

tools = [get_weather, calculator]

# ===================== 2. 状态定义（可自由扩展） =====================
class ReActState(TypedDict):
    messages: Sequence[BaseMessage]  # 短期记忆：对话历史
    loop_count: int                   # 防死循环计数
    user_id: str                      # 用于绑定长期记忆

# ===================== 3. 初始化持久化组件（Redis） =====================
# 连接 Redis（需本地启动 Redis：redis-server）
redis_client = redis.Redis(host="localhost", port=6379, db=0)

# 短期记忆：Checkpointer（会话持久化）
checkpointer = RedisSaver(redis_client)

# 长期记忆：Store（跨会话持久化）
store = RedisStore(redis_client)

# ===================== 4. 初始化 LLM（智谱 GLM-4.6） =====================
llm = ChatZhipuAI(
    api_key=os.getenv("ZHIPU_API_KEY"),
    model="glm-4.6",
    temperature=0
).bind_tools(tools)

# ===================== 5. 节点定义 =====================
# 5.1 LLM 思考节点
def llm_think(state: ReActState):
    print(f"\n-- 第 {state['loop_count']} 轮：LLM 思考 --")
    messages = state["messages"]
    user_id = state["user_id"]

    # ==== 从长期记忆加载用户偏好（示例） ====
    user_preference = store.get(("user_memory", user_id), "preference")
    if user_preference:
        print(f"📌 加载长期记忆：{user_preference.value}")
        # 把长期记忆注入 prompt
        messages = [HumanMessage(content=f"用户偏好：{user_preference.value}")] + messages

    response = llm.invoke(messages)

    return {
        "messages": messages + [response],
        "loop_count": state["loop_count"] + 1,
        "user_id": user_id
    }

# 5.2 工具执行节点
tool_executor = ToolNode(tools)

# 5.3 条件路由（继续/结束）
def should_continue(state: ReActState) -> Literal["continue", "end"]:
    last_msg = state["messages"][-1]
    if not last_msg.tool_calls or state["loop_count"] >= 5:
        print("→ 结束：无需工具 或 达到最大轮次")
        return "end"
    print("→ 继续：需要调用工具")
    return "continue"

# ===================== 6. 构建图（带持久化） =====================
builder = StateGraph(ReActState)

builder.add_node("llm_think", llm_think)
builder.add_node("tool_executor", tool_executor)

builder.add_edge(START, "llm_think")
builder.add_conditional_edges("llm_think", should_continue, {"continue": "tool_executor", "end": END})
builder.add_edge("tool_executor", "llm_think")

# ✅ 关键：编译时绑定 checkpointer 和 store
react_agent = builder.compile(
    checkpointer=checkpointer,  # 短期记忆（会话恢复）
    store=store                  # 长期记忆（跨会话）
)

# ===================== 7. 测试：会话恢复 + 长期记忆 =====================
if __name__ == "__main__":
    print("===== LangGraph 长期记忆 + 会话恢复 Agent =====")

    # 配置：同一个 thread_id 代表同一会话，user_id 绑定长期记忆
    config = {
        "configurable": {
            "thread_id": "session_001",  # 会话ID（恢复时用同一个）
            "user_id": "user_001"        # 用户ID（长期记忆绑定）
        }
    }

    # ==== 第一轮对话（自动保存到 Redis） ====
    print("\n--- 第一轮对话 ---")
    initial_state = {
        "messages": [HumanMessage(content="北京今天天气？上海比北京高3度，上海多少度？")],
        "loop_count": 0,
        "user_id": "user_001"
    }
    result1 = react_agent.invoke(initial_state, config=config)
    print("第一轮答案：", result1["messages"][-1].content)

    # ==== 写入长期记忆（保存用户偏好） ====
    store.put(("user_memory", "user_001"), "preference", "喜欢用摄氏度，偏好简洁回答")
    print("\n✅ 已保存长期记忆：用户偏好")

    # ==== 模拟服务重启：用同一个 thread_id 恢复会话 ====
    print("\n--- 模拟重启：恢复会话（thread_id=session_001）---")
    result2 = react_agent.invoke(
        {"messages": [HumanMessage(content="上海天气符合我的偏好吗？")]},
        config=config
    )
    print("恢复后答案：", result2["messages"][-1].content)

运行结果

✅️ Redis store 索引已创建
===== LangGraph 会话恢复 Agent =====

--- 第一轮对话 ---

-- 第 0轮：LLM思考 --
📌 加载长期记忆：喜欢用摄氏度，偏好简介回答
→ 结束： 无需工具 或 达到最大轮次
第一轮答案：北京今天天气：晴，25℃，微风

上海比北京高3度，上海是28度

上海天气符合您的摄氏度偏好。

 ✅️ 已保存长期记忆：用户偏好

--- 模拟服务重启：恢复会话（thread_id=session_001 ---）

-- 第 1轮：LLM思考 --
📌 加载长期记忆：喜欢用摄氏度，偏好简介回答
→ 继续：需要调用工具
恢复后的答案： 上海：多云，28℃，东南风

进程已结束，退出代码为 0

四、关键功能验证

1. 会话恢复（短期记忆）

• 运行代码 → 第一轮对话完成
• 停止程序 → 重新运行
• 用同一个 thread_id 调用 → Agent 自动加载之前的对话历史，继续聊

2. 长期记忆（跨会话）

• 代码中 store.put() 保存用户偏好
• 新会话（不同 thread_id）只要用同一个 user_id → 自动加载偏好

3. 持久化介质

• 示例用 Redis（生产推荐）
• 开发可改用 MemorySaver（内存，重启丢失）
• 生产可换 PostgresSaver/PostgresStore

五、总结：LangGraph 记忆优势

1. 双记忆架构：会话记忆（恢复）+ 长期记忆（跨会话）
2. 状态完全可控：自定义 State，存任何你需要的信息
3. 自动持久化：每步自动快照，无需手动写数据库
4. 无缝恢复：指定 thread_id 即可回到任意会话状态LangGraph

六、Docker安装redis-stack

4. 运行 Redis Stack
docker run -d --name redis-stack -p 6379:6379 redis/redis-stack-server:latest

如果你的本地 redis-cli 已经能正常连接，也可以用本地连接：

redis-cli -h 127.0.0.1 -p 6379

2. 执行查看模块命令

进入 Redis 交互界面后，输入：

MODULE LIST

3. 查看返回结果

如果返回的列表中包含 name 为 search 的项，说明 RediSearch 已成功加载：

2. 执行测试命令

执行一个最简单的 RediSearch 命令，例如列出所有索引（即使没有索引也不会报错）：

FT._LIST

或者尝试创建一个测试索引：

FT.CREATE test_idx ON HASH PREFIX 1 test: SCHEMA title TEXT