langgraph-长短期记忆

作者： /
内容纲要

一、没有记忆能力的图结构（大模型没有记忆）

1、基础实现

下面的代码中实现的是一个最简单、最基础的基于 LangGraph 单轮对话图

# 1、配置模型
from dotenv import load_dotenv
load_dotenv()
import os
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key,base_url=base_url,temperature=0,)

import os
from langchain_openai import ChatOpenAI
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 定义状态模式
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 定义对话节点
def call_model(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": response}

# 构建状态图
builder = StateGraph(State)

# 向图中添加节点
builder.add_node("call_model", call_model)

# 构建边
builder.add_edge(START, "call_model")
builder.add_edge("call_model", END)

# 编译图
simple_graph = builder.compile()

# 2、验证
import asyncio
async def main():
    input_data = {"messages": ["你好，我是山姆奥特曼"]}
    async for chunk in simple_graph.astream(input_data, stream_mode="values"):
        if "messages" in chunk:
            last_msg = chunk["messages"][-1]
            print(last_msg.content, end="", flush=True) 

async def main2():
    input_data = {"messages": ["你好，请问我是谁"]}
    async for chunk in simple_graph.astream(input_data, stream_mode="values"):
        if "messages" in chunk:
            last_msg = chunk["messages"][-1]
            print(last_msg.content, end="", flush=True) 

if __name__ == "__main__":   

    try:
        # 生成 Mermaid PNG 数据
        image_data = simple_graph.get_graph().draw_mermaid_png()

        # 定义保存路径
        file_path = "langgraph_structure.png"

        # 写入文件
        with open(file_path, "wb") as f:
            f.write(image_data)

        print(f"✅ 图片已成功生成: {os.path.abspath(file_path)}")

        # 自动打开图片
        # 使用 start 命令调用 Windows 默认图片查看器
        # os.startfile(os.path.abspath(file_path))

    except Exception as e:
        print(f"❌ 生成失败: {e}")

    asyncio.run(main())
    print("\n------------------")
    asyncio.run(main2())

    """
    ✅ 图片已成功生成: D:\projects\code\class-study\langgraph_structure.png
    你好，我是山姆奥特曼
    你好！很高兴认识你。不过我需要说明的是，我无法验证用户的真实身份，也不记得之前与任何人的对话记录。
    如果你有关于编程、人工智能或其他方面的问题，我很乐意帮助你。你可以直接告诉我你需要什么帮助。
    ------------------
    你好，请问我是谁
    你好！很高兴见到你。
    不过很抱歉，我并没有关于你身份的信息。作为AI助手，我没有记忆功能，也不会保存任何用户信息。每次对话对我来说都是全新的开始。
    如果你愿意的话，可以告诉我你的名字或其他你想分享的信息，这样我就能在这次对话中更好地称呼你！有什么我可以帮助你的吗？
    """

二、实现记忆功能（把对话记录传递过去）

从上述两轮提问和响应来看，虽然每轮的问题都是输入到同一个编译后的图实例，但是每次表现得都像一个“新图”一样，完全不知道自己之前做了什么。所以很显然，当前的图结构是缺乏上下文记忆能力的。
首先要明确的是：大模型本身是不具备记忆能力的。 而为大模型赋予记忆能力的方法，通常是通过维护一个消息列表来保存之前的对话内容，然后在每次新的对话时填充到整体的Prompt中，从而实现上下文的持续传递，比如下面的代码：

from dotenv import load_dotenv
load_dotenv()
import os
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from openai import OpenAI
client = OpenAI(api_key=key,base_url=base_url)

completion = client.chat.completions.create(
model="gpt-4o",
messages=[
    {"role": "user", "content": "你好，我叫马斯克"},
    {"role": "assistant", "content": "你好，马斯克，很高兴认识你。"},
    {"role": "user", "content": "你知道我叫什么吗？"}
]
)

print(completion.choices[0].message.content)

"""
你刚才告诉我，你叫马斯克。对吗？
"""

三、短期记忆

LangGraph Memory Docs:
https://docs.langchain.com/oss/python/concepts/memory
该地址是 LangGraph 官方文档中 记忆（Memory） 模块的专属页面，专门讲解对话记忆、状态持久化、会话存储相关能力。

核心覆盖内容：
基础会话记忆保存机制
不同存储后端（内存 / 数据库）持久化方案
多轮对话状态隔离、会话 ID 管理
记忆截断、上下文窗口控制、自动摘要等优化手段

file

从下面的代码可以看出，每一个消息都有一个对应的消息id，每一个任务都有一个任务id。如果把这些消息id，都管理起来，加到一个列表里。就可以实现记忆功能。

0、代码示例

# 1、配置模型
from dotenv import load_dotenv
load_dotenv()
import os
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key,base_url=base_url,temperature=0,)
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
from langgraph.graph.message import add_messages

# 定义状态模式
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 定义大模型交互节点
def call_model(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": response}

# 定义翻译节点
def translate_message(state: State):
    system_prompt = """
Please translate the received text in any language into English as output
"""
    messages = state['messages'][-1]
    messages = [SystemMessage(content=system_prompt)] + [HumanMessage(content=messages.content)]
    response = llm.invoke(messages)
    return {"messages": response}

# 构建状态图
builder = StateGraph(State)

# 向图中添加节点
builder.add_node("call_model", call_model)
builder.add_node("translate_message", translate_message)

# 构建边
builder.add_edge(START, "call_model")
builder.add_edge("call_model", "translate_message")
builder.add_edge("translate_message", END)

# 编译图
simple_short_graph = builder.compile()

# 生成可视化图像结构
try:
    # 生成 Mermaid PNG 数据
    image_data = simple_short_graph.get_graph().draw_mermaid_png()
    # 定义保存路径
    file_path = "check-point.png"
    # 写入文件
    with open(file_path, "wb") as f:
        f.write(image_data)
    print(f"✅ 图片已成功生成: {os.path.abspath(file_path)}")
    # 自动打开图片
    # 使用 start 命令调用 Windows 默认图片查看器
    # os.startfile(os.path.abspath(file_path))
except Exception as e:
    print(f"❌ 生成失败: {e}")

async def debug1():
    async for chunk in simple_short_graph.astream({"messages": ["你好，我叫sam"]}, stream_mode="debug"):
        # print(chunk)
        print(f"Task id : {chunk['payload']['id']}")
        if chunk["type"] == "task":
            for message in chunk["payload"]["input"]["messages"]:
                print(f"Message id:{message.id}, Message content:{message.content}")
        if chunk["type"] == "task_result":
            result_message = chunk['payload']['result']['messages']
            print(f"Message id:{result_message.id}, Message content:{result_message.content}")
        print("--------------------------------------------------")
        print("--------------------------------------------------")

if __name__ == "__main__":
    import asyncio
    asyncio.run(debug1());

"""
print(chunk)
✅ 图片已成功生成: D:\projects\code\class-study\check-point.png
{'step': 1, 'timestamp': '2026-07-02T00:48:30.859531+00:00', 'type': 'task', 'payload': {'id': 'bafc5f1e-0a58-2434-5de8-50bddeed4c49', 'name': 'call_model', 'input': {'messages': [HumanMessage(content='你好，我叫sam', additional_kwargs={}, response_metadata={}, id='1aba26b6-2b64-4851-92a8-342045d75f6d')]}, 'triggers': ('branch:to:call_model',)}}l', 'error': None, 'result': {'messages': AIMessage(content='\n\n你好，sam！很高兴认识你！👋\n\n有什么我可以帮助你的吗？无论是编程问题、技术咨询，还是其他事情，都可以随时问我。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 49, 'prompt_tokens': 42, 'total_tokens': 91, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_provider': 'openai', 'model_name': 'MiniMax-M2.5', 'system_fingerprint': None, 'id': 'chatcmpl-e89316df-3950-4b99-b60a-cf9850cb91de', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--019f204c-7a8d-77d0-a281-216d42597c83-0', tool_calls=[], invalid_tool_calls=[], usage_metadata={'input_tokens': 42, 'output_tokens': 49, 'total_tokens': 91, 'input_token_details': {}, 'output_token_details': {}})}, 'interrupts': []}}
{'step': 2, 'timestamp': '2026-07-02T00:48:32.087377+00:00', 'type': 'task', 'payload': {'id': '1b1fcb2b-2fee-dfe3-60b1-47f7d0fb2469', 'name': 'translate_message', 'input': {'messages': [HumanMessage(content='你好，我叫sam', additional_kwargs={}, response_metadata={}, id='1aba26b6-2b64-4851-92a8-342045d75f6d'), AIMessage(content='\n\n你好，sam！很高兴认识你！👋\n\n有什么我可以帮助你的吗？无论是编程问题、技术咨询，还是其他事情，都可以随时问我。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 49, 'prompt_tokens': 42, 'total_tokens': 91, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_provider': 'openai', 'model_name': 'MiniMax-M2.5', 'system_fingerprint': None, 'id': 'chatcmpl-e89316df-3950-4b99-b60a-cf9850cb91de', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--019f204c-7a8d-77d0-a281-216d42597c83-0', tool_calls=[], invalid_tool_calls=[], usage_metadata={'input_tokens': 42, 'output_tokens': 49, 'total_tokens': 91, 'input_token_details': {}, 'output_token_details': {}})]}, 'triggers': ('branch:to:translate_message',)}}
{'step': 2, 'timestamp': '2026-07-02T00:48:36.482268+00:00', 'type': 'task_result', 'payload': {'id': '1b1fcb2b-2fee-dfe3-60b1-47f7d0fb2469', 'name': 'translate_message', 'error': None, 'result': {'messages': AIMessage(content="\n\nHello, sam! Nice to meet you! 👋\n\nIs there anything I can help you with? Whether it's programming questions, technical consultation, or other things, feel free to ask me anytime.", additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 224, 'prompt_tokens': 58, 'total_tokens': 282, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_provider': 'openai', 'model_name': 'MiniMax-M2.5', 'system_fingerprint': None, 'id': 'chatcmpl-062fb3e8-7196-4ebd-9d3d-7656d2c46a07', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--019f204c-7f5a-7821-b043-2fb8af5884b9-0', tool_calls=[], invalid_tool_calls=[], usage_metadata={'input_tokens': 58, 'output_tokens': 224, 'total_tokens': 282, 'input_token_details': {}, 'output_token_details': {}})}, 'interrupts': []}}

✅ 图片已成功生成: D:\projects\code\class-study\check-point.png
Task id : 8cb09609-b65a-6a8b-e9fc-83dce8a27d55
Message id:004985e7-119f-4381-9dfb-ddef045f5404, Message content:你好，我叫sam
--------------------------------------------------
--------------------------------------------------
Task id : 8cb09609-b65a-6a8b-e9fc-83dce8a27d55
Message id:lc_run--019f2093-9319-7ab0-ad29-689c33ea0a48-0, Message content:
你好，sam！很高兴认识你！👋
有什么我可以帮助你的吗？无论是编程问题、技术咨询，还是其他事情，都可以随时问我。
--------------------------------------------------
--------------------------------------------------
Task id : c0fb21c7-f5ec-14e7-c194-2f055d10f3bf
Message id:004985e7-119f-4381-9dfb-ddef045f5404, Message content:你好，我叫sam
Message id:lc_run--019f2093-9319-7ab0-ad29-689c33ea0a48-0, Message content:
你好，sam！很高兴认识你！👋
有什么我可以帮助你的吗？无论是编程问题、技术咨询，还是其他事情，都可以随时问我。
--------------------------------------------------
--------------------------------------------------
Task id : c0fb21c7-f5ec-14e7-c194-2f055d10f3bf
Message id:lc_run--019f2093-9788-7fb1-a195-537f8208a2b8-0, Message content:
Hello, sam! Nice to meet you! 👋
Is there anything I can help you with? Whether it's programming problems, technical consultation, or other things, feel free to ask me anytime.
--------------------------------------------------
--------------------------------------------------
"""
sessions = {
    "会话1": [
        {
            "id": "message_1",
            "content": "你好！",
            "timestamp": "2025-10-30T09:00:00"
        },
        {
            "id": "message_2",
            "content": "请问有什么可以帮助您的？",
            "timestamp": "2025-10-30T09:05:00"
        },
        {
            "id": "message_3",
            "content": "感谢，再见！",
            "timestamp": "2025-10-30T09:10:00"
        }
    ],
    "会话2": [
        {
            "id": "message_5",
            "content": "怎么了解更多产品信息？",
            "timestamp": "2025-10-30T10:00:00"
        },
        {
            "id": "message_7",
            "content": "请查看我们的产品目录。",
            "timestamp": "2025-10-30T10:05:00"
        },
        {
        "id": "message_3",
        "content": "感谢，再见！",
        "timestamp": "2025-10-30T09:10:00"
        }
    ]
}

1、短期记忆的类型（MemorySaver、SqliteSaver、PostgresSaver）

理解到这里，现在我们就可以说：LangGraph 框架中的 checkpointer 做的就是这样的事。具体来说，它就是通过一些数据结构来存储 State 状态中产生的信息，并且在每个 task 开始时去读取全局的状态。主要通过以下四种方式来实现：

MemorySaver：用于实验性质的记忆检查点。（因为内存容易丢失）
SqliteSaver / AsyncSqliteSaver：使用 SQLite 数据库实现的记忆检查点，适合实验性质和本地工作流程。
PostgresSaver / AsyncPostgresSaver：使用 Postgres 数据库实现的高级检查点，适合在生产系统中使用。
支持自定义检查点。
不同类型的 checkpointer 以不同的形式去管理 State 状态中记录的中间状态信息。但这还不够。为了将一系列产生的消息归属到正确的类别中，就像上面的 会话 1 包含一系列的问答，而 会话 2 包含另一批系列的回答，LangGraph 框架引入 Thread（线程）概念来充当 会话 的角色。每个线程代表一个独特的交互或对话流。而 thread_id 是与特定执行线程关联的唯一标识符各个概念之间的关联如下图所示：

file

checkpointer 是 memory 的一种特定实现，它在执行期间保存图在各个点的状态，使系统能够在中断时从该点恢复。这与 LangGraph 中状态的一般概念不同，后者表示应用程序在任何给定时刻的当前快照。虽然状态是动态的并且随着图形的执行而变化，但 checkpointer 提供了一种存储和检索历史状态的方法，从而促进更复杂的工作流程和人机交互。
接下来，将以 MemorySaver 这个实现 checkpointer 的方法为例，去理解这个过程。

2、 检查点的特定实现类型 – MemorySaver

LangGraph 框架有一个内置的持久层，通过 checkpointer 实现。当使用 checkpointer 编译图时，检查点会在每个super step中保存图状态的 checkpoint（可以理解每次 State 更新，checkpoint 就打一次快照）。这些 checkpoint 被保存到一个 thread 中，可以在图执行后访问。如下图所示：
super step可以被认为是图节点上的单次迭代 (前面我们看到的一个 [Task,TaskResult], 就是一个super step)。并行运行的节点是同一super step的一部分，而顺序运行的节点则属于单独的super step。在图执行开始时，所有节点都开始处于 inactive 状态。当节点在其任何传入边缘 (或一个 “通道”) 上接收到新消息 (状态) 时，该节点将变为 active。然后，活动节点运行其功能并以更新进行响应。在每个超级步骤结束时，没有传入消息的节点通过将自己标记为 inactive。当所有节点 inactive 并且没有消息在传输时，图执行终止。

file

使用的方法非常简单，就是在创建任何 LangGraph 图时，通过在编译图时添加 MemorySaver 来将其设置为保留其 State 状态中的数据，即：

from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
graph.compile(checkpointer=checkpointer)

通过一个图来理解这个中间过程，构建如下图结构。注意：在编译图的时候，添加 MemorySaver 作为 checkpointer 提供 Memory 功能。

# 1、配置模型
from dotenv import load_dotenv
load_dotenv()
import os
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key,base_url=base_url,temperature=0,)

# 导入检查点
from langgraph.checkpoint.memory import MemorySaver

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

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

def translate_message(state: State):
    system_prompt = """
    Please translate the received text in any language into English as output
    """
    messages = state['messages'][-1]
    messages = [SystemMessage(content=system_prompt)] + [HumanMessage(content=messages.content)]
    response = llm.invoke(messages)
    return {"messages": response}

builder = StateGraph(State)

builder.add_node("call_model", call_model)
builder.add_node("translate_message", translate_message)

builder.add_edge(START, "call_model")
builder.add_edge("call_model", "translate_message")
builder.add_edge("translate_message", END)

memory = MemorySaver()
graph_with_memory = builder.compile(checkpointer=memory)  # 在编译图的时候添加检查点

# 2、生成流程图
try:
    # 生成 Mermaid PNG 数据
    image_data = graph_with_memory.get_graph().draw_mermaid_png()
    # 定义保存路径
    file_path = "memory-saver.png"
    # 写入文件
    with open(file_path, "wb") as f:
        f.write(image_data)
    print(f"✅ 图片已成功生成: {os.path.abspath(file_path)}")
    # 自动打开图片
    # 使用 start 命令调用 Windows 默认图片查看器
    # os.startfile(os.path.abspath(file_path))
except Exception as e:
    print(f"❌ 生成失败: {e}")

# 当添加了 checkpointer 后，在该图执行的每个超级步骤中会自动创建检查点。即每个节点处理其输入并更新状态后，会将当前状态保存为检查点。但如果像普通图一样，仅传入输入的问题是会报错的，如下所示：
async def main0():
    async for chunk in graph_with_memory.astream(input = {"messages": ["你好，我叫齐天大圣"]},  stream_mode="values"):
        if "messages" in chunk:
            last_msg = chunk["messages"][-1]
            print(last_msg.content, end="", flush=True) 

"""
ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id
"""

"""
这是因为当增加了 checkpointer 后，需要 Thread 来作为 checkpointer 保存图中每个检查点的唯一标识，
而 Thread（线程）又是通过 thread_id 来指定执行线程，所以在使用 checkpointer 调用图时，
必须指定 thread_id，指定的方式是作为配置量 configurable 的一部分进行声明，正确调用示例如下所示：
"""

# 这个 thread_id 可以取任意数值
config = {"configurable": {"thread_id": "1"}}
async def main1():
    async for chunk in graph_with_memory.astream(input = {"messages": ["你好，我叫齐天大圣"]}, config = config  , stream_mode="values"):
        if "messages" in chunk:
            last_msg = chunk["messages"][-1]
            print(last_msg.content, end="", flush=True) 

async def main2():
    async for chunk in graph_with_memory.astream(input = {"messages": ["你好，请问我是谁"]}, config = config, stream_mode="values"):
        if "messages" in chunk:
            last_msg = chunk["messages"][-1]
            print(last_msg.content, end="", flush=True) 

async def main3():
    async for chunk in graph_with_memory.astream(input = {"messages": ["我刚才都问了你什么问题？"]}, config = config, stream_mode="values"):
        if "messages" in chunk:
            last_msg = chunk["messages"][-1]
            print(last_msg.content, end="", flush=True) 

if __name__ == "__main__":
    import asyncio
    # asyncio.run(main0())
    asyncio.run(main1())
    asyncio.run(main2())
    asyncio.run(main3())

3、 检查点的特定实现类型 – SqlitSaver

SqliteSaver 是 checkpointer 的第二种实现形式，不同于 MemorySaver 仅通过字典的形式将状态信息存储在当前的运行环境下，SqliteSaver 的底层持久存储会把 checkpoint 实际的存储在本地的 SQLite 数据库中，同时提供了异步环境下的实现 AsyncSqliteSaver，适用于量级级的应用场景。

SqliteSaver 有两种存储形式，一种是类似于 MemorySaver 将 checkpointer 存储在内存中，另外一种是存储在 sqlite 数据库中。首先来看第一种：
内存存储（in-Memory Storage）是指存储在计算机主存储器 (RAM) 中的数据，这种类型的存储允许非常快速地访问和检索数据，因为它不涉及磁盘 I/O 操作。这个过程是将 checkpointer 最初保存到内存中，在需要时从内存中进行检索。内部完整的实现思路如下：

3.1 安装依赖库

需要单独安装 langgraph-checkpoint-sqlite 库。

uv add langgraph-checkpoint-sqlite
3.2 定义内存的存储形式

通过 :memory: 方法指定在内存中存储 checkpointer。代码如下：

from langgraph.checkpoint.sqlite import SqliteSaver
# 创建一个内存中的检查点
memory = SqliteSaver.from_conn_string(":memory:")
3.3. 构建 checkpointer

这里为了演示 SqliteSaver 的执行原理，我们手动构建一个测试的 checkpointer，其默认实现的是从 State 中进行提取。

checkpoint_data = {
    "thread_id": "新建会话窗口",
    "thread_ts": "2025-10-30T07:23:38.656547+00:00",
    "checkpoint": {
        "id": "1ef968fe-1eb4-6049-bfff",
    },
    "metadata": {"timestamp": "2025-10-30T07:23:38.656547+00:00"}
}
3.4. 存储 checkpointer

在源码中，from_conn_string 方法使用了 Python 的 contextmanager 装饰器，所以它是一个生成器函数。这个方法创建的实例必须在 with 语句中使用。即我们需要修改其构建的方式，并通过 put 方法进行 checkpointer 配置的写入，代码如下：

with SqliteSaver.from_conn_string(":memory:") as memory:
    # 保存检查点，包括时间戳
    saved_config = memory.put(
        config={"configurable": {"thread_id": checkpoint_data["thread_id"], "thread_ts": checkpoint_data["thread_ts"], "checkpoint_ns": ""}},
        checkpoint=checkpoint_data["checkpoint"],
        metadata=checkpoint_data["metadata"],
        new_versions= {"writes": {"key": "value"}}
    )

除此之外，还可以通过 list 方法查看到 thread_id 下所有的检查点信息，代码如下：

with SqliteSaver.from_conn_string(":memory:") as memory:
    # 保存检查点，包括时间戳
    saved_config = memory.put(
        config={"configurable": {"thread_id": checkpoint_data["thread_id"], "thread_ts": checkpoint_data["thread_ts"], "checkpoint_ns": ""}},
        checkpoint=checkpoint_data["checkpoint"],
        metadata=checkpoint_data["metadata"],
        new_versions= {"writes": {"key": "value"}}
    )

    # 检索检查点的数据
    config = {"configurable": {"thread_id": checkpoint_data["thread_id"]}}

    # 获取给定 thread_id 的所有检查点
    checkpoints = list(memory.list(config))
    for checkpoint in checkpoints:
        print(checkpoint)

除此之外，SQLiteSaver 还支持持久化存储。这种方式是指存储在非易失性介质上的数据，例如硬盘驱动器、SSD 或云存储，即使应用程序停止或系统断电，这种类型的存储也会保留数据。使用的方式也非常简单，只需要把 from_conn_string 中的 :memory: 更换为指向本地的 sqlite.db 的文件即可，这允许数据持久保存，便于长期存取。代码如下：


import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

# 创造一个数据 checkpointer
checkpoint_data = {
    "thread_id": "新建会话窗口1",
    "thread_ts": "2025-10-30T07:23:38.656547+00:00",
    "checkpoint": {
        "id": "1ef968fe-1eb4-6049-bfff",
    },
    "metadata": {"timestamp": "2025-10-30T07:23:38.656547+00:00"}
}

with SqliteSaver.from_conn_string("checkpoints20260101.sqlite") as memory:
    # 保存检查点，包括时间戳
    saved_config = memory.put(
        config={"configurable": {"thread_id": checkpoint_data["thread_id"], "thread_ts": checkpoint_data["thread_ts"], "checkpoint_ns": ""}},
        checkpoint=checkpoint_data["checkpoint"],
        metadata=checkpoint_data["metadata"],
        new_versions= {"writes": {"key": "value"}}
    )

    # 检索检查点的数据
    config = {"configurable": {"thread_id": checkpoint_data["thread_id"]}}

    # 获取给定 thread_id 的所有检查点
    checkpoints = list(memory.list(config))
    for checkpoint in checkpoints:
        print(checkpoint)

"""
CheckpointTuple(config={'configurable': {'thread_id': '新建会话窗口1', 'checkpoint_ns': '', 'checkpoint_id': '1ef968fe-1eb4-6049-bfff'}}, checkpoint={'id': '1ef968fe-1eb4-6049-bfff'}, metadata={'timestamp': '2025-10-30T07:23:38.656547+00:00', 'thread_ts': '2025-10-30T07:23:38.656547+00:00'}, parent_config=None, pending_writes=[])
"""

可以使用标准的 SQL 语法直接与数据库进行交互。


import sqlite3
# sql语法交互

# 建立数据库连接
conn = sqlite3.connect("checkpoints20260101.sqlite")

# 创建一个游标对象来执行你的SQL查询
cursor = conn.cursor()

# 查询数据库中所有表的名称
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")

# 获取查询结果
tables = cursor.fetchall()

# 打印所有表名
for table in tables:
    print(f"表名: {table}")

# 从检查点表中检索所有数据
cursor.execute(f"SELECT * FROM checkpoints;")
all_data = cursor.fetchall()

# 打印检查点表中的所有数据
print("Data in the 'checkpoints' table:")
for row in all_data:
    print(f"打印检查点表中的所有数据: \n{row}")

"""
表名: ('checkpoints',)
表名: ('writes',)
Data in the 'checkpoints' table:
打印检查点表中的所有数据: 
('新建会话窗口1', '', '1ef968fe-1eb4-6049-bfff', None, 'msgpack', b'\x81\xa2id\xb71ef968fe-1eb4-6049-bfff', b'{"timestamp": "2025-10-30T07:23:38.656547+00:00", "thread_ts": "2025-10-30T07:23:38.656547+00:00"}')
"""
3.5. 存储 checkpointer调用工具
from langchain_core.tools import tool
from typing import Union, Optional
from pydantic import BaseModel, Field
import requests
import json

from dotenv import load_dotenv
load_dotenv()
import os
weather_key = os.environ["OPEN_WEATHER_KEY"]

class WeatherLoc(BaseModel):
    location: str = Field(description="The location name of the city")

@tool(args_schema=WeatherLoc)
def get_weather(location):
    """
    Function to query current weather.
    :param loc: Required parameter, of type string, representing the specific city name for the weather query. \
    Note that for cities in China, the corresponding English city name should be used. For example, to query the weather for Beijing, \
    the loc parameter should be input as 'Beijing'.
    :return: The result of the OpenWeather API query for current weather, with the specific URL request address being: https://api.openweathermap.org/data/2.5/weather
    The return type is a JSON-formatted object after parsing, represented as a string, containing all important weather information.
    """
    # Step 1. 构建请求
    url = "https://api.openweathermap.org/data/2.5/weather"

    # Step 2. 设置查询参数
    params = {
        "q": location,
        "appid": weather_key,   # 输入API key
        "units": "metric",                           # 使用摄氏度而不是华氏度
        "lang":"zh_cn"                               # 输出语言为简体中文
    }

    # Step 3. 发送GET请求
    response = requests.get(url, params=params)

    # Step 4. 解析响应
    data = response.json()
    return json.dumps(data)

tools = [get_weather]

# 1、配置模型
from dotenv import load_dotenv
load_dotenv()
import os
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key,base_url=base_url,temperature=0,)
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.prebuilt import create_react_agent
# 同步，创建 ReAct 代理的时候添加 memory，代码如下:
def sync_method1():
    # 当前方法需要把所有代码都放在with下面执行，一旦顶格书写代码就失效了。

    # config = {"configurable": {"thread_id": "1"}}
    with SqliteSaver.from_conn_string(":memory:") as checkpointer:
        graph = create_react_agent(llm, tools=tools, checkpointer=checkpointer)
        # 2、生成流程图

        # 生成 Mermaid PNG 数据
        image_data = graph.get_graph().draw_mermaid_png()
        # 定义保存路径
        file_path = "langgraph_structure.png"
        # 写入文件
        with open(file_path, "wb") as f:
            f.write(image_data)
        print(f"✅ 图片已成功生成: {os.path.abspath(file_path)}")

        config = {"configurable": {"thread_id": "1"}}

        for chunk in graph.stream({"messages": ["你好，我叫iron man"]}, config, stream_mode="values"):
            chunk["messages"][-1].pretty_print()

        for chunk in graph.stream({"messages": ["请问我叫什么？"]}, config, stream_mode="values"):
            chunk["messages"][-1].pretty_print()

from contextlib import ExitStack
def sync_method2():
    stack = ExitStack()
    checkpointer = stack.enter_context(SqliteSaver.from_conn_string(":memory:"))
    """
    上述代码实现的机制是:
    1. 创建 ExitStack 实例: stack = ExitStack() 创建了一个可以管理多个上下文管理器的堆栈。
    2. 注册上下文管理器: checkpointer = stack.enter_context(SqliteSaver.from_conn_string(":memory:")) 这一行将
    SqliteSaver.from_conn_string(":memory:") 注册到了 stack 中。SqliteSaver.from_conn_string 返回的上下文管理器对象会被 ExitStack 管
    理, 这意味着只有退出 stack 的上下文 (即调用 stack.close()) , 与之相关的 SqliteSaver 才会自动关闭。
    """
    graph = create_react_agent(llm, tools=tools, checkpointer=checkpointer)

    config = {"configurable": {"thread_id": "2"}}

    for chunk in graph.stream({"messages": ["你好，我叫sam"]}, config, stream_mode="values"):
        chunk["messages"][-1].pretty_print()

    for chunk in graph.stream({"messages": ["请问我叫什么？"]}, config, stream_mode="values"):
        chunk["messages"][-1].pretty_print()

    for chunk in graph.stream({"messages": ["帮我查询一下北京的天气"]}, config, stream_mode="values"):
        chunk["messages"][-1].pretty_print()

    for chunk in graph.stream({"messages": ["请问我刚才都问了哪些问题？"]}, config, stream_mode="values"):
        chunk["messages"][-1].pretty_print()
    stack.close()

# 这是一个更灵活的方法，适用于同步的环境，其中代码可以跨单元传播，并且仍然可以使用上下文信息。异步版本也是类似的实现思路：

import asyncio
from contextlib import AsyncExitStack
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

async def async_method():
    stack = AsyncExitStack()
    memory = await stack.enter_async_context(AsyncSqliteSaver.from_conn_string(":memory:"))

    graph = create_react_agent(llm, tools=tools, checkpointer=memory)

    config = {"configurable": {"thread_id": "33"}}

    async for chunk in graph.astream({"messages": ["帮我查一下北京的天气"]}, config, stream_mode="values"):
        chunk["messages"][-1].pretty_print()

    async for chunk in graph.astream({"messages": ["我刚才问了你什么事情？"]}, config, stream_mode="values"):
        chunk["messages"][-1].pretty_print()

    await stack.aclose()

if __name__ =="__main__":
    # result = get_weather.invoke({"location": "beijing"})
    # print(result)
    # sync_method1()
    # sync_method2()
    asyncio.run(async_method())

总的来说，LangGraph 框架中的 Memory 通过状态管理与检查点与 thread_id 进行绑定，从而隔离不同线程的记忆和状态，防止并发交互之间的干扰，保证每个线程独立运行。总体而言，thread_id 是 LangGraph 中组织和管理记忆的关键机制，可实现有效的状态跟踪和交互连续性。但是， 通过 thread_id 不能实现的是：跨线程的交互。 我们虽然可以通过 thread_id 去中断或者恢复某个对话，但是让不同线程间共享各自的消息，在 checkpointer 的实现机制下并不能做到，所以有了 LangGraph 的长期记忆的实现模块。

四. 长期记忆和 Store（仓库）

仅使用 checkpointer，我们无法做到跨线程共享信息。这激发了对 Store 的需求。LangGraph 通过 BaseStore 接口提供内置文档存储。与通过线程 ID 保存状态的 checkpointer 不同，存储使用自定义命名空间来组织数据。常见用例包括存储用户配置文件、构建知识库以及管理所有线程的全局首选项。具体的实现形式是：LangGraph 将长期记忆作为 JSON 文档存储在 Store 中，每个 memory 都组织在自定义 namespace（类似于文件夹）和不同的 key（例如文件名）下。命名空间通常包含用户或组织 ID 或其他标签，以便更轻松地组织信息。这种结构可以实现存储器的分层组织。然后通过内容过滤器支持跨命名空间搜索。

整体而言，LangGraph 中的长期记忆允许系统保留不同对话或会话中的信息。与线程范围的短期记忆不同，长期记忆保存在自定义"命名空间"中。


"""
Store 的实现源码: https://langchain-ai.github.io/langgraph/reference/store/

具体的实现方法，是我们可以定义一个 InMemoryStore 来跨线程存储有关用户的信息。InMemoryStore 会与 checkpointer 协同工作：
由 checkpointer 将状态保存到线程，而 InMemoryStore 允许我们存储任意信息以供跨线程访问。我们看一下其实现细节:

"""
from langgraph.store.memory import InMemoryStore
in_memory_store = InMemoryStore()

"""
namespace 的类型是 tuple，需要一个键值对。可以理解为: 我们以 user_id=1 这个人的电脑创建了一个 memories 文件夹，所有这个人的数据，都存放在这个文件夹中。
"""
user_id = "1"
namespace_for_memory = (user_id, "memories")

# 接下来使用 .put 方法将 memory 保存到存储中的命名空间中。每个 memory 都有唯一的一个对应的 id .
import uuid

memory_id = str(uuid.uuid4())
memory = {"user" : "你好，我叫jude"}
in_memory_store.put(namespace_for_memory, memory_id, memory)

# 当创建完成后，可以使用 store.search 读取命名空间中的记忆，这将以列表的形式返回给定用户的所有记忆。最近的记忆是列表中的最后一个。

memories = in_memory_store.search(namespace_for_memory)
print(memories[-1].dict())

"""
{'namespace': ['1', 'memories'], 
'key': 'dca93d3c-966e-4a92-a16c-21734fbec200',
'value': {'user': '你好，我叫jude'},
'created_at': '2026-07-06T08:26:26.429277+00:00',
'updated_at': '2026-07-06T08:26:26.429277+00:00',
'score': None}
"""

1、代码示例

import getpass
import os
from langchain_openai import ChatOpenAI
from typing import Annotated
from typing_extensions import TypedDict
from IPython.display import Image, display
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.messages import AnyMessage, SystemMessage, HumanMessage, ToolMessage
from langgraph.graph.message import add_messages
from langchain_core.runnables import RunnableConfig
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langgraph.checkpoint.memory import MemorySaver

in_memory_store = InMemoryStore()
memory = MemorySaver()

from langchain_core.tools import tool
from pydantic import BaseModel, Field
import requests
import json

from dotenv import load_dotenv
load_dotenv()
import os
weather_key = os.environ["OPEN_WEATHER_KEY"]

class WeatherLoc(BaseModel):
    location: str = Field(description="The location name of the city")
@tool(args_schema=WeatherLoc)
def get_weather(location):
    """
    Function to query current weather.
    :param loc: Required parameter, of type string, representing the specific city name for the weather query. \
    Note that for cities in China, the corresponding English city name should be used. For example, to query the weather for Beijing, \
    the loc parameter should be input as 'Beijing'.
    :return: The result of the OpenWeather API query for current weather, with the specific URL request address being: https://api.openweathermap.org/data/2.5/weather
    The return type is a JSON-formatted object after parsing, represented as a string, containing all important weather information.
    """
    # Step 1. 构建请求
    url = "https://api.openweathermap.org/data/2.5/weather"

    # Step 2. 设置查询参数
    params = {
        "q": location,
        "appid": weather_key,   # 输入API key
        "units": "metric",                           # 使用摄氏度而不是华氏度
        "lang":"zh_cn"                               # 输出语言为简体中文
    }

    # Step 3. 发送GET请求
    response = requests.get(url, params=params)

    # Step 4. 解析响应
    data = response.json()
    return json.dumps(data)

tools = [get_weather]

# 1、配置模型
from dotenv import load_dotenv
load_dotenv()
import os
key = os.environ["OPENAI_API_KEY"]
base_url = os.environ["OPENAI_BASE_URL"]
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", api_key=key,base_url=base_url,temperature=0,)
import uuid
# 定义状态模式
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 定义对话节点，访问记忆并在模型调用中使用它们。
def call_model(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    # 获取用户id
    user_id = config["configurable"]["user_id"]

    # 定义命名空间
    namespace = ("memories", user_id)

    # 根据用户id检索记忆
    memories = store.search(namespace)
    info = "\n".join([d.value["data"] for d in memories])

    ## 存储记忆
    last_message = state["messages"][-1]
    store.put(namespace, str(uuid.uuid4()), {"data": last_message.content})

    system_msg = f"Answer the user's question in context: {info}"

    response = llm.invoke(
        [{"type": "system", "content": system_msg}] + state["messages"]
    )

    # 存储记忆
    store.put(namespace, str(uuid.uuid4()), {"data": response.content})
    return {"messages": response}

# 构建状态图
builder = StateGraph(State)
# 向图中添加节点
builder.add_node("call_model", call_model)
# 构建边
builder.add_edge(START, "call_model")
builder.add_edge("call_model", END)
# 编译图
graph = builder.compile(checkpointer=memory, store=in_memory_store)

# 可视化
image_data = graph.get_graph().draw_mermaid_png()
        # 定义保存路径
file_path = "langgraph_structure.png"
# 写入文件
with open(file_path, "wb") as f:
    f.write(image_data)
print(f"✅ 图片已成功生成: {os.path.abspath(file_path)}")

# 输入被验证的身份信息。线程id（模拟同一个会话窗口），用户id（模拟大模型同一个聊天账户）
config = {"configurable": {"thread_id": "10"}, "user_id": "6"}
for chunk in graph.stream({"messages": ["你好，我是sam"]}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

# 情况1：同一个线程id，相同用户id
# 理论结果：可以从thread id找到上下文中得到正确的回答
# 实际结果：符合预期，模型可以识别出用户叫 sam
config = {"configurable": {"thread_id": "10"}, "user_id": "6"}
for chunk in graph.stream({"messages": ["1、你知道我叫什么吗？"]}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
# 我们可以直接去访问 store 查看存储的 memory 信息。   
for memory in in_memory_store.search(("memories", "6")):
    print(f"情况1 同一个线程id，相同用户id 的store空间内容： \n {memory.value}")

# 情况2：不同线程id，相同用户id。
# 理论结果：可以从store id中找到所有的thread id，然后找到上下文中得到正确的回答
# 实际结果：
# store检索直接按 ("memories", user_id) 这个命名空间去查，"user_id相同→共享同一个memory桶"
config = {"configurable": {"thread_id": "100"}, "user_id": "6"}
for chunk in graph.stream({"messages": ["2、你知道我叫什么吗？"]}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
# 我们可以直接去访问 store 查看存储的 memory 信息。
for memory in in_memory_store.search(("memories", "6")):
    print(f"情况2：不同线程id，相同用户id。 的store空间内容： \n {memory.value}")

# 情况3：相同线程id，不同用户id。
# 理论结果：可以从thread id找到上下文中得到正确的回答，这其实是数据泄露风险。
# 实际结果：
# 因为 MemorySaver 只按thread_id存取消息历史，跟user_id完全无关，所以哪怕user_id变了，thread=10的历史消息（含"我是sam"）照样被加载进上下文，模型能看到。
"""
thread_id 才是对话历史的隔离边界，user_id 只隔离 store（长期记忆），二者是两套独立的隔离机制。
如果不同用户共用同一个 thread_id（比如demo里这样测试），即使 user_id 不同，也会看到彼此的对话内容，这在真实多用户系统里是数据泄露风险。
"""
config = {"configurable": {"thread_id": "10"}, "user_id": "60"}
for chunk in graph.stream({"messages": ["3、你知道我叫什么吗？"]}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
# 读取某个 thread_id 当前存储的完整消息历史（checkpointer里的短期记忆）
read_config = {"configurable": {"thread_id": "10"}}
state = graph.get_state(read_config)

print(f"\n thread_id=10 当前存储的消息数量：{len(state.values['messages'])}")
for msg in state.values["messages"]:
    print(f"[{msg.type}] {msg.content}")

# 情况4：不同线程id，不同用户id。
# 理论结果：没有正确回答，不知道我输入的身份
# 实际结果：
# 新thread_id没有消息历史，新user_id在store里也没有任何memory条目（因为从没往("memories","6000")这个命名空间put过东西），所以两边都拿不到线索，模型答不出来
config = {"configurable": {"thread_id": "1000"}, "user_id": "6000"}
for chunk in graph.stream({"messages": ["4、你知道我叫什么吗？"]}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()
# 我们可以直接去访问 store 查看存储的 memory 信息。   
for memory in in_memory_store.search(("memories", "6000")):
    print(f"情况4：不同线程id，不同用户id。 的store空间内容： \n {memory.value}")

#犯过的问题要注意
"""
假设，我的四种情况。做链式设计，在线程id和用户id，刻意设计了两两之间只变动一个维度的链式结构（10→1→1→2，6→6→5→5），形成了一条"thread接力"和"user接力"交替的链条。
类似于：
问题1 ：{"configurable": {"thread_id": "10"}, "user_id": "6"}
问题2 ：{"configurable": {"thread_id": "1"}, "user_id": "6"}
问题3 ：{"configurable": {"thread_id": "1"}, "user_id": "5"}
问题4 ：{"configurable": {"thread_id": "2"}, "user_id": "5"}
如果按照上述这么设置的话，则尽管问题4的线程id和用户id和问题1的完全不同了。但是由于问题2的用户id和问题1的相同，问题2知道了答案”我叫sam“。问题3的线程id=问题2的线程id。所以问题3也知道了。以此类推。问题四也从问题3的store中检索到了答案。这些问题都会把上下文加载一份到自己的内存空间里去是吗

关键澄清:不是"复制到自己的空间",而是"共享同一个桶,持续追加"
checkpointer（按thread_id）：同一个 thread_id 对应的是同一份消息列表，每次调用只是往这份列表里 append，不是每次都拷贝一份新的。
store（按user_id）：同一个 user_id 对应的是同一个 ("memories", user_id) 命名空间桶，每次 store.put 只是往这个桶里加一条新记录，也不是各自拷贝。

"""
问题 thread_id / user_id 依赖哪个id接力 为什么能知道答案
1 thread=10, user=6 — 直接说"我是sam"，建立起 thread10 的消息历史 + user6 的store记忆
2 thread=1(新), user=6(同问题1) 靠user_id接力 thread1是全新的，checkpointer没有历史消息；但 store.search(("memories","6")) 能查到问题1存的记忆，从store拿到答案
3 thread=1(同问题2), user=5(新) 靠thread_id接力 user5是全新的，store里对"5"这个命名空间还是空的；但checkpointer按thread_id=1恢复完整消息列表，这个列表里已经包含问题2那一轮的问答（其中提到了sam），模型直接从对话历史里看到答案
4 thread=2(新), user=5(同问题3) 靠user_id接力 thread2全新，无消息历史；但问题3执行时已经往 ("memories","5") 里 store.put 了新的问答记录（含sam），所以问题4能从store查到

2、一问一答：

这么说长期记忆store其实和短期记忆checkpoint一样，都是创造出一个储存空间，把上下文进行储存，只是store储存的更多，仅此而已？也就是说，市面上的智能体的长期记忆，只是把所有短期记忆中有价值的做了摘录整理，put到store的空间里就是长期记忆了。而短期记忆，为了防止上下文爆炸，也是选择只记录最近的几条，或者把重要的信息，做简要摘录。上下文记忆都是这么处理的？

你的理解方向对了大半，但有两点需要修正——一是这份demo代码本身没有做任何"摘录/整理"，二是真正的长期记忆系统在架构上和短期记忆有本质区别，不只是"存得更多"。

先说这份代码里的实际情况
看 Store-Code.py：
store.put(namespace, str(uuid.uuid4()), {"data": last_message.content})
...
store.put(namespace, str(uuid.uuid4()), {"data": response.content})
这里每一条消息原封不动地整条塞进store，没有任何筛选、摘要、去重逻辑。所以在这份demo里，你说的"store只是存得更多"是成立的——它俩本质上就是两个key不同（thread_id vs user_id）的原始日志堆，没有任何"智能"加工。这是一个教学用的最简API演示，不代表真实长期记忆系统的设计。

但真正的智能体长期记忆，架构上不是这样
短期记忆（working memory / checkpoint）
目的：让同一次对话内的多轮交互连贯，作用域=一个thread的生命周期
常见处理方式（你说的基本对）：
滑动窗口，只保留最近N条
超过token预算时对旧消息做摘要，替换成一条压缩的system message（LangGraph里常见的SummarizationNode模式）
按重要性/相关性裁剪
长期记忆（long-term memory）
真正的实现通常不是"把短期记忆里有价值的东西put进去"这么简单，而是一整套独立流水线：

提取（extraction）：用一次额外的LLM调用，从对话中判断"这句话是不是值得记住的事实"（比如"我叫sam"是事实，"今天天气不错"大概率不是）
去重/更新（update, not append）：如果store里已经有"用户叫XXX"这条记忆，新对话里用户改名了，应该是覆盖/更新这条记忆，而不是无限往同一个bucket里append——这份demo代码的致命问题就在这，它是纯append，永远不会更新或去重，随对话增长store会无限膨胀
分类：很多长期记忆框架（比如LangMem、MemGPT/Letta）会分成不同类型：
语义记忆：事实性信息（用户叫什么、偏好什么）
情景记忆：具体发生过的事件/经历
程序性记忆：怎么做某类任务的经验/规则
检索时的相关性过滤：不是像这份代码这样把某个user_id下所有记忆一股脑拼进prompt（Store-Code.py: store.search没有传query/limit，等于全量拿），而是按语义相似度检索最相关的少数几条
一句话总结
你说的"短期记忆截断/摘要、长期记忆摘录整理"这个直觉是对的，这确实是业界的核心思路；但这份代码只实现了"存储空间+命名空间隔离"这个最基础的骨架，摘录、去重、更新、相关性检索这些"整理"逻辑完全没写，属于最简化的教学示例，离真实产品级长期记忆还差一整层"记忆管理智能"。

发表评论

您的邮箱地址不会被公开。 必填项已用 * 标注

滚动至顶部