推荐阅读列表：

LLM之Agent（三十二）｜AI Agents（一）简介

LLM之Agent（三十三）｜AI Agents（二）：从零开始构建Agent

LLM之Agent（三十四）｜AI Agents（三）：AI Agents框架介绍

LLM之Agent（三十五）｜AI Agents（四）：AI Agent类型

LLM之Agent（三十六）｜AI Agents（五）：Workflow vs Agent

LLM之Agent（三十七）｜AI Agents（六）：AI Agents架构

LLM之Agent（三十八）｜AI Agents（七）：Multi-Agent架构

LLM之Agent（三十九）｜AI Agents（八）：构建Multi-Agent系统

       想象一下，你和一个朋友聊天，他总是忘记你说过的所有话，每次对话都从零开始。你是什么感受？不幸的是，如今大多数人工智能系统都是如此。它们确实很智能，但却缺少一个至关重要的东西：记忆 。

       我们先来谈谈人工智能中“记忆”的真正含义以及它为何如此重要。

一、引言

1.1 当今AI的记忆幻觉

       像 ChatGPT 或编码助手这样的工具一开始会让人觉得很有用，但当你一遍又一遍地重复指令或偏好设置时，就会发现它们有很多使用不便的地方。此时，需要构建能够学习、进化和协作的智能体来扩展LLM能力，对于智能体来说，记忆至关重要。

       由上下文窗口和巧妙的提示设计所营造的记忆假象，让许多人误以为智能体已经“记住”了信息。但实际上，如今大多数智能体都是无状态的 ，它们既无法从过去的交互中学习，也无法随时间推移而适应。

       要从无状态工具过渡到真正智能、自主 （有状态） 代理，我们需要赋予它们记忆， 而不仅仅是更大的提示或更好的检索。

1.2 AI Agent中的“记忆”指的是什么？

       对AI agents来说， 记忆是指跨越时间、任务和多次用户交互来保留和回忆相关信息的能力。它使智能体能够记住过去发生的事情，并利用这些信息来改进未来的行为。

       记忆并非仅仅用于存储聊天记录或向提示框中填充更多tokens，它旨在构建一个持久的内部状态，该状态会不断演进并影响智能体的每一次交互，即使间隔数周或数月。

智能体的记忆由三大支柱构成：

• 状态： 了解当前正在发生的事情；

• 持久性： 跨会话保留知识；

• 选择： 决定哪些内容值得记住

1.3 记忆如何融入AI Agent系统

让我们把记忆融入到现代智能体的架构中，常见的组件包括：

• 推理和答案生成的 LLM；
• 策略或规划工具 （例如 ReAct、AutoGPT 等）；
• 访问工具/API；
• 用于检索文档或历史数据的检索器

问题在于： 这些组件都不记得之前发生的事情，没有内部状态，没有持续迭代更新的理解能力，也没有记忆。

记忆在智能体系统的位置：

这使得智能体从一次性助手转变为持续发展的协作者。

1.4 上下文窗口≠记忆

       一个常见的误解是：长上下文窗口可以忽略对记忆的需求。然而，使用长上下文的LLM会带来另一个问题：更多token=更高的成本和延迟。

1.5 为什么 RAG 与记忆不同？

       虽然 RAG（检索增强生成）和记忆系统都获取信息以支持 LLM，但它们解决的问题截然不同。

       RAG 在推理时将外部知识加入到提示词，有助于用文件事实为回答打基础。但 RAG 本质上是无状态的——它不了解之前的交互、用户身份，或当前查询与过去对话的关联。

       而记忆则是连续性，可以捕捉用户偏好、过去的查询、决策和失败，并在未来的互动中提供这些信息。

可以这样总结：

RAG 帮助智能体更好地回答。而记忆则帮助智能体行为更聪明。

二、智能体中的记忆类型

一般来说，AI agents中的记忆有两种形式：

• 短期记忆 ：在单次互动中保持即时语境；
• 长期记忆 ：能够在跨会话、任务和时间保持记忆。

       就像人类一样，这些记忆类型承担着不同的认知功能。短期记忆帮助智能体保持当下回话的连贯性。长期记忆帮助智能体保持持续学习、个性化和适应能力。

简单总结：

短期记忆=AI 在与你交谈时“记起”的内容。

长期记忆=AI 在多次对话后“学习和回忆”的内容。

2.1 短期记忆（或工作记忆）

       人工智能系统中最基本的记忆形式是上下文窗口内容——就像一个人记住刚才对话中说过的内容。这包括：

• 对话历史：最近消息及其顺序；
• 工作记忆：临时变量与状态；
• 注意力语境：当前对话的焦点

2.2 长期记忆

更复杂的人工智能应用需要长期记忆，以在对话中保留信息。

这包括如下几个方面：

2.2.1 程序内存

       它定义了你的智能体知道该做什么，直接编码在你的代码里。

       这就是你的智能体的“ 肌肉记忆 ”——学会的行为，自动形成的。就像你打字时不会有意识地去考虑每一次按键，你的客服也会内化“始终优先处理关于 API 文档的邮件”或“回答技术问题时用更乐观的语气”等流程。

2.2.2 情节记忆（示例）

       用户具体的过去互动和经历，这会保持智能体能力的连续性、个性化和长期学习。

       这是智能体的“ 相册 ”——关于过去事件和互动的具体记忆。就像你可能记得听到重要消息时身处何地，你的智能体也会记得，“上次这位客户发邮件说延期截止日期时，我的回复太僵硬，造成了摩擦”或者“当邮件里有'快速提问'这个词时，通常需要详细的技术解释，但这些解释根本不够快速。”

2.2.3 语义记忆（事实）

通过向量搜索或 RAG 检索的事实世界知识。

       这是你的智能体的“ 百科全书 ”——它学到的关于你世界的事实。就像你知道巴黎是法国首都，但不记得具体时间和方式一样，你的智能体也会记得“Alice 是 API 文档的联系人”或“John 更喜欢早会”。

三、记忆写入

       虽然人类常在睡眠中形成长期记忆，但AI agents需要不同的方法。AI agents应该何时以及如何创造新的记忆？AI agents至少有两种主要的记忆写入方法：“前台实时写入”和“后台写入”。

      这两种方式的流程示意图如下所示：

四、增加短期记忆

       短期记忆（线程级持久性）使智能体能够跟踪多回合对话。下面是LangGraph短期记忆的示例：

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.memory import InMemorySaver
model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
def call_model(state: MessagesState):
    response = model.invoke(state["messages"])
    return {"messages": response}
builder = StateGraph(MessagesState)
builder.add_node(call_model)
builder.add_edge(START, "call_model")
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {
    "configurable": {
        "thread_id": "1"
    }
}
for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
    config,
    stream_mode="values",
):
    chunk["messages"][-1].pretty_print()
for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "what's my name?"}]},
    config,
    stream_mode="values",
):
    chunk["messages"][-1].pretty_print()

输出如下所示：​​​​​​​

================================ Human Message =================================
hi! I'm bob
================================== Ai Message ==================================
Hi Bob! How are you doing today? Is there anything I can help you with?
================================ Human Message =================================
what's my name?
================================== Ai Message ==================================
Your name is Bob.

4.1 生产环境

在生产环境中，通常需要使用一个数据库作为checkpointer：

下面我们将MongoDB 作为 checkpointer 进行演示。

pip install -U pymongo langgraph langgraph-checkpoint-mongodb

from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.mongodb import MongoDBSaver
model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
DB_URI = "localhost:27017"
with MongoDBSaver.from_conn_string(DB_URI) as checkpointer:
    def call_model(state: MessagesState):
        response = model.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": "1"
        }
    }
    for chunk in graph.stream(
        {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
        config,
        stream_mode="values"
    ):
        chunk["messages"][-1].pretty_print()
    for chunk in graph.stream(
        {"messages": [{"role": "user", "content": "what's my name?"}]},
        config,
        stream_mode="values"
    ):
        chunk["messages"][-1].pretty_print()

4.2 使用subgraphs

       如果你的图包含子图 ，编译父图时只需提供检查点，LangGraph 会自动将checkpointer传播到子子图。​​​​​​​

from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import InMemorySaver
from typing import TypedDict
class State(TypedDict):
    foo: str
# Subgraph
def subgraph_node_1(state: State):
    return {"foo": state["foo"] + "bar"}
subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph = subgraph_builder.compile()
# Parent graph
def node_1(state: State):
    return {"foo": "hi! " + state["foo"]}
builder = StateGraph(State)
builder.add_node("node_1", subgraph)
builder.add_edge(START, "node_1")
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

4.3 管理checkpointer

可以查看并删除检查点存储的信息：

• 查看线程状态（检查点）

config = {
    "configurable": {
        "thread_id": "1",
        # optionally provide an ID for a specific checkpoint,
        # otherwise the latest checkpoint is shown
        # "checkpoint_id": "1f029ca3-1f5b-6704-8004-820c16b69a5a"
    }
}
graph.get_state(config)

      输出结果如下所示：​​​​​​​

StateSnapshot(
    values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today?), HumanMessage(content="what's my name?"), AIMessage(content='Your name is Bob.')]}, next=(), 
    config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1f5b-6704-8004-820c16b69a5a'}},
    metadata={
        'source': 'loop',
        'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}},
        'step': 4,
        'parents': {},
        'thread_id': '1'
    },
    created_at='2025-05-05T16:01:24.680462+00:00',
    parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}}, 
    tasks=(),
    interrupts=()
)

• 查看该线程的历史信息（检查点）

config = {
    "configurable": {
        "thread_id": "1"
    }
}
list(graph.get_state_history(config))

       输出结果如下所示：​​​​​​​

[
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content="what's my name?"), AIMessage(content='Your name is Bob.')]}, 
        next=(), 
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1f5b-6704-8004-820c16b69a5a'}}, 
        metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Your name is Bob.')}}, 'step': 4, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:24.680462+00:00',
        parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},
        tasks=(),
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?'), HumanMessage(content="what's my name?")]}, 
        next=('call_model',), 
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-1790-6b0a-8003-baf965b6a38f'}},
        metadata={'source': 'loop', 'writes': None, 'step': 3, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:23.863421+00:00',
        parent_config={...}
        tasks=(PregelTask(id='8ab4155e-6b15-b885-9ce5-bed69a2c305c', name='call_model', path=('__pregel_pull', 'call_model'), error=None, interrupts=(), state=None, result={'messages': AIMessage(content='Your name is Bob.')}),),
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}, 
        next=('__start__',), 
        config={...}, 
        metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': "what's my name?"}]}}, 'step': 2, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:23.863173+00:00',
        parent_config={...}
        tasks=(PregelTask(id='24ba39d6-6db1-4c9b-f4c5-682aeaf38dcd', name='__start__', path=('__pregel_pull', '__start__'), error=None, interrupts=(), state=None, result={'messages': [{'role': 'user', 'content': "what's my name?"}]}),),
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob"), AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')]}, 
        next=(), 
        config={...}, 
        metadata={'source': 'loop', 'writes': {'call_model': {'messages': AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')}}, 'step': 1, 'parents': {}, 'thread_id': '1'},
        created_at='2025-05-05T16:01:23.862295+00:00',
        parent_config={...}
        tasks=(),
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': [HumanMessage(content="hi! I'm bob")]}, 
        next=('call_model',), 
        config={...}, 
        metadata={'source': 'loop', 'writes': None, 'step': 0, 'parents': {}, 'thread_id': '1'}, 
        created_at='2025-05-05T16:01:22.278960+00:00', 
        parent_config={...}
        tasks=(PregelTask(id='8cbd75e0-3720-b056-04f7-71ac805140a0', name='call_model', path=('__pregel_pull', 'call_model'), error=None, interrupts=(), state=None, result={'messages': AIMessage(content='Hi Bob! How are you doing today? Is there anything I can help you with?')}),), 
        interrupts=()
    ),
    StateSnapshot(
        values={'messages': []}, 
        next=('__start__',), 
        config={'configurable': {'thread_id': '1', 'checkpoint_ns': '', 'checkpoint_id': '1f029ca3-0870-6ce2-bfff-1f3f14c3e565'}},
        metadata={'source': 'input', 'writes': {'__start__': {'messages': [{'role': 'user', 'content': "hi! I'm bob"}]}}, 'step': -1, 'parents': {}, 'thread_id': '1'}, 
        created_at='2025-05-05T16:01:22.277497+00:00', 
        parent_config=None,
        tasks=(PregelTask(id='d458367b-8265-812c-18e2-33001d199ce6', name='__start__', path=('__pregel_pull', '__start__'), error=None, interrupts=(), state=None, result={'messages': [{'role': 'user', 'content': "hi! I'm bob"}]}),), 
        interrupts=()
    )
]

删除该线程的所有检查点​​​​​​​

thread_id = "1"
checkpointer.delete_thread(thread_id)

五、增加长期记忆

       使用长期记忆（跨线程持久性）来存储用户特定或应用特定的数据。这对像聊天机器人这样的应用非常有用，因为你需要记住用户偏好或其他信息。

       为了使用长期记忆，我们在创建图时需要提供一个存储器 ：​​​​​​​

import uuid
from typing_extensions import Annotated, TypedDict
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore
model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
def call_model(
    state: MessagesState,
    config: RunnableConfig,
    *,
    store: BaseStore,  
):
    user_id = config["configurable"]["user_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 "remember" in last_message.content.lower():
        memory = "User name is Bob"
        store.put(namespace, str(uuid.uuid4()), {"data": memory})
    response = model.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")
checkpointer = InMemorySaver()
store = InMemoryStore()
graph = builder.compile(
    checkpointer=checkpointer,
    store=store,
)

config = {
    "configurable": {
        "thread_id": "1",
        "user_id": "1",
    }
}
for chunk in graph.stream(
    {"messages": [{"role": "user", "content": "Hi! Remember: my name is Bob"}]},
    config,
    stream_mode="values",
):
    chunk["messages"][-1].pretty_print()
config = {
    "configurable": {
        "thread_id": "2",
        "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()

       输出结果如下所示：​​​​​​​

================================ Human Message =================================
Hi! Remember: my name is Bob
================================== Ai Message ==================================
Hi Bob! I'll remember that your name is Bob. How are you doing today?
================================ Human Message =================================
what is my name?
================================== Ai Message ==================================
Your name is Bob.

5.1 生产环境

     在生产环境中，你需要使用一个数据库的checkpointer，我们将使用 Postgres store进行演示：

     第一次使用 Postgres store时需要调用 store.setup（）

pip install -U "psycopg[binary,pool]" langgraph langgraph-checkpoint-postgres

from langchain_core.runnables import RunnableConfig
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore
from langgraph.store.base import BaseStore
model = init_chat_model(model="anthropic:claude-3-5-haiku-latest")
DB_URI = "postgresql://postgres:postgres@localhost:5442/postgres?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"]
        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 "remember" in last_message.content.lower():
            memory = "User name is Bob"
            store.put(namespace, str(uuid.uuid4()), {"data": memory})
        response = model.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": "Hi! Remember: my name is Bob"}]},
        config,
        stream_mode="values",
    ):
        chunk["messages"][-1].pretty_print()
    config = {
        "configurable": {
            "thread_id": "2",
            "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()

5.2 语义搜索

     可以在图的记忆存储中启用语义搜索：这可以让图智能体通过语义相似度搜索存储中的项。​​​​​​​

from typing import Optional
from langchain.embeddings import init_embeddings
from langchain.chat_models import init_chat_model
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langgraph.graph import START, MessagesState, StateGraph
llm = init_chat_model("openai:gpt-4o-mini")
# Create store with semantic search enabled
embeddings = init_embeddings("openai:text-embedding-3-small")
store = InMemoryStore(
    index={
        "embed": embeddings,
        "dims": 1536,
    }
)
store.put(("user_123", "memories"), "1", {"text": "I love pizza"})
store.put(("user_123", "memories"), "2", {"text": "I am a plumber"})
def chat(state, *, store: BaseStore):
    # Search based on user's last message
    items = store.search(
        ("user_123", "memories"), query=state["messages"][-1].content, limit=2
    )
    memories = "\n".join(item.value["text"] for item in items)
    memories = f"## Memories of user\n{memories}" if memories else ""
    response = llm.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": "I'm hungry"}]},
    stream_mode="messages",
):
    print(message.content, end="")

      在设计智能体时，可以从以下角度考虑：

• 智能体应该学习什么类型的内容 ：事实/知识？过去事件的总结？规则和风格？
• 记忆应该在什么时候形成 （以及由谁来形成记忆）
• 记忆应该存储在哪里 ？（在提示词里？语义存储？）。这在很大程度上决定了他们的召回方式。

六、构建电子邮件智能体（详细教程）

       通过将这三种长期记忆类型交织在一起，我们的智能体变得真正聪明且个性化。想象一下一位助理：

• 通知：某客户的邮件如果24小时内未回复，通常需要后续处理；
• 了解你的写作风格和语气，调整正式回复以应对外部沟通，调整轻松回复以适应团队成员；
• 能记住复杂的项目背景，而无需你反复解释；
• 它更擅长预测你想看到哪些邮件，哪些邮件会自动处理。

工作流程如下所示：

START → Triage (Episodic + Procedural Memory) → Decision → Response Agent (Semantic + Procedural Memory) → END

       我正在运营“知识星球”，在那里，我会分享一些更前沿的技术和干货，感兴趣的朋友可以加入，一些经典的代码，我也会放到那里，感谢支持。

最终实现的效果如下所示：

本示例完整的代码，请参考：https://t.zsxq.com/BnARP

七、结论

       记忆是AI agents的基石，使它们能够保留、回忆并适应信息。我们探讨了短期和长期记忆——程序性、情节性和语义性——如何融入AI agents堆栈，超越了单纯的上下文窗口或 RAG。通过构建内存增强的电子邮件代理，我们展示了这些概念的实际应用，从状态管理到语义搜索和自适应指令。掌握内存赋能人工智能提供个性化、情境感知的解决方案，为生产环境中更智能、更强大的系统铺平道路。