langgraph中state，config，context，短期记忆，长期记忆
state本质就是一个存数据的容器类似redis的短期存储，其中的核心机制就是reducer，reducer就是负责把 State 中同时出现（或者先后出现）的新旧值，进行合理的拼接和处理。
Config是智能体的存储单元：
Config 是一个标准的 Python 字典，主要用于在运行时（Runtime）向图传递一些元数据。它最核心的作用体现在 configurable 这个字段上。
隔离短期记忆（Thread ID）：在您的代码中，config = {"configurable": {"thread_id": "1"}}。这个 thread_id 就像是给对话分配的一个“独立房间号”。LangGraph 的 Checkpointer（短期记忆）会根据这个房间号，把当前对话的 State 存进去。只要 thread_id 一样，AI 就能在同一个房间里接着上次的上下文聊。
传递全局配置：除了 thread_id，您还可以用它来传递当前使用的模型名称、数据库连接参数等。
大白话：Config 就像是“系统后台的调度指令”。它不直接参与大模型的对话，但它决定了对话的走向、记忆的隔离和底层资源的调用。
Context是一个上下文对象：
Context 是 LangGraph 在较新版本中引入的一个更高级、更结构化的概念：
强类型的业务数据：Config 只是一个普通的字典，而 Context 是一个强类型的数据类（Dataclass）。您可以把用户的真实身份（如 user_id）、权限、当前所在部门等业务属性封装在里面。
注入到节点中：在您的节点函数 def call_model(state: MessagesState, runtime: Runtime[Context]) 中，LangGraph 会自动把 Context 包装进 runtime 里。节点代码就可以通过 runtime.context.user_id 非常优雅、安全地获取到当前用户的 ID。
配合长期记忆（Store）：拿到 user_id 后，AI 就可以去 Store（长期记忆仓库）里，精准地检索属于这个特定用户的记忆（namespace = ("memories", user_id)）。
大白话：Context 就像是“用户的数字身份证”。它告诉 AI：“现在跟你聊天的这个人是谁，他有什么权限，他属于哪个部门”。
1. 短期记忆（Short-Term Memory） 短期记忆的核心是“当前会话的上下文连贯性”。
它让智能体能够记住“我们刚才聊了什么”，从而支持多轮对话。 作用范围：严格限制在单个对话线程（Thread）内。不同线程之间的短期记忆是完全隔离的，互不干扰。
实现机制：通过 Checkpointer（检查点机制） 实现。LangGraph 会在图的每个节点执行后，自动保存当前状态的快照（Checkpoint）。
在调用图时，通过传入特定的 thread_id，框架会自动加载该线程的历史状态。 存储介质：开发测试时可使用内存存储（MemorySaver），生产环境通常使用持久化数据库（如 PostgreSQL、SQLite 或 MongoDB）来保证重启后记忆不丢失。 典型场景：聊天机器人、多轮问答、单次复杂任务的步骤跟踪。
2. 长期记忆（Long-Term Memory） 长期记忆的核心是“跨会话的知识与经验沉淀”。
它让智能体能够记住“你是谁”、“你的偏好是什么”，从而提供个性化的服务。 作用范围：跨线程、跨会话共享。无论用户开启多少个新的对话窗口，智能体都能随时调取这些信息。
实现机制：通过 Memory Store（记忆存储机制） 实现。与 Checkpointer 的“自动保存”不同，长期记忆需要智能体在图的节点中显式地执行读取（Search/Get）和写入（Put/Add）操作。它通常以命名空间（Namespace）和键值对（Key-Value）来组织数据，支持语义检索和过滤查询。
存储介质：本质上是一个文档数据库或向量数据库。开发时可使用 InMemoryStore，生产环境推荐接入 PostgreSQL 或 MongoDB 等外部存储。 典型场景：用户画像与偏好记录（如“用户不吃辣”）、历史交互摘要、跨任务的知识库积累。
builder.add_node("agent", create_react_agent(model, tools=[update_memory]))，就是create_react_agent(model, tools=[update_memory]构建一个带有工具update_memory
短期记忆装载：
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os
load_dotenv()
# 1 连接大模型
model = ChatOpenAI(
model=os.getenv("MODEL_NAME"),
api_key=os.getenv("API_KEY"),
base_url=os.getenv("BASE_URL"),
)
# 2 定义一个 InMemorySaver --》短期记忆的对象
checkpointer = InMemorySaver()
def call_model(state: MessagesState):
response = model.invoke(state["messages"]) # state["messages"] 所有的消息
return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
# 3 编译图时，把短期记忆对象传入，会自动记录之前的对话
graph = builder.compile(checkpointer=checkpointer)
# graph = builder.compile()
# 4 必须在同一个 线程下： "thread_id": "1"
config = {
"configurable": {
"thread_id": "1"
}
}
for chunk in graph.stream(
{"messages": [{"role": "user", "content": "你好，我是lqz"}]},
config,
stream_mode="values"
):
chunk["messages"][-1].pretty_print()
print('---------------')
for chunk in graph.stream(
{"messages": [{"role": "user", "content": "我的名字是什么?"}]},
config,
stream_mode="values"
):
chunk["messages"][-1].pretty_print()讲解一下.
1.核心组件：短期记忆的对象：
# 2 定义一个 InMemorySaver --》短期记忆的对象
checkpointer = InMemorySaver()2.将对象绑定在图上：
# 3 编译图时，把短期记忆对象传入，会自动记录之前的对话
graph = builder.compile(checkpointer=checkpointer)3.短期记忆的身份证号：
# 4 必须在同一个 线程下： "thread_id": "1"
config = {
"configurable": {
"thread_id": "1"
}
}管理短期记忆：
# 1 长期记忆只要磁盘够大，可以一直存
# 2 短期记忆存多了，会出现，token太大，超过了 【LLM的上下文窗口】：最大能接受的token数
-短期记忆所有会话，都会一次性提交给 大模型
# 3 解决：
修剪消息：删除前 N 个或后 N 个消息（在调用 LLM 之前）
删除消息 从 LangGraph 状态永久删除
总结消息：总结历史记录中的早期消息，并用摘要替换它们【我们喜欢用】
# 通过检查点checkpoint获取之前的所有对话--》自己的规则处理
管理检查点 以存储和检索消息历史记录
用redis存短期记忆：
from langgraph.graph import StateGraph, MessagesState, START
#pip install langgraph-checkpoint-redis
from langgraph.checkpoint.redis import RedisSaver
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
import os
# 1 连接大模型
model = ChatOpenAI(
model=os.getenv("MODEL_NAME"),
api_key=os.getenv("API_KEY"),
base_url=os.getenv("BASE_URL"),
)
# 2 创建 RedisSaver--使用上下文管理器
DB_URI = "redis://127.0.0.1:6380" # 使用docker跑一个
with RedisSaver.from_conn_string(DB_URI) as checkpointer:
# with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
checkpointer.setup() # 在第一次使用 Redis 检查点时调用 checkpointer.setup()，多次执行也没问题
# 3 节点
def call_model(state: MessagesState):
response = model.invoke(state["messages"])
return {"messages": response}
# 3 构建图
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
# 3.1 传入短期记忆---》放在redis中
graph = builder.compile(checkpointer=checkpointer)
config = {
"configurable": {
"thread_id": "1"
}
}
# for chunk in graph.stream(
# {"messages": [{"role": "user", "content": "你好，我是lqz"}]},
# config,
# stream_mode="values"
# ):
# chunk["messages"][-1].pretty_print()
for chunk in graph.stream(
{"messages": [{"role": "user", "content": "我是谁?"}]},
config,
stream_mode="values"
):
chunk["messages"][-1].pretty_print()使用 RedisSaver 后，LangGraph 存入 Redis 的并不是单纯的聊天记录文本，而是整个图的“状态快照（Checkpoint）”。
Redis 里具体存了：
完整的对话历史（Messages）：包括用户的提问、AI 的回复、工具调用的请求和结果。
图的执行状态（State）：当前图走到了哪一步、中间变量是什么。
元数据（Metadata）：比如创建时间、版本号等，用于支持历史回溯。
mysql存长期记忆：
from dataclasses import dataclass
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.mysql.pymysql import PyMySQLSaver
from langgraph.store.mysql import PyMySQLStore
from langgraph.runtime import Runtime
import uuid
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os
load_dotenv()
model = ChatOpenAI(
model=os.getenv("MODEL_NAME"),
api_key=os.getenv("API_KEY"),
base_url=os.getenv("BASE_URL"),
)
@dataclass
class Context:
user_id: str
def call_model(
state: MessagesState,
runtime: Runtime[Context],
):
user_id = runtime.context.user_id
namespace = ("memories", user_id)
memories = runtime.store.search(namespace, query=str(state["messages"][-1].content))
info = "\n".join([d.value["data"] for d in memories])
system_msg = f"你是一个有用的用户交互助手. 用户信息是: {info}"
# Store new memories if the user asks the model to remember
last_message = state["messages"][-1]
if "记住" in last_message.content.lower():
memory = "用户名字是 lqz"
runtime.store.put(namespace, str(uuid.uuid4()), {"data": memory})
response = model.invoke(
[{"role": "system", "content": system_msg}] + state["messages"]
)
return {"messages": response}
DB_URI = f"mysql://root:lqz123@127.0.0.1:3306/langgraph01"
with (
PyMySQLStore.from_conn_string(DB_URI) as store,
PyMySQLSaver.from_conn_string(DB_URI) as checkpointer,
):
store.setup()
checkpointer.setup()
builder = StateGraph(MessagesState, context_schema=Context)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(
checkpointer=checkpointer,
store=store,
)
config = {"configurable": {"thread_id": "1"}}
for chunk in graph.stream(
{"messages": [{"role": "user", "content": "你好，请记住，我的名字叫：lqz"}]},
config,
stream_mode="values",
context=Context(user_id="1"),
):
chunk["messages"][-1].pretty_print()
config = {"configurable": {"thread_id": "2"}}
for chunk in graph.stream(
{"messages": [{"role": "user", "content": "我的名字是什么?"}]},
config,
stream_mode="values",
context=Context(user_id="1"),
):
chunk["messages"][-1].pretty_print()构建记忆仓库：
from langgraph.store.mysql import PyMySQLStore
# ...
with PyMySQLStore.from_conn_string(DB_URI) as store:
# ...
graph = builder.compile(checkpointer=checkpointer, store=store)这里引入了一个独立于对话状态（State）之外的存储系统——PyMySQLStore
模拟了记忆提取过程：
namespace = ("memories", user_id)
last_message = state["messages"][-1]
if "记住" in last_message.content.lower():
memory = "用户名字是 lqz"
# 核心写入动作
runtime.store.put(namespace, str(uuid.uuid4()), {"data": memory})当检测到用户指令包含“记住”时触发:
命名空间隔离：namespace = ("memories", user_id)。这是一个非常关键的设计。它确保了用户 A 的记忆不会泄露给用户 B。所有属于 user_id="1" 的记忆都被归类在 ("memories", "1") 这个文件夹下。
持久化存储：runtime.store.put(...) 会直接将数据写入 MySQL。此时，即便程序重启，这条“用户名字是 lqz”的信息依然存在。
# 1. 确定去哪里找（Namespace）
namespace = ("memories", user_id)
# 2. 执行语义搜索（Search）
memories = runtime.store.search(
namespace,
query=str(state["messages"][-1].content)
)
# 3. 格式化信息
info = "\n".join([d.value["data"] for d in memories])search会调用数据库进行匹配，匹配时还会参照namespace来进行数据的隔离。
总结消息的代码如下：
# pip install -U langmem
from typing import Any, TypedDict
from langchain.messages import AnyMessage
from langchain_core.messages.utils import count_tokens_approximately
from langgraph.graph import StateGraph, START, MessagesState
from langgraph.checkpoint.memory import InMemorySaver
from langmem.short_term import SummarizationNode, RunningSummary # langchain中我们自己调模型总结的，现在使用它，自动总结
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os
load_dotenv()
model = ChatOpenAI(
model=os.getenv("MODEL_NAME"),
api_key=os.getenv("API_KEY"),
base_url=os.getenv("BASE_URL"),
)
# 增强llm--》现在token数最大是128--》用它做总结--》返回的token不能太大，如果大，违背总结的初衷了
summarization_model = model.bind(max_tokens=128)
class State(MessagesState):
context: dict[str, RunningSummary]
class LLMInputState(TypedDict):
summarized_messages: list[AnyMessage]
context: dict[str, RunningSummary]
# 1 导入的类使用
summarization_node = SummarizationNode(
token_counter=count_tokens_approximately, # 近似计算token值
model=summarization_model,
max_tokens=256,
max_tokens_before_summary=256,
max_summary_tokens=128,
)
def call_model(state: LLMInputState):
response = model.invoke(state["summarized_messages"])
return {"messages": [response]}
checkpointer = InMemorySaver()
builder = StateGraph(State)
builder.add_node(call_model)
builder.add_node("summarize", summarization_node)
builder.add_edge(START, "summarize")
builder.add_edge("summarize", "call_model")
graph = builder.compile(checkpointer=checkpointer)
# Invoke the graph
config = {"configurable": {"thread_id": "1"}}
graph.invoke({"messages": "你好，我的名字是lqz"}, config)
graph.invoke({"messages": "写一首关于猫的短诗"}, config)
graph.invoke({"messages": "现在做同样的事，但是是关于狗的"}, config)
final_response = graph.invoke({"messages": "我的名字是什么?"}, config)
final_response["messages"][-1].pretty_print()
print("\nSummary:", final_response["context"]["running_summary"].summary)
智能硬件社区聚焦AI智能硬件技术生态，汇聚嵌入式AI、物联网硬件开发者，打造交流分享平台，同步全国赛事资讯、开展 OPC 核心人才招募，助力技术落地与开发者成长。
更多推荐
1
0- 0
已为社区贡献1条内容
所有评论(0)