从零开始的LLM 9. A-MEM 与 LangChain/LangGraph学习

前言

本文将介绍 A-MEM 的核心原理、API 使用方法，以及如何基于 LangChain 构建一个具备记忆能力的 Agent。

经过前面的学习，我们已经系统地完成了大语言模型（LLM）的基础部分，主要参考了以下资料：

• 《动手学深度学习》
• 《蘑菇书》
• 《Happy-LLM》
• 《Hello-Agents》
• 《all-in-rag》

到这里，可以认为已经具备了继续深入学习 Agent 的基础能力。

接下来的内容将不再像之前那样按照固定课程循序渐进，而是更多围绕当前 Agent 领域的研究方向展开。学习内容既会涉及工程实践（如各种 Agent 框架、工作流、记忆系统等），也会涉及算法与模型层面的探索（如多模态、持续学习等）。

1. Generative Agents
2. Letta
3. A-MEM
4. Voyager

这条路线主要围绕 Agent 的持续学习（Continual Learning） 展开，也是我目前最感兴趣的方向：如何让一个 Agent 在长期运行过程中不断积累经验、提升能力，而不是每次都从零开始。

目前来看，大致有两种思路：

1. 不修改模型本身。 仅通过 Prompt、记忆系统（Memory）、RAG、工具调用等机制，使 Agent 能够不断积累知识和经验，实现能力的持续增长。
2. 修改模型本身。 基于 Agent 长期运行过程中积累的对话、笔记和总结等数据，在合适的时机对模型进行持续训练（Continual Fine-tuning），使模型真正将这些经验内化，而不仅仅依赖外部记忆。

由于这一阶段更多是阅读论文、分析开源项目、验证实验以及整理个人理解，而不是单纯跟着教程学习，因此资料收集、实验和整理都会比之前花费更多时间。

---

本章节主要是对 A-MEM 的学习和总结。A-MEM 是一个挺有创意的设计，我认为「卡片化记忆」这个思路本身非常不错，但真正将这个想法落地，实现成一个可以运行的系统，又是另一件很厉害的事情。当然，A-MEM 目前也存在一些问题，还有不少可以优化的空间。

在学习 A-MEM 的过程中，我也重新熟悉了一遍如何使用 LangChain 构造 Agent。因此，本文不会把重点全部放在 A-MEM 的理论介绍上，而是会结合实际代码实现，探索如何构建一个带有 Memory 能力的 Agent。

除此之外，在调研 Memory 方案的过程中，我发现目前使用较多的方案之一是 mem0。一个方案能够被大量使用，通常都有其存在的原因。因此，后续在学习 Voyager 之前，我会先对 mem0 的设计和实现进行学习，了解它为什么能够得到较广泛的应用。

---

1. A-MEM原理

论文链接：https://arxiv.org/abs/2502.12110

A-MEM（Agentic Memory）的整体结构并不复杂，其核心思想是：

将每次交互转换为结构化 memory note，并通过记忆之间的关联和动态更新，使长期记忆逐渐形成一个可演化的记忆网络。

1.1 Note 结构（Structured Memory Notes）

对于每一个记忆 \(m_i\)，A-MEM都会构建一个结构化 memory note： \[ m_i=\{c_i,t_i,K_i,G_i,X_i,e_i,L_i\} \] 其中：

• \(c_i\)：对话内容，即用户的问题和 LLM 的回答；
• \(t_i\)：该次交互发生的时间戳；
• \(L_i\)：关联记忆列表，保存与当前记忆存在语义关系的其他 memory id。

除了原始的交互内容和时间信息外，A-MEM还会通过 LLM 对当前记忆进行进一步分析，生成额外的结构化信息：

\[ K_i,G_i,X_i \leftarrow \text{LLM}(c_i||t_i||P_{s1}) \] 其中：

• \(K_i\)：关键词，用于表示当前记忆中的核心概念；
• \(G_i\)：标签，用于对记忆进行分类；
• \(X_i\)：内容描述，对当前记忆提供更加丰富的语义信息。

简单来说：

关键词、tag、描述 = LLM(对话内容 + 时间戳)

相比直接保存原始对话，这种方式能够提取出更适合检索的高层语义信息。

之后，将原始对话内容以及生成的关键词、标签、描述进行拼接，并通过 Encoder 转换成向量： \[ e_i=f_{enc}[\text{concat}(c_i,K_i,G_i,X_i)] \] 其中：

• \(e_i\) 是该 memory note 的向量表示；

它融合了原始交互内容以及 LLM 提取出的结构化信息，后续的 memory 检索和关联发现都会基于该向量进行。

1.2 记忆关联生成（Link Generation）

当新的记忆 \(m_n\) 加入系统时，A-MEM首先会寻找与它可能相关的已有记忆。

首先，通过 embedding （\(e_i\) ）计算新记忆与已有记忆之间的相似度： \[ s_{n,j}=\frac{e_n * e_j}{|e_n|*|e_j|} \] 取出其中最相似的k个记忆 \(m_j\) ，他们组成集合 \(M^n_\text{near}\) \[ M^n_\text{near}=\{m_j|\text{rank}(s_{n,j})\le k,m_j \in M\} \] 这些记忆作为候选关联对象。

随后，将新记忆 \(m_n\) 和候选记忆集合 \(M^n_{near}\) 输入 LLM，让模型进一步判断它们之间是否存在语义关系，如果 LLM 判断两个 memory 之间存在关联，则会将对应 memory id 添加到当前 memory 的 \(L_n\) 中。： \[ L_n \leftarrow \text{LLM}(m_n||M^n_\text{near}||P_{s2}) \] 整体流程：

新memory
    ↓
Embedding相似度检索
    ↓
获取候选memory
    ↓
LLM判断关系
    ↓
建立memory link

这里可以看到：

• Embedding 负责快速筛选可能相关的 memory；
• LLM 负责进一步理解语义关系并决定是否建立连接。

相比单纯的向量检索，memory link 可以让不同记忆之间形成显式关系。

1.3 记忆演化（Memory Evolution）

除了建立新的关联，A-MEM还会根据新产生的 memory 对已有相关 memory 进行更新。

对于候选集合 \(M^n_{near}\) 中的每一个记忆 \(m_j\)： \[ m^*_j \leftarrow \text{LLM}(m_n||m_j||P_{s3}) \]

LLM 会分析新旧 memory 之间的关系，并更新已有 memory 的内容，包括：

• 内容描述；
• 关键词；
• 标签。

例如：

已有 memory：用户正在学习机器学习。

新增 memory：用户最近开始学习 PyTorch。

经过更新后，memory 可以变成：用户正在学习机器学习，并开始深入学习 PyTorch。

---

这里我认为也是 A-MEM 一个可以进一步改进的地方。

目前 A-MEM主要采用更新机制，即通过新信息优化已有 memory，但是没有进一步处理：

• 两条 memory 高度重复时，是否应该合并；
• 新旧 memory 发生冲突时，如何判断哪个信息更加可靠；
• 长期无效或者过时的 memory 是否应该删除。

因此，可以进一步加入：

memory merge（记忆合并）
memory conflict resolution（冲突解决）
memory forgetting（记忆遗忘）

等机制，使长期记忆管理更加完善。

1.4 相关记忆检索（Retrieve Relative Memory）

在每次 LLM 回复之前，A-MEM都会执行一次 memory retrieval。

具体流程如下：

用户问题
    ↓
Embedding编码
    ↓
向量相似度搜索
    ↓
获取Top-k相关memory
    ↓
加入Prompt
    ↓
LLM生成回答

具体来说：

1. 将用户当前问题转换成向量；
2. 与已有 memory 的 embedding 计算相似度；
3. 获取最相关的 k 条 memory；
4. 将检索到的 memory 与用户问题一起输入 LLM。

1.5 总结

A-MEM的核心并不是提出一种新的向量检索方法，而是重新设计了长期 memory 的组织方式。

A-MEM 与传统的历史聊天记录存储方案（例如 Letta 的 recall memory）最大的区别在于：

传统方案更接近“历史聊天记录数据库 + 向量检索”，而 A-MEM 更关注如何将历史交互转换为可组织、可关联、可演化的长期记忆。

传统 recall memory 的流程：

历史聊天记录
      ↓
Embedding编码
      ↓
向量检索
      ↓
找到相关历史对话
      ↓
加入Prompt
      ↓
LLM生成回答

这种方式本质上是对过去对话的检索。当用户再次提到相关内容时，系统通过相似度找到之前的聊天记录，并将其提供给 LLM。

而 A-MEM 的流程：

用户交互
      ↓
生成结构化memory note
      ↓
建立memory link
      ↓
更新已有memory
      ↓
形成memory网络
      ↓
检索相关memory
      ↓
LLM生成回答

A-MEM 不只是保存“过去发生了什么”，而是进一步提取：

• 关键词；
• 标签；
• 内容描述；
• 记忆之间的关联关系。

因此，memory 不再是一堆独立的历史记录，而是逐渐形成一个具有结构和关系的知识网络。

这里要提一点。可能初学者不太区分长期记忆（Long-term Memory）和短期记忆（Short-term Memory）。

A-MEM主要针对的是长期记忆，因此采用类似 RAG 的方式，通过向量检索获取与当前问题相关的历史 memory 是合理的。

而短期记忆主要指当前会话中的上下文（多轮对话历史）。由于这些信息通常距离当前问题较近，不需要额外进行复杂的检索，可以直接根据上下文窗口大小进行裁剪，并放入 Prompt 中。

2. LangChain 和 LangGraph 学习

首先需要说明LangChain 和 LangGraph 两者之间的关系。

简单来说：

• LangChain 主要负责提供构建 LLM 应用的基础组件。
• LangGraph 主要负责定义这些组件之间的执行流程。

二者通常配合使用。

可以简单理解：

LangChain 提供 Agent 的各种“零件”，LangGraph 负责把这些零件按照一定流程组织起来。

2.1 LangChain

LangChain 是一个用于构建 LLM 应用的开发框架，主要提供各种基础能力，例如：

定义模型：

from langchain_ollama import ChatOllama
llm = ChatOllama(model="qwen2.5:7b", temperature=0)

调用模型：

llm.invoke(messages)

定义工具：

@tool
def search_memory(query:str):
    ...

绑定工具：

llm.bind_tools(tools)

还有其他向量库等各种零件工具。

2.2 LangGraph

LangGraph 主要负责 Agent 的流程编排。

它提供：

• State（全局状态）
• Node（节点）
• Edge（节点连接）
• Conditional Edge（条件分支）

2.2.1 LangChain 的消息系统

LangChain 使用统一的消息抽象：

BaseMessage

BaseMessage 是所有消息类型的基类。

常见消息：

消息类型	作用
HumanMessage	用户输入
AIMessage	LLM输出
SystemMessage	系统提示
ToolMessage	工具执行结果

一次完整 Agent 流程如下。

---

a. 用户输入

用户：

帮我查北京天气

转换为：

HumanMessage(
    content="帮我查北京天气"
)

此时状态：

messages=[
    HumanMessage("帮我查北京天气")
]

---

b. LLM 判断调用工具

LLM 根据当前绑定的工具，决定调用天气查询工具：

AIMessage(
    content="",
    tool_calls=[
        {
            "name":"weather",
            "args":{
                "city":"北京"
            }
        }
    ]
)

注意：

这里的 AIMessage 不是最终回答。

它表示：

LLM 请求执行 weather 工具。

此时：

messages=[
    HumanMessage("帮我查北京天气"),

    AIMessage(
        tool_calls=[
            weather("北京")
        ]
    )
]

---

c. ToolNode 执行工具

ToolNode 读取：

AIMessage.tool_calls

找到对应工具并执行。

工具返回：

ToolMessage(
    content="北京今天晴，25度",
    tool_call_id="xxx"
)

状态：

messages=[
    HumanMessage(...),

    AIMessage(
        tool_calls=[weather]
    ),

    ToolMessage(
        content="北京今天晴，25度"
    )
]

---

d. LLM 生成最终答案

LLM 根据工具返回结果继续生成：

AIMessage(
    content="北京今天晴，气温25度。"
)

最终：

messages=[
    HumanMessage,

    AIMessage(tool_call),

    ToolMessage,

    AIMessage(answer)
]

2.2.2 全局状态（State）

LangGraph 中，所有节点共享一个状态对象（State）。

例如：

class AgentState(TypedDict):
    messages: Sequence[BaseMessage]

这里使用 Python 的 TypedDict 定义 State 的结构，即 State 本质上是一个字典，其中 messages 字段保存由多个 BaseMessage 对象组成的消息序列，例如：

{
    "messages": [
        HumanMessage(...),
        AIMessage(...)
    ]
}

如果希望 LangGraph 在更新 messages 时自动合并新旧消息，而不是直接覆盖，可以使用 Annotated 指定 reducer：

from typing import Annotated, Sequence, TypedDict
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
    messages: Annotated[
        Sequence[BaseMessage],
        add_message
    ]

其中，Annotated 为 messages 附加了 add_messages 这一更新规则。也就是说，当节点返回新的 messages 时，不会直接覆盖原有消息，而是由 add_messages 将新旧消息进行合并，从而保留之前的消息。例如原来有 [HumanMessage]，节点返回 [AIMessage]，经过 add_messages 后会变成 [HumanMessage, AIMessage]。

我们也可以自定义 reducer，例如只保留最近 50 条消息：

def limited_messages(
    old: Sequence[BaseMessage],
    new: Sequence[BaseMessage]
):
    messages = list(old) + list(new)
    return messages[-50:]

class AgentState(TypedDict):
    messages: Annotated[
        Sequence[BaseMessage],
        limited_messages
    ]

---

定义好 State ，然后就是最开始的一步，创建图

# ---------------------------------------------------------------------------
# 5. 构建 LangGraph 图
# ---------------------------------------------------------------------------
graph = StateGraph(AgentState)

2.2.3 节点函数

LangGraph 中的 Node 本质上就是一个普通 Python 函数。

节点函数主要负责：

1. 接收当前全局状态（State）；
2. 根据任务进行处理；
3. 返回需要更新的状态字段。

def retrieve_context(state: AgentState) -> AgentState:
    ...
    system_note = SystemMessage(
        content=f"以下是与当前问题相关的历史记忆，可参考但不必照搬:\n{context_text}"
    )
    return {"messages": [system_note]}

这里的：

return {"messages": [system_note]}

并不是直接修改全局状态，而是告诉 LangGraph：

当前节点需要更新 messages 字段。

LangGraph 会将节点的返回值识别为 State Update，然后根据 AgentState 中定义的 reducer 处理新旧状态，最后更新全局 State，并将更新后的 State 传递给下一节点。

节点也可以直接通过 state["messages"] 修改状态，例如：

def retrieve_context(state: AgentState):
    state["messages"].append(system_note)
    return state

但通常不推荐这样做。

LangGraph 的设计是让节点通过 return 返回 State Update，再由 LangGraph 统一处理状态更新。这样可以：

• 由 reducer 统一决定新旧状态如何合并；
• 避免节点直接修改共享 State；
• 使节点的输入、输出更加明确，便于维护和调试。

因此，更推荐：

return {"messages": [system_note]}

而不是直接修改：

state["messages"].append(system_note)

另外，节点不需要返回完整的 AgentState，只需要返回需要更新的字段即可。

---

定义节点函数后，需要将其添加到 LangGraph 图中：

graph.add_node(
    "retrieve_context",
    retrieve_context
)

其中：

• "retrieve_context"：节点名称，用于后续定义边（Edge）时引用；
• retrieve_context：实际执行的 Python 函数。

2.2.4 流程控制（Edge）

Edge 用于定义 LangGraph 中节点之间的执行关系，可以简单理解为：

指定一个节点执行完成后，下一步进入哪个节点。

首先定义入口节点，注意这里填写的是节点名称：

graph.set_entry_point("retrieve_context")

---

普通边

添加普通单向边：

graph.add_edge("retrieve_context", "agent")

即 retrieve_context 节点执行完成后，进入 agent 节点。

---

条件边（Conditional Edge）

除了固定流向外，LangGraph 也支持根据条件动态选择下一节点：

graph.add_conditional_edges(
    "agent",
    tools_condition,
    {"tools": "tools", "__end__": "persist"},
)

其中：

• "agent"：表示从哪个节点开始判断；
• tools_condition：条件判断函数；
• 字典：定义条件函数返回值对应的目标节点。

条件函数的返回值必须是字符串，并且该字符串需要存在于映射表中。

例如，条件函数执行完后返回如下字符：

return "tools"

会根据 {"tools": "tools", "__end__": "persist"}, 进入tool节点。

2.2.5 工具处理（Tool）

LangGraph 提供了预构建工具节点：

from langgraph.prebuilt import ToolNode, tools_condition

其中：

• ToolNode：负责执行工具调用；
• tools_condition：负责判断是否需要调用工具。

---

定义工具

首先通过 LangChain 的 @tool 装饰器定义工具：

@tool
def save_memory(content: str, tags: str = "", category: str = "General") -> str:
    """把一条重要信息主动写入长期记忆(A-MEM)。tags 用逗号分隔，如 "项目,决策"。"""
	...
    return f"已写入记忆，memory_id={memory_id}"


@tool
def search_memory(query: str, k: int = 5) -> str:
    """在长期记忆(A-MEM)中检索与 query 语义相关的历史笔记，k 为返回条数。"""
	...
    return "\n".join(lines)

@tool 会根据函数名称、参数类型、描述信息等自动生成 Tool Schema，使 LLM 能够理解：

• 工具名称；
• 工具作用；
• 参数格式。

然后将工具保存：

tools = [save_memory, search_memory]

---

绑定工具到 LLM

LLM 通过 bind_tools() 将工具信息提供给模型：

llm = ChatOllama(
    model="qwen2.5:7b",
    temperature=0
).bind_tools(tools)

此时工具的名称、描述、参数结构等信息会被 LangChain 添加到模型输入中。

在真正发送给模型前，LangChain 会根据模型对应的 Chat Template，将消息和工具定义转换成模型能够理解的格式。

现代 LLM 通常通过特殊 token 和结构化消息格式处理工具调用，例如：

模型通过这些信息知道：

• 当前有哪些工具可以调用；
• 每个工具的作用；
• 工具需要哪些参数；
• 如何生成符合格式的 tool_calls。

---

创建工具节点

通过：

graph.add_node(
    "tools",
    ToolNode(tools)
)

添加工具节点。

ToolNode(tools) 可以理解为：

一个专门执行 LLM 工具调用请求的节点。

它主要负责：

1. 获取 LLM 输出中的 tool_calls；
2. 根据工具名称找到对应函数；
3. 执行工具；
4. 将结果转换为 ToolMessage。

例如 LLM 输出：

AIMessage(
    tool_calls=[
        {
            "name": "search_memory",
            "args": {
                "query": "项目资料"
            }
        }
    ]
)

ToolNode 接收到后，会转换并执行：

search_memory(query="项目资料")

然后返回：

ToolMessage(
    content="检索结果..."
)

---

tools_condition 判断工具调用

在 Agent 流程中，LLM 节点执行完成后，需要判断下一步应该：

• 调用工具；
• 还是直接生成最终答案。

from langgraph.prebuilt import ToolNode, tools_condition
# tools_condition: 若 LLM 触发了 tool_calls -> "tools"，否则 -> "__end__"
graph.add_conditional_edges(
    "agent",
    tools_condition,
    {"tools": "tools", "__end__": "persist"},
)

tools_condition 的作用：

检查最后一条 AIMessage 是否包含 tool_calls，并根据结果返回对应的路由名称。

---

如果 LLM 决定调用工具：

AIMessage(
    content="",
    tool_calls=[
        {
            "name": "search_memory",
            "args": {
                "query": "项目资料"
            }
        }
    ]
)

此时：

tools_condition(state)

返回：

"tools"

---

如果 LLM 不需要调用工具，直接生成回答：

AIMessage(
    content="答案是..."
)

由于不存在tool_calls，因此：

tools_condition(state)

返回：

"__end__"

进入结束流程。

需要注意，tools_condition 不会执行工具，它只负责判断下一步流程。

实际执行工具的是：

ToolNode(tools)

2.3 Graph应用运行与会话管理

在完成 Graph 构建后，主程序主要负责 初始化 Graph 应用、配置会话状态、处理用户交互以及执行 Agent 流程。

2.3.1 配置Graph状态持久化

首先，需要为 Graph 添加状态保存能力：

from langgraph.checkpoint.memory import MemorySaver

app = graph.compile(
    checkpointer=MemorySaver()
)

注意：如果不配置 checkpointer=MemorySaver()，Graph 执行过程中的状态不会被保存，每次调用 invoke() 都会从初始状态开始，无法保留历史对话。

---

checkpointer机制

checkpointer 是 LangGraph 中用于保存和恢复 Graph 状态（state）的组件，可以理解为 Graph 的"存档系统"。

在 LangGraph 中，节点之间通过 state 传递数据，例如：

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]

其中 messages 保存当前会话产生的消息。

如果没有配置 checkpointer，这些状态只存在于当前一次 Graph 执行过程中。当 invoke() 结束后，状态不会被保存，下一次调用无法继续之前的会话。

配置 checkpointer 后，LangGraph 会在执行过程中保存当前 state 的 checkpoint，并与 thread_id 关联。当下一次使用相同 thread_id 调用 Graph 时，会自动恢复对应的 state。

---

常见checkpointer实现

LangGraph提供了多种状态存储方式：

实现	说明	适用场景
MemorySaver	保存在内存中，程序退出后丢失	本地开发、调试
SqliteSaver	保存到SQLite文件	单机应用、小规模持久化
PostgresSaver	保存到PostgreSQL	生产环境、多实例部署

---

thread_id机制

checkpointer 需要通过唯一标识区分不同会话，这个标识就是：

config["configurable"]["thread_id"]

这里的 "thread_id" 是 LangGraph checkpoint 机制规定的固定字段，而不是自定义名称。

checkpointer 在保存和读取 state 时，会从：

config["configurable"]

中获取：

thread_id

并将其作为当前会话状态的索引。

因此：

• 相同 thread_id：恢复之前保存的状态；
• 不同 thread_id：创建新的独立状态。

2.3.2 会话模式设计

基于 thread_id，可以实现两种不同的会话模式。

---

模式一：固定thread_id（开启短期记忆）

固定使用一个 session：

thread_id = FIXED_THREAD_ID

此时，之前的消息会一直保存在 messages 中。

例如：

第一次调用：

messages=[
    HumanMessage("你好"),
    AIMessage("你好")
]

第二次调用：

messages=[
    HumanMessage("你好"),
    AIMessage("你好"),
    HumanMessage("介绍一下自己")
]

新的消息会追加到已有消息列表中，并作为当前 Graph state 继续传递。

这种模式适用于：

• 多轮聊天机器人；
• 需要上下文理解的 Agent；
• 普通对话场景。

---

模式二：每轮新thread_id（关闭短期记忆）

每次生成新的：

thread_id = uuid.uuid4().hex

此时每轮对话都是独立 session，不会携带之前的 messages。

这种方式适合测试长期记忆能力：

• 不依赖短期上下文；
• 通过 Memory / RAG 检索历史信息；
• 验证 Agent 是否真正使用长期记忆。

---

实际代码中，可以通过开关控制两种模式：

while True:
    user_input = input("你: ").strip()
    if user_input.lower() in ("exit", "quit", "退出"):
        break

    # 根据开关生成 thread_id
    if USE_PERSISTENT_THREAD:
        thread_id = FIXED_THREAD_ID
    else:
        thread_id = uuid.uuid4().hex   # 每次随机生成新 ID

随后生成 Graph 配置：

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

2.3.3 Graph执行流程

在调用 Graph 前，需要先获取当前 thread 已保存的消息。

这里主要用于：

• 判断本轮新增消息；
• 打印 Agent 执行过程中的工具调用和返回结果。

代码：

# 记录本轮开始前的消息 ID（用于打印新增内容）
try:
    prev_snapshot = app.get_state(config)
    if prev_snapshot.values:
        prev_messages = prev_snapshot.values.get("messages", [])
    else:
        prev_messages = []

    prev_ids = {m.id for m in prev_messages}
except Exception:
    # 如果 thread_id 从未使用过，get_state 可能报错，视为空历史
    prev_ids = set()

然后将用户输入传入 Graph：

result = app.invoke(
    {"messages": [HumanMessage(content=user_input)]},
    config=config,
)

result 是 这一次 invoke() 执行结束后返回的最终 state（Graph状态快照）。

result["messages"]中包含整个 Graph 执行过程产生的消息，包括：

• 用户消息；
• AI 回复；
• 工具调用消息；
• 工具执行结果。

2.3.4 打印新增消息

由于：

result["messages"]

包含历史消息，因此需要过滤出本轮新增内容：

new_messages = [m for m in result["messages"] if m.id not in prev_ids]

然后根据消息类型进行打印：

# 打印本轮新增消息（工具调用等）
for m in new_messages:
    role = type(m).__name__
    if isinstance(m, AIMessage) and m.tool_calls:
        for call in m.tool_calls:
            print(f"[{role}] 调用工具: {call['name']}({call['args']})")
    elif isinstance(m, ToolMessage):
        print(f"[{role}] 工具结果: {m.content}")
    else:
        print(f"[{role}] {m.content}")
print()

例如：

[HumanMessage] 查询天气

[AIMessage] 调用工具:
weather_api({'city':'Tokyo'})

[ToolMessage] 东京今天晴天

[AIMessage] 今天东京天气晴朗

代码中 role = type(m).__name__ 用于获取消息类型名称。

m = AIMessage(content="hello")
type(m)          # → <class 'langchain_core.messages.ai.AIMessage'>  （这是类对象，不是字符串）
type(m).__name__ # → "AIMessage"  （这才是字符串，类名本身）

3. A-MEM 使用以及 Agent 制作

在第 2 章节中，已经介绍了 Agent 的整体实现流程。本章节主要补充 A-MEM 的实际使用方式，以及如何将其接入 Agent 工作流。

A-MEM 官方仓库： https://github.com/agiresearch/A-mem

安装:

git clone https://github.com/agiresearch/A-mem.git
cd A-mem && pip install .

安装完成后导入：

from agentic_memory.memory_system import AgenticMemorySystem

虽然前面介绍了 A-MEM 的核心思想，包括记忆生成、记忆关联以及检索机制，但实际使用时，A-MEM 已经通过 API 封装好了大部分细节，我们主要关注以下几个接口：

• 初始化记忆系统；
• 添加记忆；
• 检索记忆。

3.1 初始化 A-MEM

首先创建 A-MEM 记忆系统：

memory_system = AgenticMemorySystem(
    model_name="paraphrase-multilingual-MiniLM-L12-v2",
    llm_backend="ollama",
    llm_model="qwen2.5:7b",
)

其中：

model_name

• 用于 ChromaDB 向量检索的 embedding 模型；
• 不是 LLM，仅负责文本向量化；
• 默认使用 sentence-transformers 加载。

llm_backend

• 指定生成记忆时调用的 LLM 后端。

llm_model

• 指定具体使用的模型名称，需要与 Ollama 中已经下载的模型名称保持一致。

第一次运行时 paraphrase-multilingual-MiniLM-L12-v2 会通过 sentence-transformers 自动从 HuggingFace Hub 下载模型权重，并缓存到：~/.cache/huggingface。该模型大小约几十 MB。

另外需要注意，默认情况下 A-MEM 使用的是内存版 ChromaRetriever，并没有启用持久化存储。

因此：

• 程序运行期间，记忆可以正常保存和检索；
• 程序退出后，记忆数据会丢失。

3.2 保存记忆

A-MEM 提供 add_note() 接口用于向记忆库中添加新的记忆。调用方式：

memory_system.add_note(
    content=content,
    tags=tag_list,
    category=category
)

主要参数：

• content：记忆内容。
• tags：标签信息。
• category：记忆类别。

在 Agent 中，可以封装成 Tool，让 LLM 根据上下文主动决定是否保存：

@tool
def save_memory(content: str, tags: str = "", category: str = "General") -> str:
    """把一条重要信息主动写入长期记忆(A-MEM)。tags 用逗号分隔，如 "项目,决策"。"""
    tag_list = [t.strip() for t in tags.split(",") if t.strip()]
    memory_id = memory_system.add_note(content=content, tags=tag_list, category=category)
    return f"已写入记忆，memory_id={memory_id}"

3.3 检索记忆

A-MEM 使用 search_agentic() 接口检索相关记忆。

调用方式：

results = memory_system.search_agentic(query, k=k)

其中 query 是当前查询内容，k 是初始检索数量。

同样，将其封装成 Agent Tool：

@tool
def search_memory(query: str, k: int = 5) -> str:
    """在长期记忆(A-MEM)中检索与 query 语义相关的历史笔记，k 为返回条数。"""
    results = memory_system.search_agentic(query, k=k)
    if not results:
        return "未检索到相关记忆。"
    lines = [f"- [{r['id']}] {r['content']} (tags={r.get('tags')})" for r in results]
    return "\n".join(lines)

3.4 search_agentic 源码分析

按照论文中的设计，A-MEM 的检索过程应该是：

首先通过 embedding 相似度找到当前问题相关的 Top-K 记忆，然后根据这些记忆之间建立的 link 关系继续扩展，将关联记忆加入结果集合。

因此理论上：

search_agentic 返回的结果不应该只有 k 条，而应该包含初始检索得到的 k 条记忆，以及这些记忆关联的其他记忆。

但是查看实际源码后发现，当前实现与论文描述存在一定差异。

源码首先通过 ChromaDB 获取相似度最高的 k 条记忆：

results = self.retriever.search(query, k)

然后将这些记忆加入：

memories.append(memory_dict)

之后遍历这些记忆，根据：

links = memory.get('links', [])

获取关联记忆最多k条，并继续加入：

memories.append(...)

最后：

return memories[:k]

直接截取前 k 条返回。

因此实际流程是：

1. 根据向量相似度获取 k 条候选记忆；
2. 尝试添加关联记忆；
3. 最终只返回前 k 条。

这意味着：

• 如果返回数量不足 k，可能包含部分关联记忆；
• 如果记忆数量足够，前 k 条通常仍然是最初的相似度检索结果；
• 后续添加的关联记忆大概率会因为 [:k] 被截断。

所以当前版本的 search_agentic() 实际效果更接近：

基于向量相似度检索，并尝试补充关联记忆，但最终返回数量仍受到 k 限制。

def search_agentic(self, query: str, k: int = 5) -> List[Dict[str, Any]]:
    """Search for memories using ChromaDB retrieval."""
    if not self.memories:
        return []
        
    try:
        # Get results from ChromaDB
        results = self.retriever.search(query, k)
       ...
            
        # Process ChromaDB results
        for i, doc_id in enumerate(results['ids'][0][:k]):
            ...
                # Create result dictionary with all metadata fields
                memory_dict = {
                    'id': doc_id,
                    'content': metadata.get('content', ''),
                    'context': metadata.get('context', ''),
                    'keywords': metadata.get('keywords', []),
                    'tags': metadata.get('tags', []),
                    'timestamp': metadata.get('timestamp', ''),
                    'category': metadata.get('category', 'Uncategorized'),
                    'is_neighbor': False
                }
                ...
                memories.append(memory_dict)
                seen_ids.add(doc_id)
        
        # Add linked memories (neighbors)
        neighbor_count = 0
        for memory in list(memories):  # Use a copy to avoid modification during iteration
            if neighbor_count >= k:
                break

            # Get links from metadata
            links = memory.get('links', [])
            ...
            for link_id in links:
                if link_id not in seen_ids and neighbor_count < k:
                    neighbor = self.memories.get(link_id)
                    if neighbor:
                        memories.append({
                            'id': link_id,
                            'content': neighbor.content,
                            'context': neighbor.context,
                            'keywords': neighbor.keywords,
                            'tags': neighbor.tags,
                            'timestamp': neighbor.timestamp,
                            'category': neighbor.category,
                            'is_neighbor': True
                        })
                        seen_ids.add(link_id)
                        neighbor_count += 1
        
        return memories[:k]

3.5 Agent 接入 A-MEM

在 Agent 流程中，我们会在每次对话开始前，进行一次长期记忆的读取，通过当前问题检索 A-MEM 中相关的历史记忆，并将检索结果加入上下文。 在每次对话结束后，我们会将本轮对话的内容保存到记忆库中，形成新的长期记忆。

3.5.1 对话开始前读取长期记忆

在 LangGraph 中增加一个节点，用于在 Agent 执行前检索相关记忆：

def retrieve_context(state: AgentState) -> AgentState:
    last_user_msg = None
    for m in reversed(state["messages"]):
        if isinstance(m, HumanMessage):
            last_user_msg = m.content
            break

    if not last_user_msg:
        return {}                     # 不是 return state
    # LangGraph 里空更新表示"这个 channel 不变"

    results = memory_system.search_agentic(last_user_msg, k=5)
    if not results:
        return {}                     # 不是 return state

    context_text = "\n".join(f"- {r['content']}" for r in results)
    system_note = SystemMessage(
        content=f"以下是与当前问题相关的历史记忆，可参考但不必照搬:\n{context_text}"
    )
    return {"messages": [system_note]}

该节点主要完成两件事情：

1. 获取当前对话中的用户输入；
2. 调用 A-MEM 检索相关长期记忆，并作为 SystemMessage 添加到当前消息列表中。

如果当前节点没有需要修改的 State，只需要返回空字典即可。

3.5.2 对话结束后保存长期记忆

在 Agent 完成回答后，将本轮对话保存到 A-MEM：

def persist_turn(state: AgentState) -> AgentState:
    last_ai = state["messages"][-1]
    if isinstance(last_ai, AIMessage) and not getattr(last_ai, "tool_calls", None):
        last_user_msg = ""
        for m in reversed(state["messages"]):
            if isinstance(m, HumanMessage):
                last_user_msg = m.content
                break
        memory_system.add_note(
            content=f"Q: {last_user_msg}\nA: {last_ai.content}",
            tags=["conversation"],
            category="Dialogue",
        )
    return {}                          # 不是 return state

其他的节点构造，以及完整的节点连接等，可以参考以下完整agent代码：

"""
A-MEM + LangGraph Agent
=======================
集成 A-MEM (Agentic Memory) 作为长期记忆系统的 LangGraph agent 示例。

A-MEM 仓库: https://github.com/agiresearch/A-mem
安装:
    git clone https://github.com/agiresearch/A-mem.git
    cd A-mem && pip install .

依赖:
    pip install langgraph langchain-openai langchain-core

"""

import os
from typing import Annotated, Sequence, TypedDict

from langchain_core.messages import (
    AIMessage,
    BaseMessage,
    HumanMessage,
    SystemMessage, ToolMessage,
)
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

from langgraph.graph import END, StateGraph
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

# A-MEM 官方 API：https://github.com/agiresearch/A-mem
from agentic_memory.memory_system import AgenticMemorySystem


# ---------------------------------------------------------------------------
# 1. 初始化 A-MEM 记忆系统
# ---------------------------------------------------------------------------

memory_system = AgenticMemorySystem(
    model_name="paraphrase-multilingual-MiniLM-L12-v2",  # ChromaDB 用的 embedding 模型（本地，非LLM，不用改）
    llm_backend="ollama",            # openai -> ollama
    llm_model="qwen2.5:7b",          # gpt-4o-mini -> qwen2.5:7b（要跟 ollama pull 的名字一致）
)
# "paraphrase-multilingual-MiniLM-L12-v2"
# 第一次用的时候，sentence-transformers 库自动从 HuggingFace Hub 下载一次模型权重（几十 MB，缓存到 ~/.cache/huggingface）
# 用的默认就是纯内存版 ChromaRetriever，根本没有走持久化路径，程序一退出数据就没了。


# ---------------------------------------------------------------------------
# 2. 把 A-MEM 的读写能力包装成 LangChain Tool，供 agent 在推理中主动调用
# ---------------------------------------------------------------------------
@tool
def save_memory(content: str, tags: str = "", category: str = "General") -> str:
    """把一条重要信息主动写入长期记忆(A-MEM)。tags 用逗号分隔，如 "项目,决策"。"""
    tag_list = [t.strip() for t in tags.split(",") if t.strip()]
    memory_id = memory_system.add_note(content=content, tags=tag_list, category=category)
    return f"已写入记忆，memory_id={memory_id}"


@tool
def search_memory(query: str, k: int = 5) -> str:
    """在长期记忆(A-MEM)中检索与 query 语义相关的历史笔记，k 为返回条数。"""
    results = memory_system.search_agentic(query, k=k)
    if not results:
        return "未检索到相关记忆。"
    lines = [f"- [{r['id']}] {r['content']} (tags={r.get('tags')})" for r in results]
    return "\n".join(lines)


tools = [save_memory, search_memory]
# llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(tools)

from langchain_ollama import ChatOllama

llm = ChatOllama(model="qwen2.5:7b", temperature=0).bind_tools(tools)



# ---------------------------------------------------------------------------
# 3. LangGraph 状态定义
# ---------------------------------------------------------------------------
# class AgentState(TypedDict):
#     messages: Annotated[Sequence[BaseMessage], add_messages]
#

def limited_messages(
    old: Sequence[BaseMessage],
    new: Sequence[BaseMessage]
):
    messages = list(old) + list(new)

    return messages[-50:]


class AgentState(TypedDict):
    messages: Annotated[
        Sequence[BaseMessage],
        limited_messages
    ]



# ---------------------------------------------------------------------------
# 4. 节点定义
# ---------------------------------------------------------------------------
def retrieve_context(state: AgentState) -> AgentState:
    last_user_msg = None
    for m in reversed(state["messages"]):
        if isinstance(m, HumanMessage):
            last_user_msg = m.content
            break

    if not last_user_msg:
        return {}                     # 不是 return state
    # LangGraph 里空更新表示"这个 channel 不变"

    results = memory_system.search_agentic(last_user_msg, k=5)
    if not results:
        return {}                     # 不是 return state

    context_text = "\n".join(f"- {r['content']}" for r in results)
    system_note = SystemMessage(
        content=f"以下是与当前问题相关的历史记忆，可参考但不必照搬:\n{context_text}"
    )
    return {"messages": [system_note]}



def call_model(state: AgentState) -> AgentState:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}


def persist_turn(state: AgentState) -> AgentState:
    last_ai = state["messages"][-1]
    if isinstance(last_ai, AIMessage) and not getattr(last_ai, "tool_calls", None):
        last_user_msg = ""
        for m in reversed(state["messages"]):
            if isinstance(m, HumanMessage):
                last_user_msg = m.content
                break
        memory_system.add_note(
            content=f"Q: {last_user_msg}\nA: {last_ai.content}",
            tags=["conversation"],
            category="Dialogue",
        )
    return {}                          # 不是 return state


# ---------------------------------------------------------------------------
# 5. 构建 LangGraph 图
# ---------------------------------------------------------------------------
graph = StateGraph(AgentState)

graph.add_node("retrieve_context", retrieve_context)
graph.add_node("agent", call_model)
graph.add_node("tools", ToolNode(tools))
graph.add_node("persist", persist_turn)

graph.set_entry_point("retrieve_context")
graph.add_edge("retrieve_context", "agent")

# tools_condition: 若 LLM 触发了 tool_calls -> "tools"，否则 -> "__end__"
graph.add_conditional_edges(
    "agent",
    tools_condition,
    {"tools": "tools", "__end__": "persist"},
)


graph.add_edge("tools", "agent")
graph.add_edge("persist", END)


from langgraph.checkpoint.memory import MemorySaver

app = graph.compile(checkpointer=MemorySaver())



# ---------------------------------------------------------------------------
# 6. 运行示例
# ---------------------------------------------------------------------------

import uuid  # 确保在文件顶部已导入

if __name__ == "__main__":
    # ============ 你可以在下面两行修改 ============
    # USE_PERSISTENT_THREAD = True   # True: 使用固定 thread_id，False: 每次新 thread_id
    USE_PERSISTENT_THREAD = False

    FIXED_THREAD_ID = "cli-session-1"
    # =============================================

    # print("Agent 已启动，记忆文件: my_agent.amem")
    print("输入 'exit' 或 'quit' 退出。\n")

    while True:
        user_input = input("你: ").strip()
        if user_input.lower() in ("exit", "quit", "退出"):
            break

        # 根据开关生成 thread_id
        if USE_PERSISTENT_THREAD:
            thread_id = FIXED_THREAD_ID
        else:
            thread_id = uuid.uuid4().hex   # 每次随机生成新 ID

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

        # 记录本轮开始前的消息 ID（用于打印新增内容）
        try:
            prev_snapshot = app.get_state(config)
            if prev_snapshot.values:
                prev_messages = prev_snapshot.values.get("messages", [])
            else:
                prev_messages = []

            prev_ids = {m.id for m in prev_messages}
        except Exception:
            # 如果 thread_id 从未使用过，get_state 可能报错，视为空历史
            prev_ids = set()

        result = app.invoke(
            {"messages": [HumanMessage(content=user_input)]},
            config=config,
        )

        # 打印本轮新增消息（工具调用等）
        new_messages = [m for m in result["messages"] if m.id not in prev_ids]
        for m in new_messages:
            role = type(m).__name__
            if isinstance(m, AIMessage) and m.tool_calls:
                for call in m.tool_calls:
                    print(f"[{role}] 调用工具: {call['name']}({call['args']})")
            elif isinstance(m, ToolMessage):
                print(f"[{role}] 工具结果: {m.content}")
            else:
                print(f"[{role}] {m.content}")
        print()

4. 总结

虽然当前 A-MEM 提供的 API 实现仍然存在一些不足，例如检索逻辑与论文描述存在一定差异、默认存储方式不适合长期使用等问题，但其整体设计思想仍然具有较高的参考价值。

其中，基于卡片化的记忆表示方式、记忆之间的关联关系以及通过 Agent 主动管理长期记忆的思路，都为构建具有长期记忆能力的 Agent 提供了一种较好的设计方向。

如果需要将 A-MEM 应用于实际生产环境，可能需要根据具体业务需求重新设计和实现对应的记忆管理模块，包括记忆存储、检索策略、关联关系维护以及生命周期管理等部分，而不是直接依赖当前提供的 API。

总体来说，A-MEM 更像是一个验证长期记忆 Agent 思路的研究实现，其核心价值在于提出了一种较新的记忆组织方式，而实际工程落地仍需要进一步优化。