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

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。

classState(MessagesState):current_user_text:strassistant_text:strmemories:list[dict[str,Any]]saved_memories:list[dict[str,Any]]

几个关键字段：

一条消息的流转：

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",]returnany(patterninloweredforpatterninpatterns)

例子：

I am Denial

转成小写后：

i am Denial

命中：

i am

所以保存。

写入 JSON：

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

对应节点：

defwrite_memory(state:State):current_text=state.get("current_user_text")ifnotcurrent_textornot_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)ifmemory_textandmemory_textinquery:score+=2returnscore

关键是这一句：

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

检索节点：

defretrieve_memory(state:State):current_message=_latest_human_message(state)query=_message_content(current_message)ifcurrent_messageelse""all_memories=_read_memories()ranked=sorted(all_memories,key=lambdamemory:_score_memory(query,memory),reverse=True,)relevant=[mforminrankedif_score_memory(query,m)>0]ifnotrelevant:relevant=all_memories[-5:]return{"memories":relevant[:5],"current_user_text":query,}

一句话：

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

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

模型不会自己读 JSON。

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

defcall_model(state:State):current_text=state.get("current_user_text")memories=state.get("memories",[])memory_context="\n".join(f"-{memory['text']}"formemoryinmemories)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 就不是黑盒。