AI Agent 开发(23):Agent 记忆系统

QuibblerAgentQuibblerAgent 2026-10-25 约 23 分钟 3 次阅读

第 12 篇讲 checkpoint 内部结构时埋过一句话:"第 23 篇的长期记忆走 Store 跨线程共享,与线程内的 checkpoint 互补。"现在兑现。checkpointer 加 thread_id 解决了会话内记忆,但它有一个天然边界:记忆锁死在单个线程里。用户今天在线程 A 里说"我对花生过敏",明天开个新线程问"推荐午餐",Agent 一脸茫然——checkpointer 不会跨线程去翻别的线程状态,这个问题它天生解决不了。

本篇是《AI Agent 开发》系列第 23 篇,也是第五模块"进阶架构"的开篇,代码基于 langgraph 1.2.x。全文按"记忆分类学 → 两大流派 → Store 最小接入 → 写入策略 → 完整闭环 → 踩坑 → 压缩与摘要"推进,示例沿用第 12 篇的聊天图底座。

1、记忆分类学

和认知科学对齐,官方Memory 概念文档把 Agent 记忆先按作用域分两层:会话内的短期记忆、跨会话的长期记忆;长期记忆再按内容形态分三类。全貌一棵树:

Agent 记忆
├── 短期记忆 —— 会话内,checkpointer 托管(第 12 篇)
│     消息历史与 State 快照,按 thread_id 隔离
└── 长期记忆 —— 跨会话,Store 托管(本篇)
      ├── 语义记忆:事实与偏好("她对花生过敏")
      ├── 情景记忆:过往事件("上周一起修过部署脚本")
      └── 程序性记忆:操作流程("输出代码前先跑测试")

四类记忆在工程上的对应关系一张表说清:

类型 存什么 LangGraph 载体
短期 本轮会话的消息历史与 State checkpointer
语义 用户事实与偏好 Store
情景 过往对话与任务的经过 Store(常为摘要形态)
程序性 Agent 学到的行为规则 Store(注入提示词)

只有短期记忆属于 checkpointer,其余三类都落在 Store——后三类共用同一套 put/search API,靠 namespace 分拣。多数 Agent 先做语义记忆就够用,情景与程序性按需追加;分类学是设计地图,不是起步清单。

2、长期记忆两大流派

让 Agent 拥有跨会话知识,业界就两条路。

检索式:对话中抽取事实存进外部库,来日对话前检索、拼进提示词。记忆与模型分离,store.put 一行写入,下一次对话立即可见。参数式:微调,把知识直接烧进模型权重,问答时无需检索。两条路的成本差距是结构性的:

维度 检索式 参数式(微调)
记忆载体 外部库 + 提示词注入 模型权重
更新生效 写入即可见 收集数据 → 训练 → 评估 → 部署全流程
删除一条记忆 store.delete 一行 做不到,需重训
上下文占用 每轮注入消耗 token 零
适合内容 用户偏好、动态事实 稳定领域知识、输出风格

参数式的致命伤在第三行:用户要求"忘掉我提供过的信息",权重里烧进去的东西取不出来。工程默认检索式——记忆的增删改查是运维问题,不该升级成训练问题;参数式留给真正稳定的知识与风格,那已经不是"记忆"而是"能力"。本篇后续全部围绕检索式展开。

3、Store 最小接入

Store 是 LangGraph 的跨线程持久层,统一实现 BaseStore 接口,与 checkpointer 在 compile 处会师。两者的分工一张图看懂:

                Agent 图(单例)
               /               \
    compile(checkpointer=…)    compile(store=…)
             │                    │
     会话内:按 thread_id        跨会话:按 namespace
     ├─ thread-1 ● 快照          ├─ ("alice", "memories")
     ├─ thread-2 ● 快照          ├─ ("bob", "memories")
     └─ thread-3 ● 快照          └─ ("alice", "episodes")
     线程之间互相隔离            线程之间按 namespace 共享

解读:同一个图实例同时挂两个持久层,checkpointer 记"这个线程聊到哪了",Store 记"这个用户是谁"——记忆版图由此拼完整。Store 的最小 API 四件套如下:

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
ns = ("alice", "memories")  # namespace:字符串元组,可任意分层

store.put(ns, "mem-1", {"text": "用户对花生过敏"})  # value 必须是 dict
item = store.get(ns, "mem-1")  # 精确读取一条
print(item.value["text"])
items = store.search(ns, limit=10)  # 列出 namespace 内条目
store.delete(ns, "mem-1")  # 删除

解读:put 收三样——namespace、key、value。namespace 像目录路径,("alice", "memories") 就是"alice 的 memories 目录",不同用户各写各的目录、互不可见;key 是条目身份证,对同一个 (namespace, key) 再次 put 是覆盖更新而不是新增——第 6 节"记忆冲突"的解药就藏在这。value 必须是 dict,传纯字符串直接报错。开发期用 InMemoryStore(重启即失,定位同 InMemorySaver),生产换数据库后端的 Store 实现,接口不变,详见 Stores 文档。

条目多起来后,全量列出不够用,search 要按语义检索。给 Store 配一个 embedding 索引:

from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore

store = InMemoryStore(
    index={
        "embed": init_embeddings("openai:text-embedding-3-small"),
        "dims": 1536,  # 维度须与 embedding 模型一致
        "fields": ["text"],  # 只对 text 字段做嵌入
    }
)

# 配了 index 后,带 query 的 search 走语义检索
hits = store.search(ns, query="饮食忌口", limit=3)

解读:index 配置让 Store 在每次 put 时对指定字段自动嵌入,search 带 query 就按相似度返回 top-k。没配 index 时带 query 调用会报错,这是新手高频坑。语义检索与第 14 篇的向量库同源,细节见 InMemoryStore 参考与官方语义检索公告。

4、记忆写入策略

图挂上了 Store,下一个设计决策:谁来决定"记什么"?两条路线。

4.1、模型自主式

把"保存记忆"做成一个工具交给模型,提示词里写明标准,模型在对话中自行判断何时调用。memory-agent 官方仓库就是这个思路:

import uuid

from langchain_core.tools import tool
from langgraph.config import get_store

@tool
def save_memory(memory: str) -> str:
    """记住用户透露的长期事实或偏好,供未来会话使用。"""
    store = get_store()
    store.put(("alice", "memories"), str(uuid.uuid4()), {"text": memory})
    return f"已记住:{memory}"

解读:工具函数不在节点签名里,拿 Store 靠 get_store()——它从运行时上下文取出编译期挂上的那个实例,图内任何代码(节点、工具、task)都能这么取,见 get_store 参考。返回值把记下的内容原样回显,让用户看得见 Agent 记了什么——第 6 节会讲这也是防记忆污染的关键一环。

4.2、规则式抽取

不信模型的自觉,就写死规则:每轮对话后追加一个抽取节点,用便宜的模型从最近消息里抽事实。核心实现:

def extract_memories(state: MessagesState, store: BaseStore):
    prompt = (
        "从下面的对话里抽取值得长期记住的用户事实,"
        "每条一行,没有则输出空行:\n"
        + "\n".join(m.text for m in state["messages"][-2:])
    )
    result = small_model.invoke(prompt)
    for line in result.text.splitlines():
        if line.strip():
            store.put(("alice", "memories"),
                      str(uuid.uuid4()), {"text": line.strip()})
    return {}  # 不改 State,只写记忆

解读:节点签名里声明 store: BaseStore 参数,LangGraph 自动注入——这是节点内访问 Store 的第二种方式,与 get_store() 等价。返回 {} 表示纯副作用节点,用一条边串在模型节点后面即可。两种策略对照:

维度 工具式(模型自主) 规则式(每轮后处理)
触发时机 模型判断,需要时才记 每轮必跑
额外成本 零,工具调用走主模型 每轮固定多一次抽取调用
漏记风险 有,取决于提示词 低
噪声风险 低 高,需去重过滤

对话型 Agent 优先工具式——用户在语境里的强调程度本身就是"值不值得记"的信号,模型拿得到这个信号;规则式适合无对话循环的流水线,或用户要求全量沉淀的场景。

5、偏好记忆完整闭环

把前两节拼成完整闭环:节点每次调用前检索记忆拼进 system prompt(注入),模型对话中调工具写入(沉淀),checkpointer 与 Store 并存。先看注入节点:

from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage
from langgraph.graph import MessagesState
from langgraph.store.base import BaseStore

model = init_chat_model("openai:gpt-5").bind_tools([save_memory])

def call_model(state: MessagesState, store: BaseStore):
    items = store.search(("alice", "memories"), limit=10)
    facts = "\n".join(f"- {i.value['text']}" for i in items)
    system = SystemMessage(
        content=f"已知用户信息:\n{facts}\n"
                "发现新的长期事实或偏好时,调用 save_memory 记住。"
    )
    return {"messages": [model.invoke([system] + state["messages"])]}

解读:每次模型调用前现场检索、现场拼接,记忆更新后下一轮立即生效;save_memory 沿用 4.1 的定义。装配与第 9 篇手写 ReAct 同构,只是 compile 多挂一个 store:

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.prebuilt import ToolNode
from langgraph.store.memory import InMemoryStore

def route(state: MessagesState):
    return "tools" if state["messages"][-1].tool_calls else END

builder = StateGraph(MessagesState)
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode([save_memory]))
builder.add_edge(START, "model")
builder.add_conditional_edges("model", route)
builder.add_edge("tools", "model")

store = InMemoryStore()
graph = builder.compile(checkpointer=InMemorySaver(), store=store)

解读:checkpointer 管线程内历史,store 管跨线程记忆,互不干扰。验证闭环是否真的跨会话——开两个线程:

c1 = {"configurable": {"thread_id": "chat-1"}}
graph.invoke({"messages": [{"role": "user", "content": "我对花生过敏,帮我记住。"}]}, c1)

c2 = {"configurable": {"thread_id": "chat-2"}}
r = graph.invoke({"messages": [{"role": "user", "content": "推荐一道午餐。"}]}, c2)
print(r["messages"][-1].text)

解读:第一轮模型调 save_memory 把过敏写进 Store;第二轮全新线程,checkpointer 历史为零,但注入节点 search 拿到了那条记忆,推荐会避开花生——跨线程记忆闭环生效。把 alice 换成真实用户 ID、namespace 第一段跟着走,多用户隔离就有了;记忆过百条后把全量列出换成第 3 节的语义检索,取 top-3 即可。

6、记忆系统的坑

闭环跑通只是及格线,长期记忆是要长期运营的数据资产,四个坑提前避开。

**过期与冲突。**用户三月说"我住北京",十月说"我搬到上海了",堆积式写入会让两条并存,检索命中哪条看运气。解法是更新而非堆积:写入前先 search 同主题条目,命中就复用它的 key 再 put——同 key 写入即覆盖;再给 value 加 updated_at 字段,注入时带上时间,让模型知道哪条更新。

**隐私。**记忆是用户数据的长期沉淀,法规层面(如 GDPR 的被遗忘权)要求可查看、可删除。好在 get、search、delete 就是现成的后端——产品上给用户一个"查看我的记忆、删除某条"的入口,成本极低;健康、财务类敏感信息要么单独 namespace 加权限,要么干脆不入库。

**记忆污染。**模型幻觉出的"事实"被工具固化,之后每轮当真话注入,错误自我强化——这是长期记忆独有的失败模式。缓解手段:抽取提示词写死"只记用户明确陈述的内容";工具返回值回显给用户,当场可纠正;重要决策让模型复述记忆来源,而不是直接照办。

**检索漂移。**记忆条数上千后,top-k 里混进无关项,注入的噪声反而稀释提示词。对策三板斧:limit 给小,3~5 条足够;namespace 按主题细分,饮食偏好与工作安排分开存;定期清理长期零命中的条目——记忆库也要做 housekeeping。

7、记忆压缩与摘要

记忆版图还剩最后一块:会话本身变长之后怎么办。两个方向,一个治短期、一个治长期。

短期侧,第 10 篇埋的伏笔正面兑现——SummarizationMiddleware。历史超过阈值时自动把旧消息摘要成一条系统消息,近期消息保留原文:

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model,
    tools=[save_memory],
    middleware=[
        SummarizationMiddleware(
            model="openai:gpt-5-mini",  # 摘要交给便宜的小模型
            trigger=("tokens", 4000),  # 历史超过 4000 token 触发
            messages_to_keep=5,  # 摘要后保留最近 5 条原文
        )
    ],
    checkpointer=InMemorySaver(),
)

解读:上下文占用从随轮次线性增长压到近似常数,且对图结构零侵入。trigger 用元组形式 ("tokens", N),旧的 max_tokens_before_summary 参数已弃用;中间件全家族见中间件文档。

长期侧,把快结束的线程整段摘要成"情景记忆"存进 Store——第 12 篇的 get_state 正好取得到完整历史:

def archive_thread(graph, config, store):
    state = graph.get_state(config)
    dialog = "\n".join(m.text for m in state.values["messages"])
    summary = model.invoke("把这段对话摘要成要点:\n" + dialog)
    store.put(("alice", "episodes"),
              config["configurable"]["thread_id"],
              {"summary": summary.text})

解读:归档进 ("alice", "episodes") 后,未来任何会话检索这个 namespace,都能拿到"上周一起修过部署脚本"这类背景——正是第 1 节说的情景记忆。与 SummarizationMiddleware 的分工:前者压缩当下线程的窗口,后者把经验沉淀给未来的线程,一急一缓。

8、总结

回收全文:记忆分四类,短期归 checkpointer,语义、情景、程序性三类归 Store;实现选检索式而非参数式,更新成本与可删除性差一个量级;Store 的组织轴是 namespace,put、get、search、delete 四件套,同 key 写入即更新;写入策略工具式优先、规则式兜底;闭环 = 启动检索注入 + 工具写入沉淀 + 两个持久层并存;运营上盯住过期冲突、隐私、污染、漂移四个坑;会话变长就摘要——短期用 SummarizationMiddleware,长期把线程归档成情景记忆。

留一个练习:回到第 20 篇的 SQLite 查询 Server,给它加一张可写的 memories 表和 save_memory、search_memory 两个工具(只读连接之外单开一个可写连接),接到 Claude Code 或 Claude Desktop,让任意 Host 拥有跨会话记忆——本质上是用 MCP 工具重做一遍本篇的 Store。加分项:写入前先 SELECT 查重,命中即 UPDATE,把"更新而非堆积"做进 SQL。

对于任何要长期服务用户的 Agent,建议第一天就把 Store 挂上:记忆是用户回头的起点,金鱼 Agent 留不住人。下一篇进入多 Agent 架构:Supervisor、handoff 与层级团队。

相关推荐

Hermes Agent 详解:自进化学习型 AI 代理
Agent

Hermes Agent 详解:自进化学习型 AI 代理

Hermes Agent 详解:自进化学习型 AI 代理Hermes Agent 是由 Nous Research 于 2026 年 2 月开源的自主 AI 代理框架,其核心定位是"与你共同成长的智能体"。与传统的 IDE 编程助手或单一 API 聊天机器人不同,Hermes 运行在你的服务器上,能够记住所学内容,运行时间越长能力越强。它采用 MIT 开源许可,支持 Linux、macOS 和 W

393
Hermes Agent:与你一起成长的开源 AI 代理
Agent

Hermes Agent:与你一起成长的开源 AI 代理

Hermes Agent:与你一起成长的开源 AI 代理大多数 AI 助手是"健忘的租客"——换个窗口就失忆、换个平台就重学、干过的活不留下任何沉淀。Nous Research 出品的 Hermes Agent 反其道而行,口号是"The Agent That Grows With You"(与你一起成长的代理):它是一个开源(MIT)、自托管的 AI 代理,具备持久记忆(学习你的项目、自动生成技

322
speech-to-speech 详解:HuggingFace 的开源语音 Agent 流水线
AI

speech-to-speech 详解:HuggingFace 的开源语音 Agent 流水线

speech-to-speech 详解:HuggingFace 的开源语音 Agent 流水线语音 Agent 的"全开源"路线一直有个尴尬:框架开源了,模型却离不开云端 API。huggingface/speech-to-speech 把这条路走通了:VAD → STT → LLM → TTS 四段低延迟流水线,每个组件都可换成开源模型本地跑——Apple Silicon 一台 Mac 就能撑起

264