做了几个Agent项目之后，你会发现一个尴尬的事实：Agent的记忆力比金鱼还差。上下文窗口一刷新，前面对话说了啥全忘了。用户第三次问同一个问题，它照样一脸茫然。这不是模型能力的问题，是记忆架构的问题。
这篇文章分享一套向量数据库+知识图谱的混合记忆系统，让Agent真正拥有跨会话的长期记忆。
为什么需要长期记忆
大模型的上下文窗口再大，也是有边界的。4K、32K、128K——不管多大，总会有溢出的时候。而且每次把整个历史塞进上下文，token成本也扛不住。
真正实用的Agent需要三种记忆：
工作记忆（Working Memory） → 当前会话的上下文，随会话结束清空
短期记忆（Short-term） → 最近几轮对话的摘要，TTL过期
长期记忆（Long-term） → 用户偏好、历史事实、学到的教训，持久存储
我们的方案聚焦长期记忆。核心思路：用向量库做语义检索，用知识图谱做结构化推理，两者互补。
整体架构
用户对话 → 记忆提取器 → ┬→ 向量库（Qdrant） → 语义回忆
└→ 知识图谱（Neo4j） → 关系推理
↓
记忆组装器 → 注入上下文 → Agent响应
三个核心组件：记忆提取器从对话中抽取值得记住的信息；两个存储各司其职；记忆组装器在Agent响应前把相关记忆注入上下文。
记忆提取：什么值得记住
不是所有对话都值得记。必须有一个过滤层：
from pydantic import BaseModel
from enum import Enum
from openai import AsyncOpenAI
class MemoryType(str, Enum):
FACT = "fact" # 事实：用户说过的客观信息
PREFERENCE = "preference" # 偏好：用户的选择倾向
EVENT = "event" # 事件：发生过的事情
LESSON = "lesson" # 教训：从错误中学到的
class ExtractedMemory(BaseModel):
content: str # 记忆内容
memory_type: MemoryType
importance: float # 0-1，重要性评分
entities: list[str] # 涉及的实体
relations: list[dict] # 实体间关系
class MemoryExtractor:
"""从对话中提取值得长期保存的记忆"""
def __init__(self, client: AsyncOpenAI, model: str = "gpt-4o"):
self.client = client
self.model = model
async def extract(self, user_msg: str, assistant_msg: str) -> list[ExtractedMemory]:
prompt = f"""分析以下对话，提取值得长期记忆的信息。
用户：{user_msg}
助手：{assistant_msg}
只提取以下类型的信息：
1. 用户明确表达的偏好或习惯
2. 用户提到的客观事实（姓名、地址、项目等）
3. 重要的交互事件
4. 从错误或反馈中学到的教训
不要提取：通用知识、闲聊内容、一次性指令。
以JSON数组返回，每个元素包含：content, memory_type, importance(0-1), entities, relations。
如果没有值得记忆的内容，返回空数组[]。"""
resp = await self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
temperature=0,
)
import json
data = json.loads(resp.choices[0].message.content)
memories = data.get("memories", [])
return [ExtractedMemory(**m) for m in memories]
关键设计：importance评分让系统能区分"用户叫张三"和"用户今天心情不错"。低重要性的记忆会在清理时优先被淘汰。
向量库：语义回忆
向量库存储记忆的embedding，支持"用户之前提过什么关于Python的事"这类模糊查询。
from qdrant_client import AsyncQdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from sentence_transformers import SentenceTransformer
import uuid
class VectorMemoryStore:
"""基于Qdrant的向量记忆存储"""
def __init__(self, qdrant_url: str = "http://localhost:6333",
collection: str = "agent_memory",
embed_model: str = "BAAI/bge-small-zh-v1.5"):
self.client = AsyncQdrantClient(url=qdrant_url)
self.collection = collection
self.encoder = SentenceTransformer(embed_model)
self.dim = 512
async def setup(self):
"""初始化集合"""
collections = await self.client.get_collections()
names = [c.name for c in collections.collections]
if self.collection not in names:
await self.client.create_collection(
collection_name=self.collection,
vectors_config=VectorParams(
size=self.dim,
distance=Distance.COSINE,
),
)
async def store(self, memory: ExtractedMemory, user_id: str):
"""存储一条记忆"""
embedding = self.encoder.encode(
memory.content,
normalize_embeddings=True
)
point = PointStruct(
id=str(uuid.uuid4()),
vector=embedding.tolist(),
payload={
"content": memory.content,
"type": memory.memory_type.value,
"importance": memory.importance,
"entities": memory.entities,
"user_id": user_id,
"created_at": datetime.now().isoformat(),
"access_count": 0,
"last_accessed": None,
},
)
await self.client.upsert(
collection_name=self.collection,
points=[point],
)
async def recall(self, query: str, user_id: str,
top_k: int = 5,
min_score: float = 0.7) -> list[dict]:
"""检索相关记忆"""
embedding = self.encoder.encode(
query,
normalize_embeddings=True
)
results = await self.client.search(
collection_name=self.collection,
query_vector=embedding.tolist(),
query_filter={
"must": [
{"key": "user_id", "match": {"value": user_id}}
]
},
limit=top_k,
score_threshold=min_score,
)
memories = []
for hit in results:
mem = hit.payload
mem["score"] = hit.score
memories.append(mem)
# 更新访问计数（衰减淘汰用）
await self.client.set_payload(
collection_name=self.collection,
payload={
"access_count": mem.get("access_count", 0) + 1,
"last_accessed": datetime.now().isoformat(),
},
points=[hit.id],
)
return memories
知识图谱：关系推理
向量库擅长"找相似"，但不擅长"找关系"。比如"张三的老板是谁"、“这个项目的依赖有哪些”——这类问题需要图谱。
from neo4j import AsyncGraphDatabase
class GraphMemoryStore:
"""基于Neo4j的知识图谱记忆存储"""
def __init__(self, uri: str = "bolt://localhost:7687",
user: str = "neo4j", password: str = "password"):
self.driver = AsyncGraphDatabase.driver(uri, auth=(user, password))
async def store_entity(self, name: str, entity_type: str, properties: dict = None):
"""存储实体"""
props = properties or {}
async with self.driver.session() as session:
await session.run(
"MERGE (e:Entity {name: $name, type: $type}) "
"SET e += $props",
name=name, type=entity_type, props=props,
)
async def store_relation(self, source: str, target: str,
relation: str, properties: dict = None):
"""存储关系"""
props = properties or {}
async with self.driver.session() as session:
await session.run(
"MERGE (s:Entity {name: $source}) "
"MERGE (t:Entity {name: $target}) "
"MERGE (s)-[r:RELATION {type: $rel}]->(t) "
"SET r += $props",
source=source, target=target, rel=relation, props=props,
)
async def query_relations(self, entity: str, depth: int = 2) -> list[dict]:
"""查询实体的关系网络"""
async with self.driver.session() as session:
result = await session.run(
"MATCH path = (e:Entity {name: $entity})"
"-[:RELATION*1..$depth]-(related) "
"RETURN DISTINCT related.name AS name, "
"related.type AS type, "
"length(path) AS distance",
entity=entity, depth=depth,
)
return [dict(record) async for record in result]
async def find_path(self, source: str, target: str) -> list[dict]:
"""查找两个实体之间的路径"""
async with self.driver.session() as session:
result = await session.run(
"MATCH path = shortestPath("
"(s:Entity {name: $source})-[:RELATION*]-(t:Entity {name: $target})"
") RETURN [n IN nodes(path) | n.name] AS path",
source=source, target=target,
)
record = await result.single()
return record["path"] if record else []
实际使用时，从记忆提取器出来的entities和relations会同时写入图谱：
async def store_memory(self, memory: ExtractedMemory, user_id: str):
"""双写：向量库 + 知识图谱"""
# 向量库存储完整记忆
await self.vector_store.store(memory, user_id)
# 知识图谱存储实体和关系
for entity in memory.entities:
await self.graph_store.store_entity(
name=entity,
entity_type="extracted",
properties={"source_user": user_id},
)
for rel in memory.relations:
await self.graph_store.store_relation(
source=rel["source"],
target=rel["target"],
relation=rel["type"],
)
记忆组装：注入上下文
Agent响应前，需要把检索到的记忆组装成自然语言注入上下文。这里有个技巧：不要把所有记忆都塞进去，按相关性和重要性排序，只取Top-K。
class MemoryAssembler:
"""记忆组装器：将检索结果转化为上下文"""
def __init__(self, vector_store: VectorMemoryStore,
graph_store: GraphMemoryStore):
self.vector = vector_store
self.graph = graph_store
async def build_context(self, user_msg: str, user_id: str,
max_tokens: int = 1500) -> str:
"""构建记忆上下文"""
# 1. 向量检索相关记忆
vector_memories = await self.vector.recall(
user_msg, user_id, top_k=5, min_score=0.7
)
# 2. 提取消息中的实体，查图谱
entities = self._extract_entities(user_msg)
graph_contexts = []
for entity in entities[:3]: # 最多查3个实体
relations = await self.graph.query_relations(entity, depth=1)
if relations:
graph_contexts.append({
"entity": entity,
"relations": relations,
})
# 3. 组装上下文
parts = []
if vector_memories:
parts.append("## 相关记忆")
for mem in vector_memories:
parts.append(f"- [{mem['type']}] {mem['content']}")
if graph_contexts:
parts.append("\n## 相关关系")
for ctx in graph_contexts:
rels = ", ".join(
f"{r['name']}({r['type']})" for r in ctx["relations"][:5]
)
parts.append(f"- {ctx['entity']}相关: {rels}")
context = "\n".join(parts)
# 4. 截断到token限制
if self._count_tokens(context) > max_tokens:
context = self._truncate(context, max_tokens)
return context
def _extract_entities(self, text: str) -> list[str]:
"""简单实体提取（生产环境用NER模型）"""
import re
patterns = [
r"[A-Z][a-z]+(?:\s[A-Z][a-z]+)*",
r"[\u4e00-\u9fa5]{2,4}(?:项目|系统|平台|服务)",
]
entities = []
for p in patterns:
entities.extend(re.findall(p, text))
return list(set(entities))
记忆淘汰：遗忘也是能力
无限增长的记忆库最终会变慢、变贵。需要一个淘汰策略：
class MemoryDecay:
"""记忆衰减与淘汰"""
def __init__(self, vector_store: VectorMemoryStore):
self.store = vector_store
async def apply_decay(self, user_id: str, max_memories: int = 10000):
"""对超出上限的记忆做淘汰"""
all_points = await self.store.client.scroll(
collection_name=self.store.collection,
scroll_filter={
"must": [{"key": "user_id", "match": {"value": user_id}}]
},
limit=max_memories + 1,
)
points = all_points[0]
if len(points) <= max_memories:
return
scored = []
now = datetime.now()
for point in points:
payload = point.payload
importance = payload.get("importance", 0.5)
access_count = payload.get("access_count", 0)
last_accessed = payload.get("last_accessed")
if last_accessed:
days_since = (now - datetime.fromisoformat(last_accessed)).days
time_decay = max(0, 1 - days_since / 365)
else:
time_decay = 0.5
frequency_bonus = min(1, access_count / 10)
score = importance * 0.4 + time_decay * 0.4 + frequency_bonus * 0.2
scored.append((point.id, score))
scored.sort(key=lambda x: x[1])
to_delete = scored[:len(points) - max_memories]
delete_ids = [s[0] for s in to_delete]
if delete_ids:
await self.store.client.delete(
collection_name=self.store.collection,
points_selector=delete_ids,
)
淘汰公式：score = importance * 0.4 + time_decay * 0.4 + frequency * 0.2。重要的、最近访问的、高频访问的记忆保留下来；不重要的、很久没碰的自然淘汰。
完整流程串起来
class AgentMemorySystem:
"""Agent长期记忆系统"""
def __init__(self):
self.extractor = MemoryExtractor(client=AsyncOpenAI())
self.vector = VectorMemoryStore()
self.graph = GraphMemoryStore()
self.assembler = MemoryAssembler(self.vector, self.graph)
self.decay = MemoryDecay(self.vector)
async def setup(self):
await self.vector.setup()
async def process_turn(self, user_msg: str, assistant_msg: str,
user_id: str) -> str:
"""处理一轮对话：提取记忆 + 返回上下文供下一轮用"""
memories = await self.extractor.extract(user_msg, assistant_msg)
for mem in memories:
await self.store_memory(mem, user_id)
context = await self.assembler.build_context(user_msg, user_id)
return context
性能与成本
实测数据（1000用户，30天运行）：
记忆总量： ~45,000条
向量检索P99延迟： 23ms
图谱查询P99延迟： 45ms
记忆提取Token消耗： ~200 tokens/轮（GPT-4o）
淘汰执行频率： 每周一次，每次约5分钟
存储占用： Qdrant ~180MB, Neo4j ~90MB
成本可控，但要注意embedding模型的选择。中文场景推荐BGE或M3E，英文用all-MiniLM就够。别用OpenAI的embedding API做高频写入，成本会爆。
踩坑记录
-
不要把敏感信息存进记忆。用户密码、身份证号——这些应该走加密存储，不是向量库。记忆提取器的prompt里要明确排除。
-
实体消歧很重要。“张三"和"三哥"可能是同一个人，不做消歧图谱会乱成一锅粥。
-
记忆冲突处理。用户说"我喜欢咖啡"过了一个月说"我不喝咖啡了”，需要更新而非重复存储。提取时加一个conflict_detection步骤。
-
别过度记忆。每次对话都提取十几条记忆，库会膨胀很快。importance阈值设高一点，宁缺毋滥。
总结
Agent长期记忆的核心不是"记住所有东西"，而是"在对的时候想起对的事"。向量库解决语义检索，知识图谱解决关系推理，两者配合才能覆盖真实的记忆需求。加上合理的淘汰策略，系统可以长期稳定运行而不膨胀。
这套架构已经在几个客服和助手类Agent上跑了几个月，效果比纯RAG方案好很多——因为Agent不再把用户当陌生人了。