Diseño del sistema de memoria
La memoria humana se divide en memoria a corto plazo y memoria a largo plazo; el sistema de memoria de los Agentes de IA también adopta un diseño similar.
Comprender las diferencias y usos de estos dos tipos de memoria es la base para construir agentes inteligentes.
Memoria a corto plazo: memoria de trabajo del Agente
La memoria a corto plazo es como el banco de trabajo del Agente; almacena el contexto de la conversación actual e información temporal.
Características：
- Capacidad limitada: generalmente solo puede guardar las últimas conversaciones (por ejemplo, las últimas 10-20 rondas de diálogo)
- Acceso rápido: la velocidad de lectura y escritura es muy rápida
- Temporalidad: generalmente se elimina o comprime después de que finaliza la conversación
- Dependiente del contexto: afecta directamente la generación de la respuesta actual
Usos principales：
- Mantener la coherencia de la conversación
- Almacenar los resultados intermedios de la tarea actual
- Recordar las preferencias que el usuario mencionó en la conversación actual
Implementación técnica：
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()
Memoria a largo plazo: base de conocimientos del Agente
La memoria a largo plazo es como el archivo del Agente; almacena información importante que debe conservarse durante mucho tiempo.
Características：
- Gran capacidad: puede almacenar una gran cantidad de información
- Persistencia: la información se guarda durante mucho tiempo y no se borra automáticamente
- Acceso por recuperación: recupera información relevante mediante consultas, en lugar de lectura secuencial
- Almacenamiento estructurado: la información suele almacenarse de forma estructurada para facilitar su recuperación
Usos principales：
- Almacenar información personal y preferencias del usuario
- Acumular conocimientos y experiencia
- Recordar contenido importante de conversaciones
- Guardar los resultados de la ejecución de tareas
Implementación técnica：
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]
Análisis comparativo
| Característica | Memoria a corto plazo | Memoria a largo plazo |
|---|---|---|
| Capacidad | Pequeña (nivel KB) | Grande (nivel GB o incluso mayor) |
| Persistencia | Temporal (nivel de conversación) | Persistente (conservación a largo plazo) |
| Método de acceso | Acceso secuencial | Acceso por recuperación |
| Velocidad | Rápido | Relativamente lenta |
| Usos principales | Contexto de conversación | Acumulación de conocimiento |
| Complejidad de implementación | Simple | Compleja |
Arquitectura del sistema de memoria
Un sistema de memoria completo generalmente incluye tanto memoria a corto plazo como memoria a largo plazo:
Gestión del historial de conversación
Una gestión eficaz del historial de conversación es clave para garantizar que el Agente de IA pueda mantener conversaciones coherentes y significativas. Dado que la longitud de contexto de los LLM es limitada, es necesario gestionar de forma inteligente qué información histórica debe conservarse y cuál puede descartarse.
Desafíos del historial de conversación
- Límite de longitud de contexto: la mayoría de los LLM tienen un límite fijo de tokens (como 4K, 8K, 16K, 32K, etc.)
- La importancia de la información varía: no todas las conversaciones históricas son igualmente importantes
- La relevancia cambia con el tiempo: las conversaciones más recientes suelen ser más relevantes
- Costo acumulado: un contexto más largo implica un mayor costo de llamadas a la API
Gestión básica del historial de conversación
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()
Estrategias inteligentes de selección de historial
No todas las conversaciones históricas son igualmente importantes. Una selección inteligente del historial puede mejorar la eficiencia del uso del contexto:
1. Decaimiento basado en el tiempo
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. Selección basada en relevancia
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. Estrategia de selección mixta
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"])
Compresión del historial de conversación
Cuando el historial de conversación es demasiado largo, se puede comprimir para conservar la información esencial:
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)
Aplicaciones de bases de datos vectoriales
La base de datos vectorial es la tecnología central de los sistemas de memoria a largo plazo; permite al Agente recuperar información relevante basándose en la similitud semántica, no solo en la coincidencia de palabras clave.
¿Qué es una base de datos vectorial?
Una base de datos vectorial se especializa en almacenar y recuperar vectores (embeddings). En el contexto de los Agentes de IA:
- Texto → Vector: convierte texto en vectores de alta dimensión mediante un modelo de embeddings
- Similitud → Distancia: los textos similares tienen vectores cercanos; la similitud puede medirse mediante la distancia entre vectores
- Recuperación → Búsqueda de vecinos más cercanos: busca los vectores más similares al vector de consulta
Flujo de trabajo de una base de datos vectorial
Uso de bases de datos vectoriales para implementar memoria a largo plazo
1. Instalar las librerías necesarias
# 安装 embedding 模型和向量数据库 pip install sentence-transformers # 用于生成embeddings pip install chromadb # 轻量级向量数据库
2. Crear un sistema de memoria vectorial
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. Ejemplo completo de uso de memoria vectorial
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字符
Consejos de optimización para bases de datos vectoriales
Elegir un modelo de embeddings adecuado：
- Soporte multilingüe：
paraphrase-multilingual-MiniLM-L12-v2 - Optimizado para inglés：
all-MiniLM-L6-v2 - Alta calidad：
text-embedding-ada-002（OpenAI）
Filtrado de metadatos: utiliza la función de filtrado de metadatos de la base de datos vectorial para mejorar la precisión de la recuperación
# 使用元数据过滤
results = collection.query(
query_embeddings=[query_embedding],
n_results=5,
where={"type": "personal_preference"}, # 只检索个人偏好类型的记忆
where_document={"$contains": "编程"} # 只检索包含"编程"的文档
)
Búsqueda híbrida:Combinar búsqueda vectorial y búsqueda por palabras clave
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]
Estrategias de compresión y resumen de memoria
A medida que avanza la conversación, la memoria se acumula continuamente. Para evitar la sobrecarga de información y reducir el consumo de recursos, es necesario comprimir y resumir la memoria periódicamente.
¿Por qué se necesita compresión de memoria?
- Ahorrar espacio de almacenamiento: los registros de conversación originales pueden ser muy extensos
- Mejorar la eficiencia de recuperación: la memoria comprimida es más fácil de recuperar y procesar
- Extraer información esencial: eliminar redundancias y conservar lo esencial
- Reducir el costo computacional: procesar memoria comprimida es más eficiente que procesar memoria original
Estrategias de compresión de memoria
1. Compresión basada en importancia
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. Compresión basada en agrupamiento (clustering)
Agrupar memorias similares y luego resumir cada grupo:
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. Estrategia de resumen incremental
Resumir gradualmente durante la conversación, en lugar de procesar todo el historial de una sola vez:
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)
Gestión del ciclo de vida de la memoria
Un sistema de memoria completo debe gestionar todo el ciclo de vida de la memoria:
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"]) Otras extensiones