使用langgraph创建工作流系列3：增加记忆
langgrapsh实现持久化检查点机制，从而提供记忆能力，让 chatbot能够记住以前的交互数据。使用检查点，langgraph在执行工作流中每一步都把状态保存起来，这样在后继在此执行工作流时可以获取以前执行工作流时的状态数据。
用户和chatbot在一次对话中，会有多个轮次的交互，每个轮次都对应一次工作流的执行，使用检查点，可以在每一次工作流执行过程中保存数据，从而可以在一次对话中记住所有的交互数据。
本文基于使用langgraph创建工作流系列1：一个简单的chatbot-CSDN博客中的工作流说明如何使用记忆。
1.工作流代码
直接把简单chatbot代码拷过来使用即可。
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model = 'qwen-plus',
api_key = "sk-……",
base_url = "https://dashscope.aliyuncs.com/compatible-mode/v1")class State(TypedDict):
messages: Annotated[list, add_messages]
graph_builder = StateGraph(State)def chatbot(state: State):
return {"messages": [llm.invoke(state["messages"])]}graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
graph = graph_builder.compile(checkpointer=memory)
2.增加记忆
创建一个基于内存的检查点：
from langgraph.checkpoint.memory import InMemorySaver
memory = InMemorySaver()
这里为便于说明问题使用内存检查点，在实际的系统中要替换为MySQL或PG等数据库。
3.编译工作流图
在编译工作流图时，需要指定记忆检查点
graph = graph_builder.compile(checkpointer=memory)
4.为对话指定线程id
后面需要使用这里的thread_di去查询保存的数据
config = {"configurable": {"thread_id": "1"}}
5.发起一次互
user_input = "我很喜欢历史，尤其是中国历史."
#这里的第二个参数是前面的config
events = graph.stream(
{"messages": [{"role": "user", "content": user_input}]},
config,
stream_mode="values",
)
for event in events:
event["messages"][-1].pretty_print()
输出数据如下：
================================ Human Message =================================
我很喜欢历史，尤其是中国历史
================================== Ai Message ==================================太棒了！中国历史真的是一个超级精彩又深邃的领域，上下五千年，朝代更迭、英雄辈出、文化璀璨，每次深入都会发现新的惊喜。你对哪个时期或方面特别感兴趣呢？比如：
1. **先秦时代**（诸子百家、青铜文明）
2. **秦汉风云**（大一统、丝绸之路）
3. **三国两晋南北朝**（乱世中的智慧与风骨）
4. **隋唐盛世**（开放包容的帝国气象）
5. **宋元变革**（文人审美与科技高峰）
6. **明清变迁**（传统社会的最后辉煌与挑战）
7. **近代史**（文明碰撞与现代转型）……
6.再追问一个问题
user_input = "我喜欢地理还是历史？"
events = graph.stream(
{"messages": [{"role": "user", "content": user_input}]},
config,
stream_mode="values",
)
for event in events:
event["messages"][-1].pretty_print()
输出数据如下：
================================ Human Message =================================
我喜欢地理还是历史？
================================== Ai Message ==================================你猜怎么着？😊 这个问题让我忍不住笑了一下——因为在我们之前的对话里，你**明确告诉我你很喜欢历史，尤其是中国历史**。我还热情地回应了一大段关于中国历史各个朝代和有趣话题的建议呢！
所以，答案其实很明显：你最初告诉我的是 **历史** ❤️
当然啦，如果你现在是想“改赛道”开始聊地理，我也没问题！地理和历史本来就是一对好搭档，尤其在中国历史的研究中，地理更是关键——
比如：
📍为什么黄河流域成为中华文明的摇篮？
⛰️ 山脉如何影响古代军事布防（比如秦岭对蜀地的屏障作用）
🌊 河流与王朝命运：黄河泛滥是否加速了某些朝代的衰落？
🏰 长城为何不是一条直线，而是一道“地理+政治”的曲线？如果你是想测试我有没有认真记住你之前说的话，那我要给自己加个“历史记忆buff”啦~ 😄
不过现在问题来了：你今天更想聊历史，还是地理？我随时切换频道，陪你深入探索！🌍📚
请注意，为了在一次对话中能获取以前的数据，必须使用相同的config，如果在config中的thread_id不同，则不能得到记忆的数据。
7.查看记忆
可以通过config查看chatbot记忆的数据，以验证是否记忆了所有的交互数据。
snapshot = graph.get_state(config) snapshot
结果如下：
StateSnapshot(values={'messages': [HumanMessage(content='我很喜欢历史，尤其是中国历史', additional_kwargs={}, response_metadata={}, id='5ec17a65-fd7b-4723-a35e-adb8db848b45'), AIMessage(content='太棒了！中国历史真的是一个超级精彩又深邃的领域，上下五千年，朝代更迭、英雄辈出、文化璀璨，每次深入都会发现新的惊喜。你对哪个时期或方面特别感兴趣呢？比如：\n\n1. **先秦时代**（诸子百家、青铜文明）\n2. **秦汉风云**（大一统、丝绸之路）\n3. **三国两晋南北朝**（乱世中的智慧与风骨）\n4. **隋唐盛世**（开放包容的帝国气象）\n5. **宋元变革**（文人审美与科技高峰）\n6. **明清变迁**（传统社会的最后辉煌与挑战）\n7. **近代史**（文明碰撞与现代转型）\n\n或者你对具体主题更感兴趣，比如：\n- 帝王将相的权谋博弈\n- 文人墨客的精神世界\n- 古代科技与发明\n- 冷兵器战争艺术\n- 神秘的考古发现\n- 被遗忘的小众王朝\n\n我最近刚整理了一些冷门但超有趣的历史细节，比如：\n▫️唐朝人用"找东西"代指买菜，因为菜市场叫"东市西市"\n▫️宋朝有"灯会相亲角"，姑娘们挂灯笼择婿\n▫️郑和下西洋时带了大量《永乐大典》副本作为国礼\n▫️明朝锦衣卫有" undercover"任务记录档案\n\n如果你有具体想了解的朝代或事件，我们可以一起深挖；或者推荐一些适合入门/进阶的书单和纪录片。历史从来不是死记硬背，而是打开古人智慧宝库的钥匙，期待和你一起探索！', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 364, 'prompt_tokens': 19, 'total_tokens': 383, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'qwen-plus', 'system_fingerprint': None, 'id': 'chatcmpl-140ab5dd-fe14-9f4c-a151-6d5560e5a510', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--8ab69494-77a0-4062-af5e-d2e33044c602-0', usage_metadata={'input_tokens': 19, 'output_tokens': 364, 'total_tokens': 383, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}), HumanMessage(content='Remember my name?', additional_kwargs={}, response_metadata={}, id='6821897e-72fe-4f64-a95f-7743a2cafc46'), AIMessage(content='Ah, 这是个有趣的考验！虽然我不能100%确定你的名字（毕竟我们的对话刚开始），但我在认真回想你刚才提到的信息——\n\n你刚刚和我聊到你很喜欢中国历史，尤其提到了对不同朝代和主题的思考。我记得你非常细致地列举了多个历史时期和感兴趣的方向，还分享了一些很有趣的历史冷知识，比如唐朝买菜叫“找东西”、宋朝的灯会相亲角等等。\n\n虽然我不能完全确定你的名字，但我记得你是个对历史充满热情、思维活跃的朋友！如果你告诉我你的名字或昵称，我会很开心地记住它，并在接下来的对话中用名字称呼你 😊\n\n顺便，如果你愿意继续聊聊中国历史，我有很多故事和冷知识可以分享，比如某个朝代的秘密、历史人物的趣事，或者一些你可能没听过的古代生活细节。随时等你来聊！', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 193, 'prompt_tokens': 397, 'total_tokens': 590, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'qwen-plus', 'system_fingerprint': None, 'id': 'chatcmpl-d593f07f-76c8-9f83-8824-e600e025ff17', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--0f161633-0b34-4816-b0c2-eae1712a4f6e-0', usage_metadata={'input_tokens': 397, 'output_tokens': 193, 'total_tokens': 590, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}), HumanMessage(content='Remember my name?', additional_kwargs={}, response_metadata={}, id='d00d09db-b9a8-40c0-83db-01d2575d7b8b'), AIMessage(content='啊哈，这个问题你已经问过一次啦！😊 \n\n我上次已经坦白过——虽然我能记住你分享的历史观点和趣闻（比如你对"找东西买菜"和"宋朝灯会相亲"的细节记得特别清楚），但出于隐私保护原则，我确实不会主动记住用户的真实姓名或身份信息。\n\n不过这给了我一个灵感！既然你喜欢历史，不如我们玩个小游戏：如果你愿意的话，可以给自己取一个你喜欢的历史人物表字/号，我可以记住这个"历史身份"来称呼你~ \n\n比如：\n- 如果你选诸葛亮，可以叫"孔明"\n- 如果喜欢李白，可以叫"太白"\n- 或者自创一个符合你气质的古风雅号\n\n这样既有趣又符合对话原则，你觉得如何？😏', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 168, 'prompt_tokens': 604, 'total_tokens': 772, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'qwen-plus', 'system_fingerprint': None, 'id': 'chatcmpl-b2a22839-7cc5-9f15-9785-f626db0d64a3', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--d74e1d17-9b8b-49eb-9c05-5a0f59564450-0', usage_metadata={'input_tokens': 604, 'output_tokens': 168, 'total_tokens': 772, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}), HumanMessage(content='我喜欢地理还是历史？', additional_kwargs={}, response_metadata={}, id='8516b252-a5bd-43b1-ab85-8d32e46dd8e4'), AIMessage(content='你猜怎么着？😊 这个问题让我忍不住笑了一下——因为在我们之前的对话里，你**明确告诉我你很喜欢历史，尤其是中国历史**。我还热情地回应了一大段关于中国历史各个朝代和有趣话题的建议呢！\n\n所以，答案其实很明显：你最初告诉我的是 **历史** ❤️\n\n当然啦，如果你现在是想“改赛道”开始聊地理，我也没问题！地理和历史本来就是一对好搭档，尤其在中国历史的研究中，地理更是关键——\n\n比如：\n📍为什么黄河流域成为中华文明的摇篮？\n⛰️ 山脉如何影响古代军事布防（比如秦岭对蜀地的屏障作用）\n🌊 河流与王朝命运：黄河泛滥是否加速了某些朝代的衰落？\n🏰 长城为何不是一条直线，而是一道“地理+政治”的曲线？\n\n如果你是想测试我有没有认真记住你之前说的话，那我要给自己加个“历史记忆buff”啦~ 😄\n\n不过现在问题来了：你今天更想聊历史，还是地理？我随时切换频道，陪你深入探索！🌍📚', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 240, 'prompt_tokens': 787, 'total_tokens': 1027, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}}, 'model_name': 'qwen-plus', 'system_fingerprint': None, 'id': 'chatcmpl-c19dc444-5a8e-9289-ae27-86a220da7fc2', 'service_tier': None, 'finish_reason': 'stop', 'logprobs': None}, id='run--8f2f3c11-2663-410f-981c-3011fc462725-0', usage_metadata={'input_tokens': 787, 'output_tokens': 240, 'total_tokens': 1027, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}})]}, next=(), config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f0889cc-ac6c-6548-800a-1314be114b24'}}, metadata={'source': 'loop', 'step': 10, 'parents': {}}, created_at='2025-09-03T08:05:44.678519+00:00', parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f0889cc-62b8-60a6-8009-72b1d60c1f10'}}, tasks=(), interrupts=())
可以看到所有交互轮次的数据，其中的next标识工作流中要执行的下一个节点，因为交互已经结束，进入了END节点，所以next为空。