LLM之Agent（四十）｜AI Agents（九）：Agentic Memory介绍

推荐阅读列表：

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