Memory System Design
Human memory is divided into short-term and long-term memory, and the AI Agent's memory system adopts a similar design.
Understanding the differences and uses of these two types of memory is the foundation for building an intelligent Agent.
Short-term Memory: Agent's Working Memory
Short-term memory is like the Agent's workbench, storing the context and temporary information of the current conversation.
Characteristics：
- Limited capacity: Usually only the most recent few conversations can be saved (e.g., the latest 10-20 rounds of dialogue)
- Fast access: Read and write speed is very fast
- Temporary nature: Usually cleared or compressed after the conversation ends
- Context-dependent: Directly affects current response generation
Main uses：
- Maintain conversational coherence
- Store intermediate results of the current task
- Remember preferences mentioned by the user in the current conversation
Technical implementation：
class ShortTermMemory:
"""短期记忆实现"""
def __init__(self, max_turns=10):
self.max_turns = max_turns # 最大对话轮数
self.conversation_history = [] # 对话历史
self.temporary_data = {} # 临时数据存储
def add_message(self, role, content):
"""添加消息到对话历史"""
message = {"role": role, "content": content, "timestamp": time.time()}
self.conversation_history.append(message)
# 保持历史不超过最大长度
if len(self.conversation_history) > self.max_turns:
self.conversation_history.pop(0)
def get_context(self):
"""获取对话上下文（用于发送给LLM）"""
return self.conversation_history[-self.max_turns:]
def store_temp(self, key, value):
"""存储临时数据"""
self.temporary_data[key] = value
def get_temp(self, key, default=None):
"""获取临时数据"""
return self.temporary_data.get(key, default)
def clear_temp(self):
"""清除临时数据"""
self.temporary_data.clear()
Long-term Memory: Agent's Knowledge Base
Long-term memory is like the Agent's archive room, storing important information that needs to be retained long-term.
Characteristics：
- Large capacity: Can store large amounts of information
- Persistence: Information is stored long-term and will not be automatically cleared
- Retrieval-based access: Retrieve relevant information through queries, rather than sequential reading
- Structured storage: Information is usually stored in a structured manner for easy retrieval
Main uses：
- Store users' personal information and preferences
- Accumulate knowledge and experience
- Remember important conversation content
- Save task execution results
Technical implementation：
class LongTermMemory:
"""长期记忆基类"""
def __init__(self):
self.memories = [] # 记忆条目列表
def add_memory(self, content, metadata=None):
"""添加记忆"""
memory = {
"id": str(uuid.uuid4()),
"content": content,
"metadata": metadata or {},
"timestamp": time.time(),
"importance": 0.5 # 默认重要性
}
self.memories.append(memory)
return memory["id"]
def search_memories(self, query, limit=5):
"""搜索记忆（子类需要实现具体搜索逻辑）"""
raise NotImplementedError
def get_memory(self, memory_id):
"""获取特定记忆"""
for memory in self.memories:
if memory["id"] == memory_id:
return memory
return None
def delete_memory(self, memory_id):
"""删除记忆"""
self.memories = [m for m in self.memories if m["id"] != memory_id]
Comparative Analysis
| Features | Short-term memory | Long-term memory |
|---|---|---|
| Capacity | Small (KB-level) | Large (GB-level or even larger) |
| persistence | Temporary (conversation-level) | Persistent (long-term storage) |
| Access method | Sequential access | Retrieval-based access. |
| Speed | Fast | relatively slow |
| Main purpose | Conversation context. | knowledge accumulation |
| Implementation complexity | Simple | complex |
Memory System Architecture
A complete memory system typically includes both short-term and long-term memory:
Conversation History Management
Effective conversation history management is key to ensuring that an AI Agent can conduct coherent, meaningful conversations. Since LLMs have limited context length, it is necessary to intelligently manage which historical information should be retained and which can be discarded.
Challenges of Conversation History
- Context length limitationMost LLMs have fixed token limits (e.g., 4K, 8K, 16K, 32K, etc.)
- Different importance of information: Not all historical conversations are equally important
- Relevance changes over time: The more recent the conversation, the more relevant it usually is
- cumulative cost: Longer context means higher API call costs
Basic Conversation History Management
class ConversationManager:
"""对话管理器"""
def __init__(self, max_context_length=4000):
self.max_context_length = max_context_length # 最大token数
self.history = [] # 完整的对话历史
self.active_context = [] # 当前活跃的上下文
def add_exchange(self, user_input, assistant_response):
"""添加一轮对话"""
self.history.append({
"user": user_input,
"assistant": assistant_response,
"timestamp": time.time()
})
def build_context(self, current_query, include_history=True):
"""构建当前查询的上下文"""
if not include_history or not self.history:
# 没有历史或不需要历史，只返回当前查询
return [{"role": "user", "content": current_query}]
# 从最近的对话开始，逐步添加历史，直到达到长度限制
context = []
context_length = self.estimate_tokens(current_query)
# 添加当前查询
context.insert(0, {"role": "user", "content": current_query})
# 从最近到最远添加历史
for exchange in reversed(self.history):
user_tokens = self.estimate_tokens(exchange["user"])
assistant_tokens = self.estimate_tokens(exchange["assistant"])
# 检查是否会超出限制
if context_length + user_tokens + assistant_tokens > self.max_context_length:
break
# 添加助理回复（在用户输入之前）
context.insert(0, {"role": "assistant", "content": exchange["assistant"]})
context.insert(0, {"role": "user", "content": exchange["user"]})
context_length += user_tokens + assistant_tokens
return context
def estimate_tokens(self, text):
"""粗略估计文本的token数量（实际应用中应使用准确的tokenizer）"""
# 简单估算：英文约0.75单词/token，中文约1-2字符/token
if self.is_chinese(text):
return len(text) // 2 # 中文每2字符约1个token
else:
words = len(text.split())
return int(words * 1.3) # 英文每单词约1.3个token
def is_chinese(self, text):
"""判断文本是否主要为中文"""
chinese_chars = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
return chinese_chars / max(len(text), 1) > 0.3
def clear_history(self):
"""清空对话历史"""
self.history.clear()
self.active_context.clear()
Intelligent History Selection Strategy
Not all historical conversations are equally important. Intelligent history selection can improve the efficiency of context usage:
1. Time-based decay
def time_based_selection(history, current_time, max_items=10):
"""基于时间的选择：越近的对话权重越高"""
scored_items = []
for item in history:
# 计算时间衰减分数（越近分数越高）
time_diff = current_time - item["timestamp"]
time_score = max(0, 1 - time_diff / 3600) # 1小时内完全保留，之后衰减
# 结合其他因素（如对话长度、重要性标记等）
total_score = time_score
scored_items.append((total_score, item))
# 按分数排序，选择分数最高的
scored_items.sort(key=lambda x: x[0], reverse=True)
return [item for score, item in scored_items[:max_items]]
2. Relevance-based selection
def relevance_based_selection(history, current_query, embedding_model, max_items=5):
"""基于与当前查询相关性的选择"""
if not history:
return []
# 计算当前查询的向量
query_embedding = embedding_model.encode(current_query)
scored_items = []
for item in history:
# 将历史对话内容转换为向量
content = item["user"] + " " + item["assistant"]
content_embedding = embedding_model.encode(content)
# 计算余弦相似度
similarity = cosine_similarity([query_embedding], [content_embedding])[0][0]
scored_items.append((similarity, item))
# 按相似度排序，选择最相关的
scored_items.sort(key=lambda x: x[0], reverse=True)
return [item for similarity, item in scored_items[:max_items]]
3. Hybrid selection strategy
class SmartHistorySelector:
"""智能历史选择器"""
def __init__(self, embedding_model=None):
self.embedding_model = embedding_model
def select_history(self, history, current_query, max_context_length=3000):
"""智能选择历史对话"""
selected = []
current_length = 0
# 策略1：总是包含最近的一轮对话
if history:
recent = history[-1]
recent_length = self.estimate_length(recent)
if recent_length <= max_context_length:
selected.append(recent)
current_length += recent_length
# 策略2：基于相关性的选择
if self.embedding_model and len(history) > 1:
relevant = self.select_by_relevance(history[:-1], current_query, 3)
for item in relevant:
item_length = self.estimate_length(item)
if current_length + item_length <= max_context_length:
selected.append(item)
current_length += item_length
# 策略3：如果还有空间，按时间顺序添加
remaining_space = max_context_length - current_length
if remaining_space > 100: # 至少100token的空间
for item in history:
if item not in selected:
item_length = self.estimate_length(item)
if item_length <= remaining_space:
selected.append(item)
remaining_space -= item_length
# 按时间顺序排序
selected.sort(key=lambda x: x["timestamp"])
return selected
def select_by_relevance(self, history, query, max_items):
"""基于相关性选择"""
# 简化实现，实际应使用向量相似度
query_lower = query.lower()
scored = []
for item in history:
content = (item["user"] + " " + item["assistant"]).lower()
# 简单关键词匹配
score = sum(1 for word in query_lower.split() if word in content)
scored.append((score, item))
scored.sort(key=lambda x: x[0], reverse=True)
return [item for score, item in scored[:max_items]]
def estimate_length(self, exchange):
"""估计对话长度"""
return len(exchange["user"]) + len(exchange["assistant"])
Conversation History Compression
When conversation history becomes too long, it can be compressed to retain core information:
class HistoryCompressor:
"""对话历史压缩器"""
def __init__(self, llm_client):
self.llm = llm_client
def compress_conversation(self, conversation_history, max_summary_length=500):
"""压缩对话历史"""
if len(conversation_history) <= 2:
return conversation_history # 太短不需要压缩
# 将对话历史转换为文本
conversation_text = self.history_to_text(conversation_history)
# 使用LLM生成摘要
prompt = f"""
请将以下对话历史压缩为一个简短的摘要，保留核心信息和重要细节：
对话历史：
{conversation_text}
摘要要求：
1. 不超过{max_summary_length}字
2. 保留用户的主要需求和助理的关键回答
3. 忽略问候语、重复内容和无关细节
摘要：
"""
summary = self.llm.generate(prompt, max_tokens=max_summary_length)
# 创建压缩后的历史（摘要 + 最近几轮对话）
compressed_history = [
{
"role": "system",
"content": f"之前的对话摘要：{summary}"
}
]
# 保留最近的1-2轮对话以保持连贯性
for exchange in conversation_history[-2:]:
compressed_history.append({
"role": "user" if exchange["role"] == "user" else "assistant",
"content": exchange["content"]
})
return compressed_history
def history_to_text(self, history):
"""将对话历史转换为文本"""
lines = []
for exchange in history:
role = "用户" if exchange["role"] == "user" else "助理"
lines.append(f"{role}: {exchange['content']}")
return "\n".join(lines)
Vector Database Application
Vector databases are the core technology of long-term memory systems, allowing Agents to retrieve relevant information based on semantic similarity rather than just keyword matching.
What is a Vector Database?
Vector databases are specifically designed to store and retrieve vectors (embeddings). In the context of AI Agents:
- Text → vector: Convert text to high-dimensional vectors using an embedding model
- Similarity → Distance: Similar texts have similar vectors, and similarity can be measured by vector distance
- Retrieval → Nearest neighbor search: Find the vector most similar to the query vector
Vector Database Workflow
Implementing Long-term Memory with Vector Database
1. Install the required libraries
# 安装 embedding 模型和向量数据库 pip install sentence-transformers # 用于生成embeddings pip install chromadb # 轻量级向量数据库
2. Create a vector memory system
import chromadb
from sentence_transformers import SentenceTransformer
import uuid
import time
class VectorMemory:
"""基于向量数据库的记忆系统"""
def __init__(self, persist_directory="./memory_db"):
# 初始化 embedding 模型
self.embedding_model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
# 初始化 ChromaDB
self.client = chromadb.PersistentClient(path=persist_directory)
self.collection = self.client.get_or_create_collection(
name="agent_memories",
metadata={"description": "AI Agent 的长期记忆"}
)
def add_memory(self, content, metadata=None, importance=0.5):
"""添加记忆到向量数据库"""
# 生成 embedding
embedding = self.embedding_model.encode(content).tolist()
# 准备元数据
full_metadata = {
"timestamp": time.time(),
"importance": importance,
"content_length": len(content)
}
if metadata:
full_metadata.update(metadata)
# 生成唯一ID
memory_id = str(uuid.uuid4())
# 添加到数据库
self.collection.add(
documents=[content],
embeddings=[embedding],
metadatas=[full_metadata],
ids=[memory_id]
)
return memory_id
def search_memories(self, query, n_results=5, min_similarity=0.3):
"""搜索相关记忆"""
# 生成查询的 embedding
query_embedding = self.embedding_model.encode(query).tolist()
# 在向量数据库中搜索
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=n_results,
include=["documents", "metadatas", "distances"]
)
# 处理结果
memories = []
if results["documents"]:
for i in range(len(results["documents"][0])):
similarity = 1 - results["distances"][0][i] # 转换距离为相似度
if similarity >= min_similarity:
memory = {
"content": results["documents"][0][i],
"metadata": results["metadatas"][0][i],
"similarity": similarity,
"id": results["ids"][0][i]
}
memories.append(memory)
# 按相似度排序
memories.sort(key=lambda x: x["similarity"], reverse=True)
return memories
def get_relevant_context(self, query, max_memories=3):
"""获取与查询相关的记忆作为上下文"""
memories = self.search_memories(query, n_results=max_memories)
if not memories:
return ""
# 构建上下文字符串
context_parts = []
for i, memory in enumerate(memories):
content = memory["content"]
similarity = memory["similarity"]
timestamp = memory["metadata"]["timestamp"]
# 格式化时间
time_str = time.strftime("%Y-%m-%d %H:%M", time.localtime(timestamp))
context_parts.append(
f"[相关记忆 {i+1}，相似度：{similarity:.2f}，时间：{time_str}]\n{content}"
)
return "\n\n".join(context_parts)
def delete_memory(self, memory_id):
"""删除记忆"""
self.collection.delete(ids=[memory_id])
def get_all_memories(self, limit=100):
"""获取所有记忆（按时间倒序）"""
# 注意：ChromaDB 没有直接的获取所有功能
# 这里通过搜索一个通用查询来获取
results = self.collection.query(
query_embeddings=[self.embedding_model.encode(" ").tolist()], # 空查询
n_results=limit
)
memories = []
if results["documents"]:
for i in range(len(results["documents"][0])):
memory = {
"content": results["documents"][0][i],
"metadata": results["metadatas"][0][i],
"id": results["ids"][0][i]
}
memories.append(memory)
# 按时间倒序排序
memories.sort(key=lambda x: x["metadata"]["timestamp"], reverse=True)
return memories
3. Complete example of using vector memory
class AgentWithMemory:
"""带有记忆系统的AI Agent"""
def __init__(self, llm_client):
self.llm = llm_client
self.short_term_memory = ShortTermMemory(max_turns=10)
self.long_term_memory = VectorMemory()
def process_query(self, user_query):
"""处理用户查询"""
# 1. 从长期记忆中检索相关信息
relevant_context = self.long_term_memory.get_relevant_context(user_query)
# 2. 构建完整的上下文
short_term_context = self.short_term_memory.get_context()
# 3. 准备系统提示词
system_message = "你是一个有帮助的AI助手。"
if relevant_context:
system_message += f"\n\n相关背景信息：\n{relevant_context}"
# 4. 调用LLM生成回答
messages = [{"role": "system", "content": system_message}]
messages.extend(short_term_context)
messages.append({"role": "user", "content": user_query})
response = self.llm.chat_completion(messages)
# 5. 更新短期记忆
self.short_term_memory.add_message("user", user_query)
self.short_term_memory.add_message("assistant", response)
# 6. 判断是否应该保存到长期记忆
if self.should_save_to_long_term(user_query, response):
self.save_to_long_term(user_query, response)
return response
def should_save_to_long_term(self, user_query, response):
"""判断是否应该保存到长期记忆"""
# 简单的启发式规则
important_keywords = ["偏好", "喜欢", "不喜欢", "重要", "记住", "下次", "经常"]
query_lower = user_query.lower()
response_lower = response.lower()
# 检查是否包含重要关键词
for keyword in important_keywords:
if keyword in query_lower or keyword in response_lower:
return True
# 检查是否是关于个人信息
personal_keywords = ["我叫", "我是", "我住在", "我的电话", "我的邮箱"]
for keyword in personal_keywords:
if keyword in query_lower:
return True
return False
def save_to_long_term(self, user_query, response):
"""保存到长期记忆"""
# 将对话作为一条记忆保存
memory_content = f"用户：{user_query}\n助理：{response}"
metadata = {
"type": "conversation",
"user_query": user_query[:100], # 截断以避免过长
"response_length": len(response)
}
self.long_term_memory.add_memory(memory_content, metadata)
# 使用示例
if __name__ == "__main__":
# 创建带记忆的Agent
agent = AgentWithMemory(llm_client)
# 模拟对话
conversations = [
"你好，我叫张三，住在北京",
"我喜欢编程和阅读",
"我不喜欢早起",
"今天天气怎么样？",
"我的编程偏好是什么？" # 这个问题应该能检索到之前的记忆
]
for query in conversations:
print(f"\n用户: {query}")
response = agent.process_query(query)
print(f"助理: {response[:100]}...") # 只打印前100字符
Optimization Tips for Vector Database
Choose the right embedding model：
- Multilingual support：
paraphrase-multilingual-MiniLM-L12-v2 - English optimization：
all-MiniLM-L6-v2 - High quality：
text-embedding-ada-002（OpenAI）
Metadata filtering: Use the metadata filtering feature of vector databases to improve retrieval accuracy
# 使用元数据过滤
results = collection.query(
query_embeddings=[query_embedding],
n_results=5,
where={"type": "personal_preference"}, # 只检索个人偏好类型的记忆
where_document={"$contains": "编程"} # 只检索包含"编程"的文档
)
Hybrid search:Combine vector search with keyword search
def hybrid_search(query, vector_memory, keyword_weight=0.3):
"""混合搜索：结合向量相似度和关键词匹配"""
# 向量搜索
vector_results = vector_memory.search_memories(query, n_results=10)
# 关键词搜索（简化实现）
keyword_results = []
query_words = set(query.lower().split())
# 这里简化实现，实际应从数据库检索
all_memories = vector_memory.get_all_memories(limit=100)
for memory in all_memories:
content_words = set(memory["content"].lower().split())
common_words = query_words & content_words
keyword_score = len(common_words) / max(len(query_words), 1)
if keyword_score > 0:
memory["keyword_score"] = keyword_score
keyword_results.append(memory)
# 合并结果
all_results = {}
for result in vector_results:
result_id = result["id"]
all_results[result_id] = {
"vector_score": result["similarity"],
"keyword_score": 0,
"content": result["content"],
"metadata": result["metadata"]
}
for result in keyword_results:
result_id = result["id"]
if result_id in all_results:
all_results[result_id]["keyword_score"] = result["keyword_score"]
else:
all_results[result_id] = {
"vector_score": 0,
"keyword_score": result["keyword_score"],
"content": result["content"],
"metadata": result["metadata"]
}
# 计算综合分数
final_results = []
for result_id, result in all_results.items():
combined_score = (
(1 - keyword_weight) * result["vector_score"] +
keyword_weight * result["keyword_score"]
)
result["combined_score"] = combined_score
final_results.append(result)
# 按综合分数排序
final_results.sort(key=lambda x: x["combined_score"], reverse=True)
return final_results[:5]
Memory Compression and Summarization Strategy
As the conversation progresses, memories keep accumulating. To avoid information overload and reduce resource consumption, memories need to be regularly compressed and summarized.
Why is Memory Compression Needed?
- Save storage space: Raw conversation logs can be very large
- Improve retrieval efficiency: Compressed memories are easier to retrieve and process
- Extract core information: Remove redundancy and retain the essence
- Reduce computational cost: Processing compressed memories is more efficient than processing raw memories
Memory Compression Strategy
1. Importance-based compression
class ImportanceBasedCompressor:
"""基于重要性的记忆压缩器"""
def __init__(self, llm_client):
self.llm = llm_client
def compress_memories(self, memories, target_ratio=0.5):
"""压缩记忆，保留重要内容"""
if len(memories) <= 1:
return memories # 记忆太少，不需要压缩
# 评估每个记忆的重要性
scored_memories = []
for memory in memories:
importance = self.evaluate_importance(memory)
scored_memories.append((importance, memory))
# 按重要性排序
scored_memories.sort(key=lambda x: x[0], reverse=True)
# 保留最重要的部分
keep_count = max(1, int(len(memories) * target_ratio))
compressed_memories = [memory for _, memory in scored_memories[:keep_count]]
return compressed_memories
def evaluate_importance(self, memory):
"""评估记忆的重要性"""
content = memory["content"]
metadata = memory.get("metadata", {})
# 基于规则的重要性评估
importance_score = 0.0
# 1. 基于类型
memory_type = metadata.get("type", "")
if memory_type == "personal_info":
importance_score += 0.8
elif memory_type == "preference":
importance_score += 0.7
elif memory_type == "fact":
importance_score += 0.5
elif memory_type == "conversation":
importance_score += 0.3
# 2. 基于长度（适中的长度可能更重要）
content_length = len(content)
if 50 <= content_length <= 500:
importance_score += 0.2
elif content_length > 500:
importance_score += 0.1
# 3. 基于时间衰减（越新的记忆越重要）
timestamp = metadata.get("timestamp", 0)
if timestamp > 0:
age_days = (time.time() - timestamp) / (24 * 3600)
recency_score = max(0, 1 - age_days / 30) # 30天内线性衰减
importance_score += recency_score * 0.5
# 4. 基于显式重要性标记
explicit_importance = metadata.get("importance", 0.5)
importance_score += explicit_importance * 0.5
return min(importance_score, 1.0) # 归一化到0-1
2. Clustering-based compression
Cluster similar memories, then summarize each cluster:
from sklearn.cluster import KMeans
import numpy as np
class ClusterBasedCompressor:
"""基于聚类的记忆压缩器"""
def __init__(self, embedding_model):
self.embedding_model = embedding_model
def compress_by_clustering(self, memories, n_clusters=None):
"""通过聚类压缩记忆"""
if len(memories) <= 3:
return memories # 记忆太少，不需要聚类
# 确定聚类数量
if n_clusters is None:
n_clusters = min(3, len(memories) // 2)
# 获取所有记忆的embedding
embeddings = []
for memory in memories:
embedding = self.embedding_model.encode(memory["content"])
embeddings.append(embedding)
embeddings = np.array(embeddings)
# 执行K-means聚类
kmeans = KMeans(n_clusters=n_clusters, random_state=42, n_init=10)
labels = kmeans.fit_predict(embeddings)
# 对每个聚类进行总结
compressed_memories = []
for cluster_id in range(n_clusters):
# 获取该聚类的所有记忆
cluster_memories = [
memories[i] for i in range(len(memories)) if labels[i] == cluster_id
]
if cluster_memories:
# 总结该聚类的记忆
summary = self.summarize_cluster(cluster_memories)
compressed_memories.append(summary)
return compressed_memories
def summarize_cluster(self, cluster_memories):
"""总结一个聚类的记忆"""
if len(cluster_memories) == 1:
# 只有一个记忆，直接返回
return cluster_memories[0]
# 合并所有记忆内容
all_content = "\n".join([m["content"] for m in cluster_memories])
# 使用LLM生成总结（这里简化实现）
# 实际应调用LLM生成高质量的总结
summary_content = f"相关主题的{len(cluster_memories)}条记忆摘要：{all_content[:500]}..."
# 合并元数据
merged_metadata = {
"type": "cluster_summary",
"original_count": len(cluster_memories),
"compressed": True,
"timestamp": time.time()
}
return {
"content": summary_content,
"metadata": merged_metadata
}
3. Incremental summarization strategy
Summarize progressively during the conversation rather than processing all history at once:
class IncrementalSummarizer:
"""增量总结器"""
def __init__(self, llm_client, summary_interval=5):
self.llm = llm_client
self.summary_interval = summary_interval # 每多少轮对话总结一次
self.conversation_buffer = []
self.summaries = []
def add_conversation(self, user_input, assistant_response):
"""添加对话到缓冲区"""
self.conversation_buffer.append({
"user": user_input,
"assistant": assistant_response,
"timestamp": time.time()
})
# 检查是否需要总结
if len(self.conversation_buffer) >= self.summary_interval:
self.create_summary()
def create_summary(self):
"""创建总结"""
if not self.conversation_buffer:
return
# 将对话缓冲区的所有内容总结为一段文字
conversation_text = ""
for exchange in self.conversation_buffer:
conversation_text += f"用户: {exchange['user']}\n"
conversation_text += f"助理: {exchange['assistant']}\n\n"
# 使用LLM生成总结
prompt = f"""
请将以下对话内容总结为一个简短的段落，保留核心信息和重要细节：
对话内容：
{conversation_text}
总结要求：
1. 不超过200字
2. 保留用户的主要需求和助理的关键回答
3. 忽略问候语、重复内容和无关细节
总结：
"""
summary = self.llm.generate(prompt, max_tokens=200)
# 保存总结
self.summaries.append({
"content": summary,
"timestamp": time.time(),
"original_count": len(self.conversation_buffer)
})
# 清空缓冲区
self.conversation_buffer.clear()
def get_context(self, include_recent=True, include_summaries=True):
"""获取上下文"""
context_parts = []
if include_summaries and self.summaries:
# 添加所有总结
for i, summary in enumerate(self.summaries[-3:]): # 最多3个总结
context_parts.append(f"[对话总结 {i+1}]\n{summary['content']}")
if include_recent and self.conversation_buffer:
# 添加最近的对话
for exchange in self.conversation_buffer[-3:]: # 最多3轮最近对话
context_parts.append(f"用户: {exchange['user']}")
context_parts.append(f"助理: {exchange['assistant']}")
return "\n\n".join(context_parts)
Memory Lifecycle Management
A complete memory system should manage the entire lifecycle of memories:
class MemoryLifecycleManager:
"""记忆生命周期管理器"""
def __init__(self):
self.short_term = ShortTermMemory()
self.long_term = VectorMemory()
self.compressor = ImportanceBasedCompressor()
def process_memory(self, content, initial_importance=0.5):
"""处理新记忆"""
# 1. 先存入短期记忆
self.short_term.store_temp("recent_memory", content)
# 2. 评估重要性
evaluated_importance = self.evaluate_importance(content, initial_importance)
# 3. 根据重要性决定存储策略
if evaluated_importance >= 0.8:
# 高重要性：立即存入长期记忆
self.long_term.add_memory(content, {"importance": evaluated_importance})
elif evaluated_importance >= 0.5:
# 中等重要性：标记为待处理
self.queue_for_review(content, evaluated_importance)
else:
# 低重要性：仅保留在短期记忆，稍后清理
pass
def evaluate_importance(self, content, initial_importance):
"""评估记忆重要性"""
# 这里可以实现更复杂的评估逻辑
return initial_importance
def queue_for_review(self, content, importance):
"""将记忆加入待处理队列"""
# 实现队列逻辑
pass
def periodic_maintenance(self):
"""定期维护"""
# 1. 压缩长期记忆
all_memories = self.long_term.get_all_memories(limit=1000)
if len(all_memories) > 100:
compressed = self.compressor.compress_memories(all_memories, target_ratio=0.7)
# 更新长期记忆（实际实现中可能需要更复杂的逻辑）
# 这里简化处理
# 2. 清理旧的低重要性记忆
self.cleanup_old_memories()
def cleanup_old_memories(self, max_age_days=30, min_importance=0.3):
"""清理旧记忆"""
# 获取所有记忆
all_memories = self.long_term.get_all_memories(limit=1000)
current_time = time.time()
for memory in all_memories:
metadata = memory.get("metadata", {})
timestamp = metadata.get("timestamp", 0)
importance = metadata.get("importance", 0.5)
# 计算记忆年龄
age_days = (current_time - timestamp) / (24 * 3600)
# 删除旧且不重要的记忆
if age_days > max_age_days and importance < min_importance:
self.long_term.delete_memory(memory["id"]) Other extensions