LangGraph之记忆

记忆

记忆的重要性

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

记忆的两种类型

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

短期记忆

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

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

# @Time：2025/10/29 16:34
# @Author：jinglv
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.postgres import PostgresSaver


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()
    agent = create_react_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 ==================================

哈哈，小花，你刚刚告诉过我你的名字呀！🌼 在之前的对话中，你提到自己叫“小花”，我会记得这个可爱又亲切的称呼～  

如果你问的是更深的“我是谁”——比如哲学层面的自我认知，我可能暂时无法替你回答。但如果你需要，我们可以一起探索你的兴趣、性格、目标或回忆，帮你更好地梳理自己的特点和想法～ 你愿意聊聊吗？ 😊

其中代码智能体创建

agent = create_react_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, count_tokens_approximately
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent

from src.core.llms import model_client


# 如果前置消息太长，超过了给定的token数，则进行消息的裁剪
def pre_model_hook(state):
    # token = count_tokens_approximately(messages=state["messages"])
    trimmed_messages = trim_messages(
        state["messages"],
        strategy="last",  # 裁剪策略
        token_counter=count_tokens_approximately,  # 计算token数的方法
        max_tokens=384,  # 最大token数
        start_on="human",  # 从human消息开始裁剪
        end_on=("human", "tool"),  # 裁剪结束的
    )
    return {"llm_input_messages": trimmed_messages}


checkpointer = InMemorySaver()

agent = create_react_agent(
    model_client,
    [],
    pre_model_hook=pre_model_hook,
    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 langchain_core.messages.utils import count_tokens_approximately
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent
from langgraph.prebuilt.chat_agent_executor import AgentState
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]


checkpointer = InMemorySaver()

agent = create_react_agent(
    model=model_client,
    tools=[],
    pre_model_hook=summarization_node,
    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()

长期记忆

Memory Types

• 不同的应用场景需要不同类型的记忆。这意味着根据具体的应用需求，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@82.157.193.65: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="")

LangMem

功能讲解

• 学习与适应：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"
)

# 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"))

扩展

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

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

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