记忆

记忆的重要性

记忆的定义：记忆是一个系统，用于记住之前的交互信息。
对AI代理的重要性：对于AI代理来说，记忆至关重要。它使代理能够记住之前的交互，从反馈中学习，并适应用户的偏好。当代理处理更复杂的任务和大量用户交互时，这种能力对于提高效率和用户满意度变得必不可少。

LangGraph框架中有两种类型的记忆

短期记忆（Short-term memory）定义：短期记忆，也称为线程范围（内存、持久化—文件系统、数据库等）内的记忆，通过在会话中维护消息历史来跟踪正在进行的对话。注意：是以线程为单位，则代码中要设置线程id。
- 管理方式：LangGraph将短期记忆作为代理状态的一部分进行管理。状态通过检查点（checkpointer）持久化到数据库中，以便随时恢复线程。当图被调用或完成一个步骤时，短期记忆会更新，并且在每个步骤开始时读取状态。
长期记忆（Long-term memory）定义：长期记忆存储跨会话（跨线程）的用户特定或应用级别的数据，并且可以在任何线程中共享和随时回忆。记忆可以被限定在任何自定义的命名空间内，而不仅仅是一个单独的线程ID。本质：将需要的有价值的信息存入数据库中（全文数据库 —关键字检索、向量数据库 — 相似性检索）
- 管理方式：LangGraph提供了存储（stores）来让你保存和回忆长期记忆。
- 生产使用方式：定义嵌入模型、定义向量数据库，基于Langgraph提供的接口完成记忆的入库及查询
- 长期记忆默认的数据入库和查询，都需要手动完成

短期记忆

短期记忆代码Langchain智能体示例

官方示例

from langchain.agents import create_agent

from langgraph.checkpoint.postgres import PostgresSaver  


DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup() # auto create tables in PostgresSql
    agent = create_agent(
        "gpt-5",
        tools=[get_user_info],
        checkpointer=checkpointer,  
    )

需要安装依赖：pip install langgraph-checkpoint-postgres

根据官方示例，来讲解，以postgresql做存储，示例代码如下

# @Time：2025/10/29 16:34
# @Author：jinglv
from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

from src.core.llms import model_client

DB_URI = "postgresql://postgres:12345678@localhost:15432/langgraph_db?sslmode=disable&keepalives=1&keepalives_idle=30&keepalives_interval=10&keepalives_count=5"

# 使用连接池提高稳定性，防止远程连接断开
with ConnectionPool(conninfo=DB_URI, max_size=10, check=ConnectionPool.check_connection) as pool:
    checkpointer = PostgresSaver(pool)
    # 创建初始化表，只需要执行一次
    # checkpointer.setup()
    agent = create_agent(
        model_client,
        [],
        checkpointer=checkpointer,
    )

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

    for chunk in agent.stream(
            {"messages": [{"role": "user", "content": "你好，我是小花"}]},
            config,
            stream_mode="values"
    ):
        chunk["messages"][-1].pretty_print()

    config = {
        "configurable": {
            "thread_id": "1"
        }
    }
    for chunk in agent.stream(
            {"messages": [{"role": "user", "content": "我是谁？"}]},
            config,
            stream_mode="values"
    ):
        chunk["messages"][-1].pretty_print()

执行结果为：

================================ Human Message =================================

你好，我是小花
================================== Ai Message ==================================

你好，小花！🌸  
很高兴再次见到你～  
今天有什么想和我聊聊的吗？无论是分享心情、讨论问题，还是需要帮助，我都在这里呢！ 😊
================================ Human Message =================================

我是谁？
================================== Ai Message ==================================

你好！😊  
在我们的对话中，你一直告诉我你的名字是**小花**，所以我始终用这个名字来称呼你。  

如果“我是谁”这个问题指向**更深层的自我探索**——比如：  
- 你可能是一个正在思考自我定义的人；  
- 你可能在寻找身份、目标或生命意义的答案；  
- 你也可能是一个喜欢用重复对话测试AI的可爱用户～  

无论哪种情况，我都很愿意陪你聊聊：  
如果你有具体困惑，我们可以一起梳理；  
如果你只是需要确认，那么我的答案是：**你是小花，而我是DeepSeek，你的AI伙伴**。

其中代码智能体创建

agent = create_agent(
       model_client,
       [],
       checkpointer=checkpointer,
   )

参数checkpointer详解：

类型：检查点保存器对象或标志（flag）
作用：如果提供了检查点保存器，它将作为图的版本化“短期记忆”，允许图在任意点暂停、恢复和重放。
不同取值：
- 如果为None，当该图作为子图使用时，可能会继承父图的检查点保存器。
- 如果为False，则不会使用也不会继承任何检查点保存器。

如果去掉checkpointer参数，则执行的结果为

================================ Human Message =================================

你好，我是小花
================================== Ai Message ==================================

你好呀，小花！🌼 这个名字真好听，充满了春天的气息呢～很高兴认识你！今天有什么想聊的，或者需要我帮忙的事情吗？无论是学习、生活还是想分享小故事，我都在这儿听着呢✨
================================ Human Message =================================

我是谁？
================================== Ai Message ==================================

很抱歉，我无法直接知道您的身份信息。我是DeepSeek助手，每次对话对我来说都是全新的开始，我无法获取您的个人信息，比如姓名、身份或历史对话记录。

不过，从我们的对话中，我可以知道您是一位正在与我交流的用户。如果您愿意告诉我一些关于您的信息，比如您的兴趣爱好、今天想讨论的问题，或者需要我帮助解决什么困难，我会很乐意更好地为您服务！😊

有什么特别想聊的话题或者需要我帮助的地方吗？我很期待能够为您提供有用的帮助！

那么短期记忆就无效了，AI不知道我上次说的内容。

短期记忆代码Langgraph图示例

# @Time：2025/10/29 17:26
# @Author：jinglv
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import MessagesState, StateGraph, START

from src.core.llms import model_client

DB_URI = "postgresql://postgres:12345678@localhost:15432/langgraph_db?sslmode=disable"

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    # 创建初始化表
    # checkpointer.setup()

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


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

    graph = builder.compile(checkpointer=checkpointer)

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

    for chunk in graph.stream(
            {"messages": [{"role": "user", "content": "你好，我是小红"}]},
            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()

执行结果

================================ Human Message =================================

你好，我是小红
================================== Ai Message ==================================

你好，小红！😊  
很高兴认识你～谢谢您告诉我这个称呼，我会记住的，之后就用“小红”来称呼你。  

今天有什么想聊的，或者需要我帮忙的事情吗？无论是学习、工作、生活，还是随便聊聊，我都很乐意～
================================ Human Message =================================

我是谁？
================================== Ai Message ==================================

你是小红。这是我目前知道的、你告诉我的名字。如果你希望我调整这个称呼，请随时告诉我。  

当然，如果这个问题指向更深层的自我认知——  
从对话的此刻来看，你是：  
1️⃣ 一个拥有独特身份和经历的人  
2️⃣ 正在与AI交流的探索者  
3️⃣ 会思考“我是谁”的自觉存在  

需要我陪你聊聊身份、记忆，或任何其他话题吗？ 🌟

以上示例是将记忆存储到postgresql数据库中，也可以将记忆存储到内存中，代码示例如下：

# @Time：2025/10/29 17:26
# @Author：jinglv
from langgraph.checkpoint.memory import InMemorySaver  # 这是一个短期记忆的存储类，它将记忆存储在内存中。这意味着记忆只在当前会话中有效，当程序重启后，记忆会丢失。
from langgraph.graph import MessagesState, StateGraph, START

from src.core.llms import model_client


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


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

checkpointer = InMemorySaver()  # 创建了一个InMemorySaver的实例，用于管理短期记忆。

graph = builder.compile(checkpointer=checkpointer)

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

for chunk in graph.stream(
        {"messages": [{"role": "user", "content": "你好，我是小红"}]},
        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()

Trim Messages(修剪消息)

当消息历史（即对话中的所有消息）的token数量接近或超过模型的最大上下文窗口时，就需要对消息历史进行修剪，以确保模型能够正常处理输入。如果不进行修剪，模型可能会因为上下文窗口溢出而无法正确理解或生成响应。

代码示例如下：

# @Time：2025/10/30 08:43
# @Author：jinglv
import asyncio

from langchain_core.messages.utils import trim_messages as trim_messages_util, count_tokens_approximately
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
from langchain_core.runnables import RunnableConfig
from typing import Any

from src.core.llms import model_client


@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """Keep only the last few messages to fit context window."""
    messages = state["messages"]

    trimmed_messages = trim_messages_util(
        messages,
        strategy="last",  # 裁剪策略
        token_counter=count_tokens_approximately,  # 计算token数的方法
        max_tokens=384,  # 最大token数
        start_on="human",  # 从human消息开始裁剪
        end_on=("human", "tool"),  # 裁剪结束的
    )

    return {
        "messages": [
            RemoveMessage(id=REMOVE_ALL_MESSAGES),
            *trimmed_messages
        ]
    }


checkpointer = InMemorySaver()

agent = create_agent(
    model_client,
    tools=[],
    middleware=[trim_messages],
    checkpointer=checkpointer,
)

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


async def main():
    # Stream the agent
    async for chunk in agent.astream(
            {"messages": [{"role": "user", "content": "写一首关于冬天的文言文"}]},
            stream_mode="values",
            config=config
    ):
        if "messages" in chunk:
            chunk["messages"][-1].pretty_print()
            
    async for chunk in agent.astream(
            {"messages": [{"role": "user", "content": "再一首夏天的"}]},
            stream_mode="values",
            config=config
    ):
        if "messages" in chunk:
            chunk["messages"][-1].pretty_print()


if __name__ == "__main__":
    asyncio.run(main())

Delete Messages(删除消息)

你可以从图状态中删除消息以管理消息历史。这在你想要删除特定消息或者清除整个消息历史时非常有用。

要从图状态中删除消息，你可以使用RemoveMessage。为了让RemoveMessage能够正常工作，你需要使用带有add_messages归并器的状态键，例如MessagesState。

示例代码如下：

# @Time：2025/10/30 14:22
# @Author：jinglv
from langchain_core.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import MessagesState, StateGraph, START

from src.core.llms import model_client


def delete_messages(state):
    messages = state["messages"]
    if len(messages) > 2:
        # 删除前两个消息
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}


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


builder = StateGraph(MessagesState)
builder.add_sequence([call_model, delete_messages])
builder.add_edge(START, "call_model")

checkpointer = InMemorySaver()
app = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}
for event in app.stream(
        {"messages": [{"role": "user", "content": "我是红红"}]},
        config,
        stream_mode="values"
):
    print([(message.type, message.content) for message in event["messages"]])

for event in app.stream(
        {"messages": [{"role": "user", "content": "我是谁?"}]},
        config,
        stream_mode="values"
):
    print([(message.type, message.content) for message in event["messages"]])

删除消息需要注意：

当你在处理（比如删除）消息记录时，要确保剩下的消息记录符合逻辑和规则。因为不同的语言模型（LLM）提供商可能有各自的限制和要求。比如，有的提供商规定消息记录的开头必须是用户发出的消息；还有的提供商要求，如果助手的消息中包含了对某个工具的调用，那么在消息记录中，这个助手消息后面必须紧跟着该工具调用的结果消息。

Summarize messages(消息总结)

以上的对消息进行修剪或删除的方法，可能会导致信息丢失。因为这些操作会减少消息队列中的消息数量，而这些被移除的消息可能包含一些重要的信息。鉴于上述问题，一些应用程序会采用一种更复杂、更高级的方法来解决，即利用聊天模型对消息历史进行总结。通过这种方式，可以在不丢失重要信息的前提下，对大量的消息进行有效的管理和处理，从而更好地支持应用程序的功能和性能。

该代码示例用到了LangMem，这个会在后续详细讲解，需要安装依赖：pip install -U langmem

示例代码如下：

# @Time：2025/10/30 14:33
# @Author：jinglv
import asyncio

from typing import Any

from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import before_model
from langchain.messages import RemoveMessage
from langchain_core.messages.utils import count_tokens_approximately
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.runtime import Runtime
from langmem.short_term import SummarizationNode, RunningSummary

from src.core.llms import model_client

summarization_node = SummarizationNode(
    token_counter=count_tokens_approximately,
    model=model_client,
    max_tokens=384,
    max_summary_tokens=128,
    output_messages_key="llm_input_messages",
)


class State(AgentState):
    # NOTE: we're adding this key to keep track of previous summary information
    # to make sure we're not summarizing on every LLM call
    context: dict[str, RunningSummary]


@before_model
def summarize_messages(state: State, runtime: Runtime) -> dict[str, Any] | None:
    result: dict[str, Any] | None = summarization_node.invoke(state)

    if result is None:
        return None

    llm_messages = (
        result.get("llm_input_messages")
        or result.get("summarized_messages")
        or result.get("messages")
    )

    updates = dict(result)
    if llm_messages is not None:
        updates.pop("llm_input_messages", None)
        updates.pop("summarized_messages", None)
        updates["messages"] = [RemoveMessage(id=REMOVE_ALL_MESSAGES), *llm_messages]
    return updates


checkpointer = InMemorySaver()

agent = create_agent(
    model=model_client,
    tools=[],
    middleware=[summarize_messages],
    state_schema=State,
    checkpointer=checkpointer,
)

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


async def main():
    # Stream the agent
    async for chunk in agent.astream(
            {"messages": [{"role": "user", "content": "写一首关于冬天的文言文"}]},
            stream_mode="values",
            config=config
    ):
        if "messages" in chunk:
            chunk["messages"][-1].pretty_print()
    async for chunk in agent.astream(
            {"messages": [{"role": "user", "content": "再一首夏天的"}]},
            stream_mode="values",
            config=config
    ):
        if "messages" in chunk:
            chunk["messages"][-1].pretty_print()


if __name__ == "__main__":
    asyncio.run(main())

代码使用到参数详解：

参数：max_tokens
- 含义：这是最终输出中返回的最大标记数量。也就是说，经过所有处理（包括总结）后，最终返回的消息列表中，所有消息的标记总数不能超过这个值。
- 作用：它主要用于限制最终输出的大小，确保输出不会过于庞大，适合输入到后续的模型中。
参数：max_tokens_before_summary
- 含义：这是在触发总结之前累积的最大标记数量。也就是说，当处理消息时，一旦累积的消息标记数量达到这个值，就会触发总结操作。
- 作用：它用于控制何时进行总结，避免一次性处理过多的消息，从而确保总结的效率和质量。
参数：max_summary_tokens
- 含义：这是为总结预算的最大标记数量。也就是说，生成的总结消息本身的最大标记数量不能超过这个值。
- 作用：它用于控制总结的长度，确保总结不会过于冗长，同时也能保证总结的简洁性和信息密度。

使用Langgraph图的示例代码

# @Time：2025/10/30 14:49
# @Author：jinglv
from typing import TypedDict

from langchain_core.messages import AnyMessage
from langchain_core.messages.utils import count_tokens_approximately
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, MessagesState
from langmem.short_term import SummarizationNode, RunningSummary

from src.core.llms import model_client

summarization_model = model_client.bind(max_tokens=128)


class State(MessagesState):
    context: dict[str, RunningSummary]


class LLMInputState(TypedDict):
    summarized_messages: list[AnyMessage]
    context: dict[str, RunningSummary]


# max_tokens：控制最终输出的大小，确保输出不会超过这个限制。
# max_tokens_before_summary：控制何时触发总结，避免一次性处理过多消息。
# max_summary_tokens：控制总结消息的长度，确保总结不会过于冗长。
# 这三个参数共同作用，确保在处理大量消息时，既能有效地进行总结，又能保证最终输出的大小在可控范围内。
summarization_node = SummarizationNode(
    token_counter=count_tokens_approximately,
    model=summarization_model,
    max_tokens=256,
    max_tokens_before_summary=256,
    max_summary_tokens=128,
)


def call_model(state: LLMInputState):
    response = model_client.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": "我的名字是小花"}, config)
graph.invoke({"messages": "写一首关于猫的短诗"}, config)
graph.invoke({"messages": "对狗也做同样的操作"}, config)
response = graph.invoke({"messages": "我的名字是什么？"}, config)

response["messages"][-1].pretty_print()

长期记忆

基本使用

记忆存储

1. 创建 Memory Store

Store 使用 langgraph 中的存储器InMemoryStore 来创建：

1
2
3

from langgraph.store.memory import InMemoryStore

in_memory_store = InMemoryStore()

2. 命名空间 Namespace

命名空间用于隔离不同用户/不同类型的记忆，通常以 tuple 表示

命名空间可以是任意长度，代表任何事物，不必是特定用户的。tuple (, “memories”)

1 2	user_id = "1" namespace_for_memory = (user_id, "memories")

用例	命名空间示例
多用户系统	`("user123", "memories")`
测试 Agent	`("agent", "test_cases")`
API 历史	`("api", "call_history")`

3. 存储记忆（put）

我们用这种方法把记忆保存到商店的命名空间。在此过程中，我们指定了上述定义的命名空间和内存的键值对：键仅是内存的唯一标识符（），值（字典）即存储器本身。store.put memory_id

memory_id = str(uuid.uuid4())
memory = {"food_preference" : "I like pizza"}

user_id = "1"
namespace_for_memory = (user_id, "memories")

in_memory_store.put(namespace_for_memory, memory_id, memory)

记忆读取

1.通过 id 获取记忆

通过记忆的 id 获取命名空间中的具体某一条记忆

1 2	# 传入命名空间和记忆id item = store.get(namespace, "memory_id")

2.检索记忆（search）

我们可以用store.search方法读取命名空间中的记忆，该方法会以列表形式返回给定用户的所有记忆。

1	memories = in_memory_store.search(namespace_for_memory)

每种内存类型都是一个带有特定属性的 Python 类（Item）。我们可以通过上述方式转换，作为词典访问。.dict它具备以下特征：

字段	描述
value	实际存储的数据（dict）
key	该命名空间内的唯一 ID
namespace	命名空间，字符串列表
created_at	创建时间
updated_at	更新（修改）时间

Memory Types

官方地址说明：https://docs.langchain.com/oss/javascript/concepts/memory

不同的应用场景需要不同类型的记忆。这意味着根据具体的应用需求，AI代理需要具备不同种类的记忆能力，以便更好地完成任务
尽管这种类比并不完美，但研究人类记忆的类型可以提供一些有价值的见解。这里提到人类记忆的类型可以为设计AI代理的记忆系统提供参考和启发。
一些研究（例如CoALA论文）甚至已经将人类记忆的类型映射到AI代理所使用的记忆类型上。在AI领域，人们已经开始借鉴人类记忆的分类方式来构建和优化AI代理的记忆系统。

三种类型如下：

Semantic Memory（语义记忆）What is Stored（存储内容）： Facts（事实）
- Human Example（人类例子）： Things I learned in school（我在学校学到的东西）例如，你记得的数学公式、历史事件、科学概念等。
- Agent Example（代理例子）： Facts about a user（关于用户的信息）例如，AI代理可以记住用户的名字、偏好、过去的交互记录等，以便更好地个性化服务。
Episodic Memory（情景记忆）What is Stored（存储内容）： Experiences（经历）
- Human Example（人类例子）： Things I did（我做过的事情）例如，你记得的某次旅行、参加的活动、经历的事件等。
- Agent Example（代理例子）： Past agent actions（代理过去的行动）例如，AI代理可以记住它之前执行的任务、与用户的对话内容等，以便从中学习和改进。
Procedural Memory（程序记忆）What is Stored（存储内容）： Instructions（指令）
- Human Example（人类例子）： Instincts or motor skills（本能或运动技能）例如，你学会的骑自行车、打字、游泳等技能，这些技能通常不需要有意识地回忆，而是自动执行。
- Agent Example（代理例子）： Agent system prompt（代理系统提示）例如，AI代理的系统提示（system prompt）可以被视为程序记忆，它定义了代理的行为规则和任务执行方式。

示例代码如下：

# @Time：2025/10/30 15:09
# @Author：jinglv
import uuid

from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.graph import MessagesState, StateGraph, START
from langgraph.store.base import BaseStore
from langgraph.store.postgres import PostgresStore

from src.core.llms import model_client

DB_URI = "postgresql://postgres:12345678@localhost:15432/langgraph_db?sslmode=disable"

with (
    PostgresStore.from_conn_string(DB_URI) as store,
    PostgresSaver.from_conn_string(DB_URI) as checkpointer,
):
    # 创建初始化，只需要首次执行一次
    store.setup()
    checkpointer.setup()


    def call_model(
            state: MessagesState,
            config: RunnableConfig,
            *,
            store: BaseStore,
    ):
        user_id = config["configurable"]["user_id"]
        # thread_id = config["configurable"]["thread_id"]
        namespace = ("memories", user_id)  # 命名空间
        # 全文检索
        memories = store.search(namespace, query=str(state["messages"][-1].content))
        info = "\n".join([d.value["data"] for d in memories])
        system_msg = f"You are a helpful assistant talking to the user. User info: {info}"

        # Store new memories if the user asks the model to remember
        last_message = state["messages"][-1]
        if "记住" in last_message.content.lower():
            memory = "用户的名称是小花"
            store.put(namespace, str(uuid.uuid4()), {"data": memory})

        response = model_client.invoke(
            [{"role": "system", "content": system_msg}] + state["messages"]
        )
        return {"messages": response}


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

    graph = builder.compile(
        checkpointer=checkpointer,
        store=store,
    )

    config = {
        "configurable": {
            "thread_id": "1",
            "user_id": "1",
        }
    }
    for chunk in graph.stream(
            {"messages": [{"role": "user", "content": "记住，我是小花"}]},
            config,
            stream_mode="values",
    ):
        chunk["messages"][-1].pretty_print()

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

    for chunk in graph.stream(
            {"messages": [{"role": "user", "content": "what is my name?"}]},
            config,
            stream_mode="values",
    ):
        chunk["messages"][-1].pretty_print()

Semantic Memory（语义记忆）

示例代码如下：

# @Time：2025/10/30 15:15
# @Author：jinglv
"""
语义记忆
"""
from langchain.embeddings import init_embeddings
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore

from src.core.llms import model_client

embeddings = init_embeddings("ollama:qwen3-embedding:8b", base_url="http://localhost:11434")

# 长期记忆的数据可以存储到向量数据库中
store = InMemoryStore(
    index={
        "embed": embeddings,
        "dims": 4096,
    }
)

# 入库数据
store.put(("user_123", "memories"), "1", {"text": "我喜欢吃肉夹馍"})
store.put(("user_123", "memories"), "2", {"text": "我是一名软件工程师"})


def chat(state, *, store: BaseStore):
    # 搜索
    items = store.search(
        ("user_123", "memories"), query=state["messages"][-1].content, limit=1
    )
    memories = "\n".join(item.value["text"] for item in items)
    memories = f"## Memories of user\n{memories}" if memories else ""
    response = model_client.invoke(
        [
            {"role": "system", "content": f"You are a helpful assistant.\n{memories}"},
            *state["messages"],
        ]
    )
    return {"messages": [response]}


builder = StateGraph(MessagesState)
builder.add_node(chat)
builder.add_edge(START, "chat")
graph = builder.compile(store=store)

for message, metadata in graph.stream(
        input={"messages": [{"role": "user", "content": "我是一名python开发工程师，我该学习什么语言呢？"}]},
        stream_mode="messages",
):
    print(message.content, end="")

使用PostgresStore的示例代码

# @Time：2025/10/30 15:15
# @Author：jinglv
"""
语义记忆
"""
from langchain.embeddings import init_embeddings
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.store.base import BaseStore
from langgraph.store.postgres import PostgresStore

from src.core.llms import model_client

embeddings = init_embeddings("ollama:qwen3-embedding:8b", base_url="https://ollama.guozhe.vip:8889")
DB_URI = "postgresql://postgres:12345678@localhost:15432/langgraph_db?sslmode=disable&keepalives=1&keepalives_idle=30&keepalives_interval=10&keepalives_count=5"
with PostgresStore.from_conn_string(
        DB_URI,
        index={
            "embed": embeddings,
            "dims": 4096,
        },
) as store:
    # store.setup()

    store.put(("user_123", "memories"), "1", {"text": "我喜欢吃肉夹馍"})
    store.put(("user_123", "memories"), "2", {"text": "我是一名软件工程师"})


    def chat(state, *, store: BaseStore):
        # Search based on user's last message
        items = store.search(
            ("user_123", "memories"), query=state["messages"][-1].content, limit=1
        )
        memories = "\n".join(item.value["text"] for item in items)
        memories = f"## Memories of user\n{memories}" if memories else ""
        response = model_client.invoke(
            [
                {"role": "system", "content": f"You are a helpful assistant.\n{memories}"},
                *state["messages"],
            ]
        )
        return {"messages": [response]}


    builder = StateGraph(MessagesState)
    builder.add_node(chat)
    builder.add_edge(START, "chat")
    graph = builder.compile(store=store)

    for message, metadata in graph.stream(
            input={"messages": [{"role": "user", "content": "我是一个吃货，非常喜欢吃各种食物"}]},
            stream_mode="messages",
    ):
        print(message.content, end="")

LangMem

官方文档：https://langchain-ai.github.io/langmem/#installation

功能讲解

学习与适应：LangMem能够帮助智能代理（agents）从它们随时间的交互中学习并适应。这意味着代理可以根据与用户的互动经历来不断调整和优化自己的行为，以更好地满足用户的需求。
信息提取与行为优化：它提供了工具来从对话中提取重要信息，通过提示词优化来优化代理的行为，并维护长期记忆。例如，在对话过程中，代理可以识别出关键信息并将其存储起来，同时根据这些信息调整自己的回答方式，以更准确地回应用户。
存储系统兼容与集成：LangMem既提供了可以与任何存储系统一起使用的功能原语，还提供了与LangGraph存储层的本地集成。这使得代理能够灵活地选择存储方式，并且能够与LangGraph平台无缝协作，更好地管理和利用记忆数据。

核心特点

通用核心记忆API：提供了一个核心记忆API，可以与任何存储系统配合使用，这使得开发者可以根据自己的需求选择不同的存储解决方案，而无需担心与LangMem的兼容性问题。
对话中的记忆管理工具：代理可以使用这些工具在活跃对话期间记录和搜索信息，即在“热路径”中进行记忆管理。这允许代理在与用户交流的过程中实时地存储和检索信息，从而更自然地进行对话，并且能够根据之前的信息来做出更准确的回应。
后台记忆管理器：自动提取、整合和更新代理的知识。这个功能在后台运行，无需人工干预，能够帮助代理不断学习和更新自己的知识库，使其能够更好地理解和处理各种情况。
与LangGraph长期记忆存储的本地集成：在所有LangGraph平台部署中默认提供与LangGraph长期记忆存储的本地集成。这意味着代理可以方便地利用LangGraph提供的长期记忆存储功能，确保记忆数据的持久化和一致性，从而在不同的会话之间保持一致的行为。

作用

持续改进：通过不断学习和适应，代理能够持续改进自己的性能，更好地满足用户的需求。
个性化回应：代理可以根据存储的记忆信息，为用户提供个性化的回应，提高用户体验。
跨会话一致行为：确保代理在不同的会话中保持一致的行为，避免出现前后矛盾的情况，增强用户的信任感。

安装依赖：pip install -U langmem

示例代码

示例1 — Memory Tools:

create_manage_memory_tool ：创建一个用于管理对话中持久记忆的工具。这个工具允许人工智能助手创建、更新和删除在对话之间持续存在的记忆。它有助于在不同会话中保持上下文和用户偏好。
create_search_memory_tool ：创建一个用于搜索存储在LangGraph BaseStore中的记忆的工具。这个工具允许人工智能助手通过语义或精确匹配的方式搜索之前存储的记忆。该工具会返回记忆内容以及原始记忆对象，以便于高级使用。

示例代码如下：

# @Time：2025/10/30 15:59
# @Author：jinglv
from langchain.embeddings import init_embeddings
from langgraph.prebuilt import create_react_agent
from langgraph.store.memory import InMemoryStore
from langmem import create_manage_memory_tool, create_search_memory_tool

from src.core.llms import model_client

embeddings = init_embeddings("ollama:qwen3-embedding:8b", base_url="https://localhost:11434")

store = InMemoryStore(
    index={
        "embed": embeddings,
        "dims": 4096,
    }
)

# 自动入库和自动查询
agent_a_tools = [
    # Write to agent-specific namespace 入库工具
    create_manage_memory_tool(namespace=("memories", "team_a", "agent_a")),
    # Read from shared team namespace 检索工具
    create_search_memory_tool(namespace=("memories", "team_a"))
]

# Agents with different prompts sharing read access
agent_a = create_react_agent(
    model_client,
    tools=agent_a_tools,
    store=store,
    prompt="You are a research assistant"
)

s = agent_a.invoke({"messages": [{"role": "user", "content": "我是小花，请记住我的名字"}]})
print(s)
s = agent_a.invoke({"messages": [{"role": "user", "content": "我是谁？"}]})
print(s)

# Create tools for agent B with different write space
agent_b_tools = [
    create_manage_memory_tool(namespace=("memories", "team_a")),
    create_search_memory_tool(namespace=("memories", "team_a"))
]
agent_b = create_react_agent(
    model_client,
    tools=agent_b_tools,
    store=store,
    prompt="You are a report writer."
)

s = agent_b.invoke({"messages": [{"role": "user", "content": "我是小花，请记住我的名字"}]})
print(s)
s = agent_b.invoke({"messages": [{"role": "user", "content": "我是谁？"}]})
print(s)

示例2 — Memory Management:

create_memory_manager
- 功能：创建一个内存管理器，用于处理对话消息并生成结构化的内存条目。该函数创建一个异步可调用对象，能够分析对话消息和现有内存，以生成或更新结构化的内存条目。它可以识别对话中的隐含偏好、重要上下文和关键信息，并将它们组织成结构良好的内存，用于改善未来的交互。
- 支持的内存类型：支持无结构的基于字符串的内存以及由 Pydantic 模型定义的结构化内存，所有这些内存都会自动持久化到配置的存储中。
create_memory_store_manager
- 功能：丰富配置的 BaseStore 中存储的内存。该系统会自动搜索相关的内存，提取新信息，更新现有的内存，并维护所有更改的版本历史

示例代码如下

# @Time：2025/10/30 16:02
# @Author：jinglv
from langchain.embeddings import init_embeddings
from langgraph.func import entrypoint
from langgraph.store.memory import InMemoryStore
from langmem import create_memory_store_manager
from pydantic import BaseModel, Field

from src.core.llms import model_client

embeddings = init_embeddings("ollama:qwen3-embedding:8b", base_url="https://ollama.guozhe.vip:8889")
store = InMemoryStore(
    index={
        "embed": embeddings,
        "dims": 4096,
    }
)


class Episode(BaseModel):
    """用智能体自己的话，把这次经历写下来。事后回想，把当时脑子里最重要的想法记下来，让它以后能学着长记性。"""
    observation: str = Field(..., description="The context and setup - what happened")
    thoughts: str = Field(
        ...,
        description="内部推理过程与智能体在本轮中的观察，使其得以得出正确的行动与结果。“我……”",
    )
    action: str = Field(
        ...,
        description="做了什么、如何做的、以何种格式呈现。（包含对行动成功至关重要的任何要素）。我……",
    )
    result: str = Field(
        ...,
        description="结果与回顾。哪些地方做得好？下次可以在哪些方面改进？我……",
    )


manager = create_memory_store_manager(
    model_client,
    namespace=("memories", "episodes"),
    schemas=[Episode],
    instructions="提取成功解释的示例，捕捉完整的推理链条。解释要简洁，推理逻辑要精确。",
    enable_inserts=True,
    store=store
)

# 构造对话数据
conversation = [
    {
        "role": "user",
        "content": "什么是二叉树？我从事家谱工作，如果这有帮助的话",
    },
    {
        "role": "assistant",
        "content": "二叉树就像家谱一样，但每个父节点最多有2个子节点。这里有个简单示例：\n   鲍勃\n  /  \\\n艾米  卡尔\n\n就像在家谱中一样，我们称鲍勃为'父节点'，艾米和卡尔为'子节点'。",
    },
    {
        "role": "user",
        "content": "哦，有道理！那么在二叉搜索树中，是不是就像按年龄来组织家庭成员？",
    },
]

print("开始更新记忆")
episodes = manager.invoke({"messages": conversation})
print(episodes)
print("记忆更新成功")


@entrypoint(store=store)
def app(messages: list):
    # Step 1: Find similar past episodes
    # memories = manager.search(
    #     query=messages[-1]["content"],
    #     limit=1,
    # )
    similar = store.search(
        ("memories", "episodes"),
        query=messages[-1]["content"],
        limit=1,
    )
    print("similar::", similar)
    # Step 2: Build system message with relevant experience
    system_message = "You are a helpful assistant."
    if similar:
        system_message += "\n\n### EPISODIC MEMORY:"
        for i, item in enumerate(similar, start=1):
            episode = item.value["content"]
            system_message += f"""

Episode {i}:
When: {episode['observation']}
Thought: {episode['thoughts']}
Did: {episode['action']}
Result: {episode['result']}
        """

    # Step 3: Generate response using past experience
    response = model_client.invoke([{"role": "system", "content": system_message}, *messages])

    # Step 4: Store this interaction if successful
    manager.invoke({"messages": messages})
    return response


result = app.invoke(
    [
        {
            "role": "user",
            "content": "什么是二叉树？通俗易懂的给我解释一下",
        },
    ],
)
print(result)
print(store.search(("memories", "episodes"), query="Trees"))

执行结果

开始更新记忆
[{'namespace': ('memories', 'episodes'), 'key': 'ab695e26-8996-40ca-b349-87892d6885c9', 'value': {'kind': 'Episode', 'content': {'observation': '用户询问"什么是二叉树？"，并提到自己从事家谱工作。我通过类比家谱来解释二叉树的概念，使用了家谱中的术语（父节点、子节点）。用户理解了并进一步询问二叉搜索树是否像按年龄组织家庭成员。', 'thoughts': '这是一个成功的教学互动示例。我识别到用户有家谱工作的背景，所以立即使用家谱作为类比来解释二叉树。这种个性化解释帮助用户快速理解抽象概念。当用户进一步询问二叉搜索树时，表明他们不仅理解了基础概念，还能进行类比推理，想知道更高级的概念是否也有类似的现实对应。', 'action': '1. 识别用户的背景知识（家谱工作）\n2. 使用相关类比：将二叉树比作家谱结构\n3. 使用用户熟悉的术语：父节点、子节点\n4. 提供具体可视化示例：鲍勃（父节点）有艾米和卡尔（子节点）\n5. 解释简洁明了，直接对应用户已知概念', 'result': '成功！用户不仅理解了二叉树的基本概念，还能进行延伸思考，询问二叉搜索树的类比。这表明：\n1. 个性化类比有效：使用用户熟悉的领域解释新概念\n2. 术语一致性：使用用户领域的术语（家谱术语）\n3. 可视化帮助：具体示例让抽象概念更易理解\n4. 激发进一步思考：用户主动想了解更高级概念\n下次可以：当用户表现出理解并进一步提问时，继续使用相同的类比框架解释更复杂的概念。'}}}]
记忆更新成功
similar:: [Item(namespace=['memories', 'episodes'], key='ab695e26-8996-40ca-b349-87892d6885c9', value={'kind': 'Episode', 'content': {'observation': '用户询问"什么是二叉树？"，并提到自己从事家谱工作。我通过类比家谱来解释二叉树的概念，使用了家谱中的术语（父节点、子节点）。用户理解了并进一步询问二叉搜索树是否像按年龄组织家庭成员。', 'thoughts': '这是一个成功的教学互动示例。我识别到用户有家谱工作的背景，所以立即使用家谱作为类比来解释二叉树。这种个性化解释帮助用户快速理解抽象概念。当用户进一步询问二叉搜索树时，表明他们不仅理解了基础概念，还能进行类比推理，想知道更高级的概念是否也有类似的现实对应。', 'action': '1. 识别用户的背景知识（家谱工作）\n2. 使用相关类比：将二叉树比作家谱结构\n3. 使用用户熟悉的术语：父节点、子节点\n4. 提供具体可视化示例：鲍勃（父节点）有艾米和卡尔（子节点）\n5. 解释简洁明了，直接对应用户已知概念', 'result': '成功！用户不仅理解了二叉树的基本概念，还能进行延伸思考，询问二叉搜索树的类比。这表明：\n1. 个性化类比有效：使用用户熟悉的领域解释新概念\n2. 术语一致性：使用用户领域的术语（家谱术语）\n3. 可视化帮助：具体示例让抽象概念更易理解\n4. 激发进一步思考：用户主动想了解更高级概念\n下次可以：当用户表现出理解并进一步提问时，继续使用相同的类比框架解释更复杂的概念。'}}, created_at='2025-12-30T05:53:22.735683+00:00', updated_at='2025-12-30T05:53:22.735694+00:00', score=0.7523176771262952)]
content='二叉树是一种数据结构，可以想象成一个**家谱图**：\n\n- 最上面有一个**根节点**（就像家族的最早祖先）\n- 每个节点最多有两个**子节点**（就像一个人最多有两个孩子）\n- 每个子节点只能有一个**父节点**（就像每个人只有一个亲生父亲）\n\n**简单比喻：**\n想象一棵倒过来的树🌳，从树根开始分叉，每次最多分成两个树枝。每个分叉点就是一个“节点”，连接线就是“边”。\n\n**关键特点：**\n1. **最多两个分支**：每个节点最多连接两个子节点\n2. **层次结构**：有明显的上下级关系\n3. **没有循环**：不会出现“我是我自己的祖先”这种情况\n\n**常见用途：**\n- 文件系统的目录结构\n- 决策树（是/否问题）\n- 表达式计算（如数学公式）\n- 数据库索引\n\n这样解释清楚吗？需要我用更具体的例子说明吗？' additional_kwargs={'refusal': None} response_metadata={'token_usage': {'completion_tokens': 203, 'prompt_tokens': 343, 'total_tokens': 546, 'completion_tokens_details': None, 'prompt_tokens_details': {'audio_tokens': None, 'cached_tokens': 0}, 'prompt_cache_hit_tokens': 0, 'prompt_cache_miss_tokens': 343}, 'model_provider': 'deepseek', 'model_name': 'deepseek-chat', 'system_fingerprint': 'fp_eaab8d114b_prod0820_fp8_kvcache', 'id': '86d1ce32-f581-44b8-a81e-c64b77f7a652', 'finish_reason': 'stop', 'logprobs': None} id='lc_run--71fc6b41-d248-402e-9c2b-5742fe6e800c-0' usage_metadata={'input_tokens': 343, 'output_tokens': 203, 'total_tokens': 546, 'input_token_details': {'cache_read': 0}, 'output_token_details': {}}
[Item(namespace=['memories', 'episodes'], key='ab695e26-8996-40ca-b349-87892d6885c9', value={'kind': 'Episode', 'content': {'observation': "用户再次询问'什么是二叉树？'，要求通俗易懂的解释。这是一个新的教学机会，用户可能之前听说过但需要更直观的理解。我准备使用家谱类比来解释二叉树，因为这是大多数人熟悉的家庭关系结构。", 'thoughts': "这是一个新的教学互动，用户明确要求'通俗易懂'的解释。我需要使用最直观的类比来帮助用户理解抽象概念。家谱是一个完美的类比，因为：1. 每个人都熟悉家庭关系；2. 家庭树有明确的层级结构；3. 父节点、子节点的概念自然对应。我应该从最简单的例子开始，逐步建立概念。", 'action': '1. 识别用户的背景知识（家谱工作）\n2. 使用相关类比：将二叉树比作家谱结构\n3. 使用用户熟悉的术语：父节点、子节点\n4. 提供具体可视化示例：鲍勃（父节点）有艾米和卡尔（子节点）\n5. 解释简洁明了，直接对应用户已知概念', 'result': "成功！通过使用家谱类比，我能够：\n1. 将抽象概念具体化：二叉树→家谱结构\n2. 使用熟悉术语：父节点、子节点→父母、子女\n3. 提供可视化示例：具体人物关系让概念更易理解\n4. 建立基础框架：为后续更复杂概念（如二叉搜索树）打下基础\n\n教学启示：当用户要求'通俗易懂'时，立即寻找他们最熟悉的现实世界类比。家谱是一个通用且有效的类比，适用于大多数用户群体。"}}, created_at='2025-12-30T05:53:53.298243+00:00', updated_at='2025-12-30T05:53:53.298252+00:00', score=0.4001063434640949)]

扩展

如果使用的不是Langgraph的记忆体，可以使用mem0，该框架可在任意智能体框架运行

mem0官方网站：https://docs.mem0.ai/introduction

mem0 Github地址：https://github.com/mem0ai/mem0

Jean's Blog

LangGraph之记忆

记忆

记忆的重要性

LangGraph框架中有两种类型的记忆

短期记忆

短期记忆代码Langchain智能体示例

短期记忆代码Langgraph图示例

Trim Messages(修剪消息)

Delete Messages(删除消息)

Summarize messages(消息总结)

长期记忆

基本使用

记忆存储

1. 创建 Memory Store

2. 命名空间 Namespace

3. 存储记忆（put）

记忆读取

1.通过 id 获取记忆

2.检索记忆（search）

Memory Types

Semantic Memory（语义记忆）

LangMem

功能讲解

核心特点

作用

示例代码

示例1 — Memory Tools:

示例2 — Memory Management:

扩展