AI应用 Agent记忆RAG向量数据库持续学习
云
云铭
进化之路 · 扫码阅读
微信 · 浏览器扫码
在手机上获得更好的阅读体验
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是可怕的——它能记住你三周前提过的一个小偏好，在下一次对话时自然地用到。但这也意味着巨大的隐私责任。最好的记忆系统，应该像一位值得信赖的管家：知道很多，但从不多嘴，并且随时准备在你需要时遗忘一切。