AI Agent记忆系统实现：从短期记忆到长期记忆的完整指南！

本文详细介绍了AI Agent记忆系统的重要性与实现方法，区分了上下文窗口与记忆的本质差异，解析了短期记忆与长期记忆（程序性、情节性、语义性）的架构与应用。通过代码示例展示了如何使用LangGraph实现记忆功能，并提供了生产环境部署方案，最终通过电子邮件智能体案例展示了记忆系统如何提升AI的个性化和适应能力。

功城师

619人浏览 · 2025-12-25 08:45:00

功城师 · 2025-12-25 08:45:00 发布

简介

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

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

一、引言

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 帮助智能体更好地回答。而记忆则帮助智能体行为更聪明。

一直在更新，更多的大模型学习和面试资料已经上传带到CSDN的官方了，有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇

在这里插入图片描述

二、智能体中的记忆类型

一般来说，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。通过构建内存增强的电子邮件代理，我们展示了这些概念的实际应用，从状态管理到语义搜索和自适应指令。掌握内存赋能人工智能提供个性化、情境感知的解决方案，为生产环境中更智能、更强大的系统铺平道路。

八、如何系统的学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

一直在更新，更多的大模型学习和面试资料已经上传带到CSDN的官方了，有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇

在这里插入图片描述