LangGraph Memory记忆系统：告别”金鱼脑”

你有没有遇到过这种情况：跟AI助手聊了半天，它突然忘了你叫啥？这就是典型的”金鱼脑”——七秒记忆，刚说完就忘。

今天咱们来聊聊 LangGraph 的 Memory 记忆系统，让你的 AI Agent 真正拥有”记性”。

引言：为什么需要记忆？

想象一下你走进一家咖啡店，店员问你要喝什么。你说：”来杯拿铁，少糖。”店员记下了。你又说：”哦对了，用燕麦奶。”店员点头。接着你补充：”还有，我叫小明，是老顾客了，上次存的积分记得用上。”

如果这位店员是个”金鱼脑”，会发生什么？

• 你刚说完燕麦奶，他就忘了你要的是拿铁
• 你说积分的时候，他问你：”您叫什么名字来着？”
• 等你坐下来，他又来问：”先生您要点什么？”

这就是没有记忆系统的 AI Agent 的写照。

短期记忆 vs 长期记忆

人类记忆分两种：

短期记忆（Short-term Memory）

• 临时保存当前对话的信息
• 容量有限，通常 5-9 个项目
• 持续几分钟到几小时
• 例如：记住对话中提到的名字、偏好

长期记忆（Long-term Memory）

• 持久保存重要信息
• 容量几乎无限
• 可以保存数天、数月甚至永久
• 例如：用户的名字、常用偏好、历史对话

在 AI Agent 中，这两种记忆分别对应：

记忆类型	LangGraph 实现	使用场景
短期记忆	InMemorySaver	开发调试、单次对话
长期记忆	SqliteSaver / PostgresSaver	生产环境、持久化存储

第一部分：InMemorySaver——开发者的”草稿纸”

什么是 InMemorySaver？

InMemorySaver 是 LangGraph 提供的最简单的记忆存储方式。顾名思义，它把记忆存在内存里。

优点：

• 配置简单，一行代码搞定
• 速度快，内存读写效率极高
• 不需要额外依赖

缺点：

• 程序重启，记忆归零
• 无法跨进程共享
• 不适合生产环境

代码示例：第一个”有记忆”的 Agent

让我们从一个最简单的例子开始——一个能记住用户偏好的助手：

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI

# 1. 创建记忆存储（内存版）
memory = InMemorySaver()

# 2. 初始化模型
model = ChatOpenAI(model="gpt-4o-mini")

# 3. 定义节点：调用模型
def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

# 4. 构建图
builder = StateGraph(MessagesState)
builder.add_node("agent", call_model)
builder.add_edge(START, "agent")
builder.add_edge("agent", END)

# 5. 编译时加入记忆
graph = builder.compile(checkpointer=memory)

# 6. 运行对话
config = {"configurable": {"thread_id": "conversation_1"}}

# 第一轮对话
response1 = graph.invoke(
    {"messages": [{"role": "user", "content": "你好，我叫小明"}]},
    config
)
print("AI:", response1["messages"][-1].content)

# 第二轮对话（看看它还记得吗）
response2 = graph.invoke(
    {"messages": [{"role": "user", "content": "我叫什么名字？"}]},
    config
)
print("AI:", response2["messages"][-1].content)

运行结果：

AI: 你好小明！很高兴认识你。
AI: 你叫小明呀，我记得的。

见证奇迹的时刻！ 第二轮对话中，AI 没有问你”你是谁”，而是准确地记住了你叫小明。这就是记忆系统的魔力。

关键概念：thread_id

你可能注意到代码里的 thread_id="conversation_1"。这是什么意思？

thread_id 是记忆的”身份证”。每个唯一的 thread_id 对应一份独立的记忆空间。

# 用户 A 的对话
config_a = {"configurable": {"thread_id": "user_a_chat"}}

# 用户 B 的对话  
config_b = {"configurable": {"thread_id": "user_b_chat"}}

这样，用户 A 和用户 B 的记忆是隔离的，互不干扰。

可视化记忆状态

想知道当前记忆里存了什么？可以用 get_state 方法：

# 查看当前状态
state = graph.get_state(config)
print("消息历史：", len(state.values["messages"]), "条")
for msg in state.values["messages"]:
    print(f"- {msg.type}: {msg.content[:50]}...")

输出示例：

消息历史： 4 条
- human: 你好，我叫小明
- ai: 你好小明！很高兴认识你。
- human: 我叫什么名字？
- ai: 你叫小明呀，我记得的。

第二部分：SqliteSaver——轻量级持久化

为什么需要持久化？

用 InMemorySaver 开发调试没问题，但生产环境有两个致命问题：

1. 程序重启，记忆丢失——用户明天再聊，AI 又成陌生人了
2. 多进程无法共享——API 服务通常多个 worker，每个 worker 的记忆是独立的

解决方案：持久化存储。

SqliteSaver 登场

SqliteSaver 使用 SQLite 数据库存储记忆，解决了上述两个问题：

优点：

• 数据持久化，重启不丢
• 单文件数据库，零配置
• 支持多进程读写
• 轻量级，无需额外服务

缺点：

• 并发性能不如 PostgreSQL
• 不适合超大规模数据

代码示例：持久化的记忆

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
import sqlite3

# 1. 创建持久化的记忆存储
conn = sqlite3.connect("chat_memory.db", check_same_thread=False)
memory = SqliteSaver(conn)

# 2. 其余代码相同
model = ChatOpenAI(model="gpt-4o-mini")

def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": [response]}

builder = StateGraph(MessagesState)
builder.add_node("agent", call_model)
builder.add_edge(START, "agent")
builder.add_edge("agent", END)

# 3. 编译时加入 SqliteSaver
graph = builder.compile(checkpointer=memory)

# 4. 运行对话
config = {"configurable": {"thread_id": "persistent_chat_1"}}

response = graph.invoke(
    {"messages": [{"role": "user", "content": "记住，我的最爱是抹茶拿铁"}]},
    config
)
print("AI:", response["messages"][-1].content)

关键区别： 现在关闭程序，重新运行，AI 依然记得你的最爱是抹茶拿铁。

查看 SQLite 数据库

好奇数据是怎么存的？用命令行看看：

sqlite3 chat_memory.db ".tables"
# 输出：checkpoints

sqlite3 chat_memory.db "SELECT * FROM checkpoints;"
# 输出：thread_id, checkpoint_ns, checkpoint_id, parent_checkpoint_id, type, checkpoint, metadata

进阶：异步支持

生产环境通常使用异步，LangGraph 也提供了异步版本：

from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

# 异步连接
async with AsyncSqliteSaver.from_conn_string("chat_memory.db") as memory:
    graph = builder.compile(checkpointer=memory)
    # ... 异步调用

第三部分：实战——认识你的”老朋友”

现在让我们做一个更实用的例子：一个能记住用户偏好的个性化助手。

场景设定

想象一个咖啡订购助手，它需要记住：

• 用户的名字
• 常点的咖啡
• 口味偏好（糖度、奶类）

完整代码

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
import sqlite3
import operator

# 1. 定义状态（扩展 MessagesState）
class CoffeeState(MessagesState):
    user_name: str
    favorite_coffee: str
    preferences: Annotated[list, operator.add]  # 累加偏好

# 2. 创建持久化记忆
conn = sqlite3.connect("coffee_memory.db", check_same_thread=False)
memory = SqliteSaver(conn)

# 3. 初始化模型
model = ChatOpenAI(model="gpt-4o-mini")

# 4. 定义智能节点：理解并记忆用户信息
def understand_user(state: CoffeeState):
    messages = state["messages"]
    last_message = messages[-1].content
    
    # 让 AI 分析用户输入，提取关键信息
    extraction_prompt = f"""
    分析用户的消息，提取以下信息（JSON格式）：
    - user_name: 用户的名字（如果有）
    - favorite_coffee: 提到的咖啡种类（如果有）
    - preferences: 口味偏好（如少糖、燕麦奶等，数组格式）
    
    如果某项不存在，设为 null。
    
    用户消息：{last_message}
    """
    
    extraction = model.invoke([
        {"role": "system", "content": "你是一个信息提取助手，只返回JSON。"},
        {"role": "user", "content": extraction_prompt}
    ])
    
    import json
    try:
        info = json.loads(extraction.content)
        updates = {}
        if info.get("user_name"):
            updates["user_name"] = info["user_name"]
        if info.get("favorite_coffee"):
            updates["favorite_coffee"] = info["favorite_coffee"]
        if info.get("preferences"):
            updates["preferences"] = info["preferences"]
        return updates
    except:
        return {}

# 5. 定义节点：生成回复
def generate_response(state: CoffeeState):
    user_name = state.get("user_name", "顾客")
    favorite = state.get("favorite_coffee", "")
    prefs = state.get("preferences", [])
    
    # 构建个性化提示
    system_msg = f"""你是咖啡助手。当前服务顾客：{user_name}
    已知信息：最爱{favorite if favorite else '待定'}，偏好{', '.join(prefs) if prefs else '无'}
    语气友好，可以适当提及已知信息让顾客感到亲切。"""
    
    response = model.invoke(
        [{"role": "system", "content": system_msg}] + state["messages"]
    )
    return {"messages": [response]}

# 6. 构建图
builder = StateGraph(CoffeeState)
builder.add_node("understand", understand_user)
builder.add_node("respond", generate_response)

builder.add_edge(START, "understand")
builder.add_edge("understand", "respond")
builder.add_edge("respond", END)

graph = builder.compile(checkpointer=memory)

# 7. 测试对话
config = {"configurable": {"thread_id": "coffee_lover_123"}}

print("=== 第一次对话 ===")
result = graph.invoke(
    {"messages": [{"role": "user", "content": "你好，我是张伟，喜欢喝美式"}]},
    config
)
print("AI:", result["messages"][-1].content)

print("\n=== 第二次对话（重启程序后）===")
# 模拟重启：重新加载图
conn2 = sqlite3.connect("coffee_memory.db", check_same_thread=False)
memory2 = SqliteSaver(conn2)
graph2 = builder.compile(checkpointer=memory2)

result2 = graph2.invoke(
    {"messages": [{"role": "user", "content": "今天想换换口味"}]},
    config
)
print("AI:", result2["messages"][-1].content)

print("\n=== 第三次对话 ===")
result3 = graph2.invoke(
    {"messages": [{"role": "user", "content": "加点燕麦奶"}]},
    config
)
print("AI:", result3["messages"][-1].content)

print("\n=== 第四次对话（检查是否记住所有信息）===")
result4 = graph2.invoke(
    {"messages": [{"role": "user", "content": "我是谁？我喜欢什么？"}]},
    config
)
print("AI:", result4["messages"][-1].content)

运行结果示例

=== 第一次对话 ===
AI: 你好张伟！欢迎光临，美式咖啡是个不错的选择，需要我为你准备一杯吗？

=== 第二次对话（重启程序后） ===
AI: 嗨张伟！今天想尝试点什么新口味呢？你平时最喜欢美式，要不要试试我们的冷萃或者加点风味糖浆？

=== 第三次对话 ===
AI: 没问题！美式加燕麦奶，这是个健康的搭配。需要加糖吗？

=== 第四次对话（检查是否记住所有信息） ===
AI: 你是张伟呀！你喜欢喝美式咖啡，今天还加了燕麦奶。需要我现在为你做一杯吗？

看到没有？ 即使程序重启，AI 依然记得：

• 你的名字是张伟
• 你喜欢美式咖啡
• 你刚刚加了燕麦奶

这就是持久化记忆的威力！

第四部分：记忆管理——什么时候该清空？

记忆不是越多越好

想象你的房间——东西越多，找东西越难。AI 的记忆也一样：

• 对话太长：Token 消耗增加，成本上升
• 信息过多：模型可能”迷失”在信息中
• 旧信息干扰：三年前的偏好可能不再适用

策略一：限制对话长度

def trim_messages(messages, max_count=10):
    """只保留最近 N 条消息"""
    return messages[-max_count:]

# 在节点中使用
def call_model(state: MessagesState):
    trimmed = trim_messages(state["messages"], 10)
    response = model.invoke(trimmed)
    return {"messages": [response]}

策略二：总结压缩

当对话太长时，可以总结历史，只保留关键信息：

from langchain_core.messages import SystemMessage

def summarize_history(messages):
    """将长对话总结为关键信息"""
    summary_prompt = f"""
    总结以下对话的关键信息（用户偏好、重要事实）：
    {messages}
    
    用一句话概括。
    """
    summary = model.invoke([{"role": "user", "content": summary_prompt}])
    return SystemMessage(content=f"[历史摘要] {summary.content}")

# 在合适的时候调用
if len(state["messages"]) > 20:
    summary = summarize_history(state["messages"][:15])
    new_messages = [summary] + state["messages"][15:]

策略三：清空记忆

有时候，用户就是想”重新开始”。

# 方法一：使用新的 thread_id（推荐）
new_config = {"configurable": {"thread_id": "conversation_new"}}

# 方法二：删除特定 thread 的记忆
memory.delete(thread_id="conversation_old")

策略四：定期归档

对于需要长期保存但不常访问的记忆，可以定期归档到外部存储：

def archive_old_memories(thread_id, days=30):
    """将旧记忆归档到外部存储"""
    old_state = memory.get(thread_id)
    # 保存到数据库/文件
    # 清空当前记忆

总结与下篇预告

今天学了什么？

1. 为什么需要记忆——没有记忆的 AI 就像金鱼，七秒就忘
2. 短期记忆 InMemorySaver——开发调试的好帮手
3. 长期记忆 SqliteSaver——轻量级持久化方案
4. 实战应用——记住用户偏好的咖啡助手
5. 记忆管理——清空、限制、总结，避免记忆过载

记忆系统的选择指南

场景	推荐方案	理由
开发调试	InMemorySaver	简单快捷，零配置
单机应用	SqliteSaver	持久化，开箱即用
生产服务	PostgresSaver	高并发，可扩展
分布式系统	RedisSaver	共享状态，低延迟

下篇预告

《LangGraph 状态管理：打造复杂多轮对话 Agent》

我们将深入探讨：

• 如何设计复杂的状态结构
• 条件边（Conditional Edges）的高级用法
• 人机协作：Human-in-the-loop 模式
• 子图（Subgraphs）：模块化你的 Agent

---

思考题： 如果你要做一个”心理咨询师 AI”，需要记住用户的哪些信息？是越详细越好吗？欢迎在评论区讨论。

延伸阅读：

• LangGraph 官方文档 - Persistence
• SQLite vs PostgreSQL 选型指南

---

文章首发于 Channing’s AI Lab，转载请联系作者。