# LangGraph 入门全解(三)：让机器人拥有记忆--集成 Mem0，大模型入门到精通，收藏这篇就足够了！

AI-智能

1017人浏览 · 2025-09-10 15:40:18

AI-智能 · 2025-09-10 15:40:18 发布

LangGraph添加记忆(任务级别记忆)

在LangGraph中又被称为检查点（checkpointing），它是 LangGraph 能够支持长周期任务执行和断点续执行的关键机制。

在前两节中，我们通过手动维护一个 messages 列表来实现多轮对话。然而，这种方式在面对多用户、多任务以及复杂图结构时，维护成本会急剧上升。

添加记忆模块

为了解决这一问题，LangGraph 提供了更为简洁的记忆模块。只需修改3行代码，就能启用内存记忆：

from langgraph.checkpoint.memory import InMemorySaver
memory = InMemorySaver()
# ...省略重复代码
app = graph_builder.compile(checkpointer=memory)

在调用时，只需配置 thread_id：

# ...省略重复代码，注意 config thread_id的配置
response = app.invoke({"messages": {"role": "user", "content": user_input}},
                      config = {"configurable": {"thread_id": "1"}})

这样便实现了基于内存的记忆。但需要注意的是：一旦程序重启，记忆会丢失。

使用 Redis 持久化记忆

为了避免重启导致数据丢失，我们可以改用 Redis 作为记忆存储。只要 Redis 保持正常运行，记忆就能持续存在。

1. 安装医依赖
```
pip install langgraph-checkpoint-redis
```

修改代码：

from redis import Redis
from langgraph.checkpoint.redis import RedisSaver

redis_client = Redis(
    host="127.0.0.1",
    port=6379,
    password="Tali1234"
)
memory = RedisSaver(redis_client=redis_client)

Redis 中的存储内容大致如下：

在 PyPI 上搜索 langgraph-checkpoint，还能发现更多数据库的存储实现，方便不同场景使用。

效果如下

LangGraph集成Mem0(用户级别记忆)

任务级别记忆帮助我们保持单个对话的上下文， Mem0 更进一步，提供了跨任务、跨会话的用户级别记忆。

Mem0 是一个轻量化、模块化的智能记忆层框架，其目标是：

“通过保留用户偏好与上下文信息，随着时间推移不断适应，提供个性化和高效的交互体验。”

它支持对话层、用户层和 Agent 层的多层记忆组织方式，常用于聊天机器人、RPA 自动化助手、AI Copilot 等场景。

Mem0 的记忆流程分为提取和更新两个阶段：

在提取阶段，系统会从最新对话、滚动摘要、最近 m 条对话中抓取关键信息，并借助 LLM 生成简洁候选记忆。
在后台，长期摘要会异步更新，从而保证推理流程不会被打断。

安装 Mem0

pip install mem0ai

示例代码（核心逻辑已标注）

from typing import Annotated
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langchain_tavily import TavilySearch
from langgraph.prebuilt import ToolNode, tools_condition
from mem0 import Memory
from langchain_openai import OpenAIEmbeddings

openai_embeddings = OpenAIEmbeddings(
    model="gte-large-zh",
    check_embedding_ctx_length=False
)
# 此处定义你自己的模型
llm = ChatOpenAI(model="qwen3_32")

# 定义工具
tool = TavilySearch(max_results=2)
tools = [tool]
# 工具绑定到模型
llm_with_tools = llm.bind_tools(tools)
# 创建Mem0记忆配置
config = {
"vector_store": {
"provider": "milvus",
"config": {
"collection_name": "mem0",
"embedding_model_dims": "1024",
"url": "http://127.0.0.1:9530",
"token": "root:root",
        },
    },
"embedder": {
"provider": "langchain",
"config": {
"model": openai_embeddings,
        },
    },
"llm": {
"provider": "openai",
"config": {
"model": "qwen3_32",
"temperature": 0.2,
"max_tokens": 20000,
        },
    },
}
memory = Memory.from_config(config)

# 定义图状态
class State(TypedDict):
    messages: Annotated[list, add_messages]  # 此处维护完整的消息历史
    mem0_user_id: str # 增加用户id

graph = StateGraph(State)

def chatbot(state: State):
    messages = state["messages"]
    user_id = state["mem0_user_id"]

# 检索记忆
    memories = memory.search(messages[-1].content, user_id=user_id)

    context = "Relevant information from previous conversations:\n"
for m in memories.get("results", []):
        context += f"- {m['memory']}\n"

    system_message = {"role":"system", "content": f"""You are a helpful customer support assistant. Use the provided context to personalize your responses and remember user preferences and past interactions.
{context}"""}

    full_messages = [system_message] + messages
    response = llm_with_tools.invoke(full_messages)

# 存储记忆
    memory.add(
f"User: {messages[-1].content}\nAssistant: {response.content}", user_id=user_id
    )
return {"messages": [response]}

# 使用LangGraph提供的工具节点
tool_node = ToolNode(tools=tools)

graph.add_node("chatbot", chatbot)
# 添加工具节点
graph.add_node("tools", tool_node)
# 添加工具 条件分支
graph.add_conditional_edges(
"chatbot",
    tools_condition,
)
graph.add_edge("tools", "chatbot")
graph.add_edge(START, "chatbot")

app = graph.compile()

if __name__ == "__main__":
while True:
        user_input = input("👨‍💻: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Exiting...")
break
        response = app.invoke({"messages": {"role": "user", "content": user_input}, "mem0_user_id":"mem0_wiley"})
        messages = response["messages"]
print(f'🤖: {response["messages"][-1].content}')