AI应用 Agent记忆RAG向量数据库持续学习
Agent的记忆与学习：让AI越用越聪明
道
量子道
量子道 · 扫码阅读
微信 · 浏览器扫码
在手机上获得更好的阅读体验
Agent的记忆与学习：让AI越用越聪明
没有记忆的Agent，每次对话都是一张白纸。给它记忆能力后，它会从工具变成伙伴——越用越懂你，越用越好用。
Agent记忆的三个层次
类比人脑
感官记忆 → 短期记忆 → 长期记忆
↓ ↓ ↓
对话上下文 会话缓存 持久化存储
毫秒 分钟-小时 天-年| 层次 | AI对应 | 技术实现 | 容量 | 持久性 |
|---|---|---|---|---|
| 感官记忆 | 当前对话窗口 | LLM上下文 | 4K-1M tokens | 单次对话 |
| 工作记忆 | 会话状态和工具结果 | Dict/Redis | 任意 | 单次会话 |
| 长期记忆 | 用户偏好、历史经验 | 向量数据库+RAG | 近乎无限 | 永久 |
第一层：扩展上下文记忆
超越默认上下文窗口的会话记忆
from typing import List, Dict
from datetime import datetime
class ConversationMemory:
会话记忆管理
def __init__(self, max_turns=20, summary_threshold=10):
self.messages: List[Dict] = []
self.summary: str =
self.max_turns = max_turns
self.summary_threshold = summary_threshold
def add_message(self, role: str, content: str):
self.messages.append({
role: role,
content: content,
timestamp: datetime.now().isoformat()
})
# 当消息积累超过阈值，自动压缩
if len(self.messages) > self.summary_threshold * 2:
self._compress()
def _compress(self):
将旧消息压缩为摘要
old_messages = self.messages[:-self.summary_threshold * 2]
recent_messages = self.messages[-self.summary_threshold * 2:]
# 用LLM总结旧消息
summary_prompt = f
将以下对话历史压缩为一段简短的摘要，保留：
- 用户的偏好和习惯
- 已经完成的重要操作
- 未完成的任务
- 关键决策和原因
对话历史：
{self._format_messages(old_messages)}
压缩摘要（不超过300字）：
self.summary = llm.invoke(summary_prompt).content
self.messages = recent_messages
def get_context(self) -> str:
获取当前上下文的完整表示
context =
if self.summary:
context += f[先前的对话摘要]\n{self.summary}\n\n[最近的对话]\n
context += self._format_messages(self.messages)
return context
def _format_messages(self, msgs):
return \n.join([f{m[role]}: {m[content][:200]} for m in msgs])记忆窗口策略
class SlidingWindowMemory:
滑动窗口+关键事件记忆
def __init__(self, window_size=10):
self.window_size = window_size
self.full_history = []
self.key_events = [] # 被标记为重要的事件
def add(self, message, importance_score=None):
self.full_history.append(message)
# 自动评估重要性
if importance_score is None:
importance_score = self._auto_score_importance(message)
if importance_score > 0.7:
self.key_events.append({
message: message,
importance: importance_score,
timestamp: datetime.now()
})
def _auto_score_importance(self, message):
让LLM评估消息重要性
prompt = f
评估以下对话消息的重要性（0.0-1.0）：
- 0.8+：关键决策、个人偏好、重要承诺
- 0.5-0.7：有用信息、具体任务要求
- 0.2-0.4：一般讨论
- 0.0-0.1：寒暄、无信息量
消息：{message[content][:200]}
只输出一个数字。
return float(llm.invoke(prompt).content.strip())
def get_context(self):
混合上下文：关键事件 + 最近窗口
context = [关键历史事件]\n
for event in self.key_events[-10:]: # 最近10个关键事件
context += f- {event[message][content][:150]}\n
context += f\n[最近{self.window_size}轮对话]\n
for msg in self.full_history[-self.window_size:]:
context += f{msg[role]}: {msg[content][:150]}\n
return context第二层：向量数据库长期记忆
用RAG构建Agent的知识库
import chromadb
from chromadb.utils import embedding_functions
import hashlib
class AgentLongTermMemory:
基于向量数据库的长期记忆
def __init__(self, collection_name=agent_memory):
self.client = chromadb.PersistentClient(path=./agent_memory_db)
self.embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
api_key=os.getenv(OPENAI_API_KEY),
model_name=text-embedding-3-small
)
# 三个集合，三种记忆类型
self.facts = self.client.get_or_create_collection(
user_facts,
embedding_function=self.embedding_fn
)
self.experiences = self.client.get_or_create_collection(
experiences,
embedding_function=self.embedding_fn
)
self.learned_patterns = self.client.get_or_create_collection(
learned_patterns,
embedding_function=self.embedding_fn
)
def remember_fact(self, fact: str, category: str, confidence: float):
记住用户相关的事实
fact_id = hashlib.md5(fact.encode()).hexdigest()
self.facts.upsert(
ids=[fact_id],
documents=[fact],
metadatas=[{
category: category,
confidence: confidence,
last_updated: datetime.now().isoformat(),
times_verified: 1
}]
)
def recall_relevant(self, query: str, memory_type=all, top_k=5) -> List[str]:
根据当前查询回忆相关内容
memories = []
if memory_type in (all, facts):
results = self.facts.query(query_texts=[query], n_results=top_k)
if results[documents][0]:
memories.extend([
f[事实] {doc} for doc in results[documents][0]
])
if memory_type in (all, experiences):
results = self.experiences.query(query_texts=[query], n_results=top_k)
if results[documents][0]:
memories.extend([
f[经验] {doc} for doc in results[documents][0]
])
if memory_type in (all, patterns):
results = self.learned_patterns.query(query_texts=[query], n_results=top_k)
if results[documents][0]:
memories.extend([
f[模式] {doc} for doc in results[documents][0]
])
return memories
def learn_from_interaction(self, interaction: dict):
从每次交互中自动学习
# 提取新事实
new_facts = self._extract_facts(interaction)
for fact in new_facts:
self.remember_fact(
fact[content],
fact[category],
fact[confidence]
)
# 提取经验教训
lesson = self._extract_lesson(interaction)
if lesson:
self.experiences.add(
ids=[hashlib.md5(lesson.encode()).hexdigest()],
documents=[lesson],
metadatas=[{timestamp: datetime.now().isoformat()}]
)
def _extract_facts(self, interaction) -> List[dict]:
从对话中自动提取事实
prompt = f
从以下对话中提取关于用户的新事实：
对话：
{interaction[messages]}
提取格式（JSON列表）：
[
{{content: 用户是软件工程师, category: 职业, confidence: 0.9}},
{{content: 用户偏好Python, category: 偏好, confidence: 0.8}}
]
只提取明确陈述的事实，不推测。如果没有新事实，输出空列表。
response = llm.invoke(prompt)
return json.loads(response.content)第三层：持续学习与自适应
从反馈中学习
class AdaptiveAgent:
能从反馈中学习的Agent
def __init__(self, llm, memory: AgentLongTermMemory):
self.llm = llm
self.memory = memory
self.feedback_buffer = [] # 收集反馈用于批量学习
def execute_with_feedback(self, task: str) -> str:
执行任务并收集隐式反馈
# 先回忆相关经验
relevant_memory = self.memory.recall_relevant(task)
memory_context = \n.join(relevant_memory) if relevant_memory else 无相关记忆
# 增强prompt
enhanced_prompt = f
[你的历史记忆]
{memory_context}
[当前任务]
{task}
利用历史记忆中的经验来更好地完成当前任务。
result = self.llm.invoke(enhanced_prompt)
# 自动评估本次表现
self.feedback_buffer.append({
task: task,
result: result.content,
memory_used: relevant_memory
})
return result.content
def learn_from_batch(self):
批量学习，提炼模式
if len(self.feedback_buffer) < 10:
return # 积累够一定量才学习
prompt = f
分析以下Agent交互记录，提炼可以指导未来行为的模式和规则：
交互记录（{len(self.feedback_buffer)}条）：
{json.dumps(self.feedback_buffer, ensure_ascii=False)[:5000]}
提炼：
1. 重复出现的任务模式（如用户经常问XX类问题）
2. 这些任务的最佳处理策略
3. 需要记住的用户偏好
4. 应该避免的错误
输出格式：
{{patterns: [...], best_practices: [...], user_preferences: [...], pitfalls: [...]}}
response = self.llm.invoke(prompt)
learnings = json.loads(response.content)
# 存储模式到记忆
for pattern in learnings.get(patterns, []):
self.memory.learned_patterns.add(
ids=[hashlib.md5(str(pattern).encode()).hexdigest()],
documents=[str(pattern)],
metadatas=[{type: pattern}]
)
# 清空缓冲区
self.feedback_buffer = []
print(f学习完成：提炼了 {len(learnings.get(patterns, []))} 个模式)记忆系统的工程考量
记忆衰减
class DecayingMemory:
模拟人类记忆的衰减特性
def __init__(self, half_life_days=30):
self.half_life = half_life_days
self.memories = [] # [(content, timestamp, initial_importance)]
def get_weight(self, timestamp, importance):
计算记忆的当前权重
age_days = (datetime.now() - timestamp).days
decay_factor = 0.5 ** (age_days / self.half_life)
return importance * decay_factor
def recall(self, query, top_k=5):
衰减加权的回忆
relevant = self._semantic_search(query)
weighted = [(m, self.get_weight(ts, imp)) for m, ts, imp in relevant]
weighted.sort(key=lambda x: x[1], reverse=True)
return [m for m, _ in weighted[:top_k] if _ > 0.1] # 过滤弱记忆记忆冲突处理
def resolve_memory_conflict(old_fact, new_fact):
当新事实与旧记忆矛盾时
prompt = f
发现记忆冲突：
旧记忆（{old_fact[last_updated]}）：
{old_fact[content]}
置信度：{old_fact[confidence]}
已验证次数：{old_fact[times_verified]}
新信息（{new_fact[timestamp]}）：
{new_fact[content]}
来源可靠性：{new_fact[source_reliability]}
判断：
1. 是否存在真正的冲突？
2. 如果是，应该保留哪个？（新旧/合并/都保留等进一步验证）
3. 推理过程
输出JSON：
{{is_conflict: true/false, action: keep_old/keep_new/merge/flag_for_review, merged_content: 如果合并的话}}
return llm.invoke(prompt)隐私与伦理
Agent记忆的隐私边界
关键原则：
1. 记忆透明化
└── 让用户知道Agent记住了什么，有遗忘按钮
2. 目的限制
└── 只为提升服务质量而记忆，不做他用
3. 分级记忆
└── 公开信息 ↔ 会话信息 ↔ 个人信息（不同安全等级）
4. 遗忘权
└── 用户可以要求删除特定记忆或全部重置class PrivacyAwareMemory:
def forget_user(self, user_id):
用户要求遗忘时，彻底清除相关记忆
self.facts.delete(where={user_id: user_id})
self.experiences.delete(where={user_id: user_id})
self.learned_patterns.delete(where={user_id: user_id})一个有记忆的Agent是可怕的——它能记住你三周前提过的一个小偏好，在下一次对话时自然地用到。但这也意味着巨大的隐私责任。最好的记忆系统，应该像一位值得信赖的管家：知道很多，但从不多嘴，并且随时准备在你需要时遗忘一切。