• 欢迎关注我的微信公众号“ Falost ” 右边扫描关注 --->>

Agent 记忆系统实战:短期记忆、长期记忆与知识图谱(架构+代码拆解)

AI / 大模型 神棍 103℃ 0评论

Agent 不能只有脑子,还得有记忆

上篇聊完 Agent 编排框架选型,我在本地搭了一套 LangGraph + CrewAI 的混合 demo。跑起来之后冒出一个很现实的问题——每轮对话结束,Agent 就「失忆」了。我跟它聊了十分钟系统架构,它记住了所有技术选型,但关掉终端再打开,它又像一个陌生人一样问我「你好,有什么可以帮你的」。

这就很尴尬了。

翻了一下 LangGraph 和 CrewAI 的文档,发现这两套框架对记忆的支持比我想象的成熟。这期就来拆解 Agent 记忆系统的三层架构:短期记忆(对话上下文)、长期记忆(跨会话持久化)和知识图谱(结构化实体关系)。

一、Agent 记忆的三种类型

先明确一个概念框架。我把它理解成三张层层递进的网:

记忆类型 作用范围 存储位置 典型实现
短期记忆 单次会话内 Prompt 上下文窗口 LangGraph Checkpointer
长期记忆 跨会话/跨线程 向量数据库/键值存储 LangGraph Store + Embedding
知识图谱 结构化实体关系 图数据库 Neo4j + LLM 实体抽取

大部分框架默认只有短期记忆(就是对话历史拼进 prompt),长期记忆和知识图谱需要自己搭。下面我逐个讲清楚怎么做。

二、短期记忆:靠上下文窗口,但有限度

短期记忆本质就是 LLM 的 context window。Agent 把历史消息逐条塞进 prompt,模型看到前面说了什么才能继续接话。

LangGraph 把这层抽象成 Checkpointer。每次 graph 跑完,当前状态(messages、agent 中间步骤)都会保存成一个 checkpoint。下一次同一 thread 唤醒时,checkpointer 自动恢复状态。

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph

# 创建一个内存级别的 checkpointer(进程级,重启丢失)
checkpointer = MemorySaver()

builder = StateGraph(AgentState)
# ... 定义节点和边 ...
graph = builder.compile(checkpointer=checkpointer)

# 第一次调用
config = {"configurable": {"thread_id": "session-001"}}
result = graph.invoke({"messages": [{"role": "user", "content": "我叫小明"}]}, config)

# 第二次调用——Agent 还记得你叫小明
result = graph.invoke({"messages": [{"role": "user", "content": "我叫什么名字?"}]}, config)
# 输出:你叫小明

⚠️ 注意:MemorySaver 存在内存里,进程重启就没了。生产环境用 PostgresSaverSqliteSaver

from langgraph.checkpoint.postgres import PostgresSaver

# 重启后依然能恢复(需先建好数据库)
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@localhost:5432/langgraph"
)
graph = builder.compile(checkpointer=checkpointer)

短期记忆的问题也明显:上下文一长,推理延迟和 token 开销都会涨。我的做法是设置 max_turns 阈值,超过后自动 trigger summarization,把早期对话浓缩成摘要塞回系统 prompt。

三、长期记忆:跨会话记住用户

用户昨天告诉你他喜欢简洁的回答风格,今天来问你技术问题——如果你还得重新问一遍,用户早跑了。

LangGraph 用 Store 做长期记忆。Store 是一个键值存储,按 namespace 组织数据,跨 thread 共享。

from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore

# 生产环境推荐用 PostgresStore
store = InMemoryStore()

# 保存用户偏好
store.put(
    ("users", "u-1001"),
    "preferences",
    {"style": "concise", "language": "zh", "topics": ["LLM", "RAG"]}
)

# 其他 thread 也能读到
prefs = store.get(("users", "u-1001"), "preferences")
print(prefs.value["style"])  # "concise"

但有个问题——Store 存的必须是明确已知的键。如果你的 Agent 需要「模糊回忆」(比如用户上次提到过一个 Python 框架,但忘了叫什么),光靠 Store 不够,还得上向量检索:

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
import uuid

# 每次 Agent 学到新信息,同时写入 Store 和向量库
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
    collection_name="agent_memories",
    embedding_function=embeddings
)

# 保存记忆
memory_text = "用户对 FastAPI 和 Pydantic 很熟悉,正在学 LangChain"
vectorstore.add_texts(
    texts=[memory_text],
    ids=[str(uuid.uuid4())],
    metadatas=[{"user_id": "u-1001", "timestamp": "2026-07-14"}]
)

# 下次用户说"上次那个 Python web 框架"——向量检索能召回
results = vectorstore.similarity_search("Python web frameworks", k=3)

这里我踩过一个坑:写入向量库之前一定要做 chunking + dedup,不然多条相似记忆反复写入,召回时全是同一个意思的不同表述。CrewAI 的 Memory 系统内置了 Consolidation 去重,写代码的时候注意检查一下。

CrewAI 的 Memory 设计更开箱即用。它在 0.105+ 版本之后统一了记忆系统,自动从 task output 中抽取 key facts,支持 scope 树(agent_memory.remember(...) 按作用域搜索),内置 dedup 和 consolidation:

from crewai import Agent, Crew, Task, Process

research_agent = Agent(
    role="研究员",
    goal="研究技术主题并记录发现",
    memory=True  # 启用记忆
)

write_agent = Agent(
    role="写手",
    goal="根据研究发现撰写报告",
    memory=True
)

task1 = Task(description="研究 Agent 记忆系统的架构设计", agent=research_agent)
task2 = Task(description="撰写技术方案报告", agent=write_agent)

crew = Crew(
    agents=[research_agent, write_agent],
    tasks=[task1, task2],
    process=Process.sequential,
    memory=True  # 开启 crew 级共享记忆
)

跑完之后,write_agent 会自动检索 research_agent 之前产出的 findings,不需要手动传 context。

四、知识图谱记忆:比向量检索更精准

向量检索擅长模糊匹配,但碰到精确问题就露怯了。

比如用户说「帮我查一下小明负责的项目」。向量库里存的是「小明是项目A的负责人」,语义相似也许能召回,但如果你问「小明和项目A的关系是什么?」——向量检索就回答不了这种结构化查询。

知识图谱就是干这个的。把实体和关系抽出来存进图数据库,Agent 用 Cypher 查询精确信息:

from neo4j import GraphDatabase

driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password"))

# Agent 从对话中抽取实体关系并写入图
def extract_and_store_entities(text):
    # 用 LLM 做 NER(命名实体识别)
    entities = llm.extract_entities(text)
    # {
    #   "entities": [{"name": "小明", "type": "Person"}, {"name": "项目A", "type": "Project"}],
    #   "relations": [{"subject": "小明", "predicate": "负责", "object": "项目A"}]
    # }
    
    with driver.session() as session:
        for ent in entities["entities"]:
            session.run(
                "MERGE (n:Entity {name: name, type:type})",
                name=ent["name"], type=ent["type"]
            )
        for rel in entities["relations"]:
            session.run(
                """
                MATCH (s:Entity {name: subj}), (o:Entity {name:obj})
                MERGE (s)-[r:RELATION {type: $pred}]->(o)
                """,
                subj=rel["subject"], obj=rel["object"], pred=rel["predicate"]
            )

# Agent 查询时
def query_graph(user_query):
    # LLM 把自然语言转成 Cypher(或者预先定义好模板)
    cypher = "MATCH (p:Person)-[:负责]->(proj:Project) WHERE p.name = '小明' RETURN proj.name"
    with driver.session() as session:
        result = session.run(cypher)
        return [r["proj.name"] for r in result]
    # 返回:["项目A"]

这一步的难点不在技术,而在 实体抽取的准确率。GPT-4 的 NER 能力很强但也有幻觉,我的做法是加一层校验——抽取出来的三元组,让模型再确认一遍「这个实体关系在原文中能找到依据吗?」。

关于记忆的融合,可以参考 MemGPT(现已更名为 Letta)的思路。MemGPT 是最早在 Agent 中引入「操作系统级内存管理」的项目,把 Agent 记忆分成了 Core Memory(系统提示中的固定信息)和 Archival Memory(通过向量检索的长期记忆),中间用 Recall Memory(对话历史)衔接。这个分层的设计理念直接影响了很多后来的框架,包括 LangGraph 的 Store + Checkpointer 架构。

总结

Agent 的记忆系统不是「加一个 vector store」就能解决的。我自己的实践经验是:短期记忆靠 checkpoint + 上下文窗口,长期记忆靠 KV store + 向量检索兜底,精确查询靠知识图谱。三层叠加,才能让 Agent 既记得住今天聊了什么,也记得住昨天聊了什么,还能精确回答实体关系问题。

下一篇我会分享 Agent 在生产环境中的观测与调试——这比开发 Agent 本身难得多。LLM 输出不确定性高,怎么打日志、怎么 trace 每步决策、怎么复现问题,这些都是上线之前必须搞定的。

参考文献

转载请注明:Falost的小窝 » Agent 记忆系统实战:短期记忆、长期记忆与知识图谱(架构+代码拆解)

如果你觉得这篇文章不错或者对你有帮助,想请我喝一杯咖啡,可以打赏
喜欢 (0)
发表我的评论
取消评论

表情

Hi,您需要填写昵称和邮箱!

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址