Agent Memory 的本质：写入、检索、注入

Agent 不是天然会记住你。

所谓 Memory，本质就是三件事：

1. 写入策略：什么信息值得保存？
2. 检索策略：当前问题需要哪些记忆？
3. 注入策略：怎么把记忆塞给模型？

我们用一个本地 json_memory.json 版本，把这三件事看清楚。

---

1. 整体流程

对应代码：

builder.add_node("retrieve_memory", retrieve_memory)
builder.add_node("assistant", call_model)
builder.add_node("write_memory", write_memory)

builder.add_edge(START, "retrieve_memory")
builder.add_edge("retrieve_memory", "assistant")
builder.add_edge("assistant", "write_memory")
builder.add_edge("write_memory", END)

流程很简单：

先读记忆 -> 再问模型 -> 最后写记忆

---

2. State 是运行时账本

LangGraph 的核心是维护 State。

class State(MessagesState):
    current_user_text: str
    assistant_text: str
    memories: list[dict[str, Any]]
    saved_memories: list[dict[str, Any]]

几个关键字段：

字段	作用
messages	对话消息
current_user_text	本轮用户输入
memories	检索出来的历史记忆
assistant_text	模型回复
saved_memories	本轮新保存的记忆

一条消息的流转：

---

3. 写入策略：什么值得记？

不是每句话都要保存。

我们用关键字判断：

def _should_save_user_fact(text: str) -> bool:
    lowered = text.lower()
    patterns = [
        "remember",
        "my name is",
        "i am",
        "i like",
        "i dislike",
        "i prefer",
        "my goal is",
    ]
    return any(pattern in lowered for pattern in patterns)

例子：

I am Denial

转成小写后：

i am Denial

命中：

i am

所以保存。

写入 JSON：

{
  "text": "I am Denial",
  "source": "json_memory_graph",
  "assistant_response": "Nice to meet you, Denial!"
}

对应节点：

def write_memory(state: State):
    current_text = state.get("current_user_text")

    if not current_text or not _should_save_user_fact(current_text):
        return {"saved_memories": []}

    memory = {
        "text": current_text.strip(),
        "assistant_response": state.get("assistant_text"),
    }

    memories = _read_memories()
    memories.append(memory)
    _write_memories(memories)

    return {"saved_memories": [memory]}

一句话：

写入策略 = 判断这句话有没有长期价值。

---

4. 检索策略：现在该想起什么？

用户问问题时，不是把所有记忆都塞给模型，而是先找相关记忆。

我们的简化版用“词交集”打分：

def _score_memory(query: str, memory: dict[str, Any]) -> int:
    query_terms = _terms(query)
    memory_text = str(memory.get("text", ""))
    memory_terms = _terms(memory_text)

    score = len(query_terms & memory_terms)

    if memory_text and memory_text in query:
        score += 2

    return score

关键是这一句：

query_terms & memory_terms

它表示两个集合的交集。

例子：

query:  What is my goal?
memory: My goal is to earn 50k per month

拆词：

query_terms  = {what, is, my, goal}
memory_terms = {my, goal, is, to, earn, 50k, per, month}

交集：

{my, goal, is}

分数：

score = 3

检索节点：

def retrieve_memory(state: State):
    current_message = _latest_human_message(state)
    query = _message_content(current_message) if current_message else ""
    all_memories = _read_memories()

    ranked = sorted(
        all_memories,
        key=lambda memory: _score_memory(query, memory),
        reverse=True,
    )

    relevant = [m for m in ranked if _score_memory(query, m) > 0]
    if not relevant:
        relevant = all_memories[-5:]

    return {
        "memories": relevant[:5],
        "current_user_text": query,
    }

一句话：

检索策略 = 从历史记忆里找当前最相关的几条。

---

5. 注入策略：怎么让模型知道？

模型不会自己读 JSON。

所以要把检索出的记忆写进 prompt。

def call_model(state: State):
    current_text = state.get("current_user_text")
    memories = state.get("memories", [])

    memory_context = "\n".join(
        f"- {memory['text']}" for memory in memories
    )

    system_message = SystemMessage(
        content=(
            "Use the long-term memories only when relevant.\n\n"
            f"Long-term memories:\n{memory_context}"
        )
    )

    response = llm.invoke([
        system_message,
        HumanMessage(content=current_text),
    ])

    return {
        "messages": [response],
        "assistant_text": response.content,
    }

如果 JSON 里有：

I am Denial

用户问：

What is my name?

实际给模型的是：

System:
Long-term memories:
- I am Denial

Human:
What is my name?

一句话：

注入策略 = 把检索出来的记忆放进模型上下文。

---

6. 一次完整运行

用户输入：

I am Denial

LangGraph API 接收到：

{
  "input": {
    "messages": [
      {
        "role": "user",
        "content": "I am Denial"
      }
    ]
  }
}

节点流转：

retrieve_memory
  读 messages
  得到 current_user_text = I am Denial
  返回 memories

assistant
  读取 current_user_text
  读取 memories
  调用 LLM
  返回 assistant_text

write_memory
  读取 current_user_text
  判断是否命中 i am
  写入 json_memory.json
  返回 saved_memories

最终：

json_memory.json 里多了一条：
I am Denial

---

7. 一个关键工程经验

一开始我们让 write_memory 自己从 messages 里找用户输入。

后来发现不稳定。

更稳的做法是：

retrieve_memory 负责提取用户输入
显式写入 state.current_user_text
后续节点统一读 current_user_text

也就是：

关键中间结果，不要让后续节点重复猜。
显式放进 State。

---

总结

Agent Memory 可以拆成三个问题：

1. 什么值得记？       -> 写入策略
2. 现在该想起什么？   -> 检索策略
3. 怎么告诉模型？     -> 注入策略

LangGraph 做的事是：

维护 State
执行节点
合并节点返回
控制流程流转

所以调试 Agent 时，只盯住三句话：

这个节点读了哪些 State？
这个节点返回了哪些 State？
下一个节点拿到的 State 变成了什么？

看清这条链路，Memory 就不是黑盒。