Módulo 6: Prompt Composition y Chaining
6. Multi-Turn Strategies y Memory
Descripción
Las conversaciones multi-turn (multi-turno) presentan un desafío único: el historial crece con cada intercambio, y eventualmente supera el límite del context window. Además, no todos los mensajes anteriores son igualmente relevantes para la pregunta actual.
En esta cápsula aprenderás tres estrategias para gestionar el historial de conversación: memoria selectiva (incluir solo los N mensajes más recientes), summarization del historial (comprimir mensajes antiguos), y relevance-based selection (seleccionar mensajes por similitud semántica).
El Problema del Historial Creciente
Turno 1: Usuario: "Hola, soy Ana" → 5 tokens
Bot: "Hola Ana, ¿en qué te ayudo?" → 10 tokens
Turno 2: Usuario: "Tengo un problema con mi API" → 10 tokens
Bot: "Cuéntame más sobre el problema..." → 15 tokens
...
Turno 20: Historial acumulado → ~8,000 tokens
Turno 50: Historial acumulado → ~25,000 tokens
Turno 100: Historial acumulado → ~60,000 tokens ← Peligroso para gpt-4o-mini
ADEMÁS:
- Mensajes de hace 50 turnos probablemente ya no son relevantes
- El costo por llamada aumenta linealmente con el historial
- A más contexto innecesario, puede haber más "ruido" en las respuestas
Arquitectura de un Sistema Conversacional
MULTI-TURN SYSTEM
┌─────────────────────────────────────────────────┐
│ MEMORIA │
│ │
│ ┌────────────────┐ ┌─────────────────────┐ │
│ │ Historial Full │ │ Perfil de Usuario │ │
│ │ (todos msgs) │ │ (datos persistentes│ │
│ └───────┬────────┘ └──────────┬──────────┘ │
│ │ │ │
│ ▼ │ │
│ ┌────────────────┐ │ │
│ │ Estrategia de │ │ │
│ │ Selección de │ │ │
│ │ Contexto │ │ │
│ └───────┬────────┘ │ │
└──────────┼────────────────────────┼─────────────┘
│ │
▼ ▼
┌─────────────────────────────────────┐
│ CONTEXT CONSTRUIDO │
│ [System prompt con perfil] │
│ [Resumen de conversación previa] │
│ [Últimos N mensajes relevantes] │
│ [Mensaje actual del usuario] │
└────────────────────┬────────────────┘
│
▼
LLM Response
Implementación Base
from openai import OpenAI
from dataclasses import dataclass, field
from typing import Optional
import json
import time
client = OpenAI()
@dataclass
class Mensaje:
"""Representa un mensaje en la conversación."""
role: str
content: str
timestamp: float = field(default_factory=time.time)
tokens: int = 0
def to_dict(self) -> dict:
return {"role": self.role, "content": self.content}
@dataclass
class PerfilUsuario:
"""Perfil persistente del usuario para personalización."""
nombre: str = ""
preferencias: dict = field(default_factory=dict)
contexto_negocio: str = ""
idioma: str = "español"
historial_temas: list[str] = field(default_factory=list)
class SistemaConversacional:
"""
Sistema conversacional con gestión de memoria.
Combina múltiples estrategias según el tamaño del historial.
"""
def __init__(
self,
sistema_prompt: str,
max_tokens_contexto: int = 4000,
n_mensajes_recientes: int = 6,
resumir_despues_n: int = 10
):
self.sistema_prompt = sistema_prompt
self.max_tokens_contexto = max_tokens_contexto
self.n_mensajes_recientes = n_mensajes_recientes
self.resumir_despues_n = resumir_despues_n
self.historial: list[Mensaje] = []
self.resumen_acumulado: str = ""
self.perfil: PerfilUsuario = PerfilUsuario()
def agregar_mensaje(self, role: str, content: str):
"""Agrega un mensaje al historial."""
msg = Mensaje(role=role, content=content)
# Estimar tokens
msg.tokens = len(content.split()) * 1.3 # Aproximación
self.historial.append(msg)
def construir_contexto(self, query_actual: str) -> list[dict]:
"""
Construye el contexto óptimo para la llamada al LLM.
Combina resumen + mensajes recientes + perfil.
"""
mensajes = []
# System prompt con contexto del perfil
sistema_completo = self.sistema_prompt
if self.perfil.nombre:
sistema_completo += f"\n\nUsuario: {self.perfil.nombre}"
if self.perfil.contexto_negocio:
sistema_completo += f"\nContexto: {self.perfil.contexto_negocio}"
mensajes.append({"role": "system", "content": sistema_completo})
# Agregar resumen si existe
if self.resumen_acumulado:
mensajes.append({
"role": "system",
"content": f"Resumen de la conversación previa:\n{self.resumen_acumulado}"
})
# Agregar últimos N mensajes
mensajes_recientes = self.historial[-self.n_mensajes_recientes:]
mensajes.extend([m.to_dict() for m in mensajes_recientes])
return mensajes
def chatear(self, mensaje_usuario: str) -> str:
"""Procesa un turno de conversación."""
# Verificar si hay que resumir historial antiguo
if len(self.historial) >= self.resumir_despues_n:
self._resumir_historial_antiguo()
# Agregar mensaje del usuario
self.agregar_mensaje("user", mensaje_usuario)
# Construir contexto
mensajes_contexto = self.construir_contexto(mensaje_usuario)
# Llamar al LLM
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=mensajes_contexto,
temperature=0.7,
max_tokens=500
)
respuesta = response.choices[0].message.content
# Agregar respuesta al historial
self.agregar_mensaje("assistant", respuesta)
# Actualizar perfil si es el primer mensaje
if len(self.historial) <= 2:
self._actualizar_perfil(mensaje_usuario)
return respuesta
def _resumir_historial_antiguo(self):
"""Comprime los mensajes antiguos en un resumen."""
# Tomar los mensajes que NO son los N más recientes
mensajes_a_resumir = self.historial[:-self.n_mensajes_recientes]
if not mensajes_a_resumir:
return
# Convertir a texto
conversacion_str = "\n".join([
f"{m.role.upper()}: {m.content}"
for m in mensajes_a_resumir
])
# Incluir resumen previo si existe
if self.resumen_acumulado:
conversacion_str = f"Resumen anterior:\n{self.resumen_acumulado}\n\nConversación adicional:\n{conversacion_str}"
# Generar nuevo resumen
self.resumen_acumulado = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""Resume esta conversación en máximo 5 oraciones.
Preserva: información personal del usuario (nombre, empresa, etc.),
decisiones tomadas, problemas resueltos, y contexto importante.
Conversación:
{conversacion_str}"""
}],
temperature=0,
max_tokens=200
).choices[0].message.content
# Eliminar mensajes resumidos del historial (mantener solo los recientes)
self.historial = self.historial[-self.n_mensajes_recientes:]
def _actualizar_perfil(self, mensaje: str):
"""Extrae información del usuario del primer mensaje."""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"""Extrae información del usuario de este mensaje si está disponible.
Si no hay información, deja campos vacíos.
Mensaje: {mensaje}
JSON: {{"nombre": str, "empresa": str, "contexto": str}}"""
}],
temperature=0,
response_format={"type": "json_object"}
).choices[0].message.content
try:
datos = json.loads(response)
if datos.get("nombre"):
self.perfil.nombre = datos["nombre"]
if datos.get("empresa") or datos.get("contexto"):
self.perfil.contexto_negocio = datos.get("contexto", datos.get("empresa", ""))
except Exception:
pass
def get_estadisticas(self) -> dict:
"""Estadísticas del estado actual de la conversación."""
return {
"mensajes_en_historial": len(self.historial),
"tiene_resumen": bool(self.resumen_acumulado),
"longitud_resumen": len(self.resumen_acumulado),
"perfil_nombre": self.perfil.nombre,
"tokens_estimados": sum(m.tokens for m in self.historial)
}
# Ejemplo de uso:
if __name__ == "__main__":
bot = SistemaConversacional(
sistema_prompt="Eres un asistente de programación Python. Ayudas con código, errores y buenas prácticas.",
n_mensajes_recientes=6,
resumir_despues_n=8
)
conversacion = [
"Hola, soy Pedro, trabajo en una startup de fintech",
"Tengo un problema con mi FastAPI, no puedo hacer que las rutas async funcionen",
"¿Cómo configuro uvicorn correctamente?",
"El error dice: RuntimeError: no running event loop",
"¿Puedes mostrarme un ejemplo completo?",
]
for mensaje in conversacion:
print(f"\nUsuario: {mensaje}")
respuesta = bot.chatear(mensaje)
print(f"Bot: {respuesta[:200]}...")
stats = bot.get_estadisticas()
print(f"[Historial: {stats['mensajes_en_historial']} msgs, Resumen: {bool(stats['tiene_resumen'])}]")Estrategia: Relevance-Based Selection con Embeddings
import math
from typing import Optional
def cosine_similarity(vec1: list[float], vec2: list[float]) -> float:
"""Calcula similitud coseno entre dos vectores."""
dot_product = sum(a * b for a, b in zip(vec1, vec2))
mag1 = math.sqrt(sum(a ** 2 for a in vec1))
mag2 = math.sqrt(sum(b ** 2 for b in vec2))
if mag1 * mag2 == 0:
return 0.0
return dot_product / (mag1 * mag2)
def get_embedding(texto: str, modelo: str = "text-embedding-3-small") -> list[float]:
"""Obtiene el embedding de un texto."""
response = client.embeddings.create(
model=modelo,
input=texto
)
return response.data[0].embedding
class MemoriaSemantica:
"""
Sistema de memoria que selecciona mensajes por relevancia semántica.
Útil cuando el historial es largo y heterogéneo.
"""
def __init__(self, k_mensajes: int = 4):
self.historial: list[tuple[dict, list[float]]] = [] # (mensaje, embedding)
self.k_mensajes = k_mensajes
def agregar_mensaje(self, role: str, content: str):
"""Agrega mensaje con su embedding pre-calculado."""
mensaje = {"role": role, "content": content}
embedding = get_embedding(content)
self.historial.append((mensaje, embedding))
def seleccionar_relevantes(
self,
query: str,
n_recientes: int = 2,
n_semanticos: int = 2
) -> list[dict]:
"""
Selecciona mensajes combinando:
- Los N más recientes (siempre incluidos para coherencia)
- Los K más similares semánticamente a la query actual
Args:
query: La pregunta o mensaje actual del usuario
n_recientes: Cuántos mensajes recientes incluir siempre
n_semanticos: Cuántos mensajes adicionales por similitud
Returns:
Lista de mensajes seleccionados, ordenados cronológicamente
"""
if not self.historial:
return []
# Mensajes recientes (siempre incluidos)
mensajes_recientes = [msg for msg, _ in self.historial[-n_recientes:]]
indices_recientes = set(range(len(self.historial) - n_recientes, len(self.historial)))
# Obtener embedding de la query actual
query_emb = get_embedding(query)
# Calcular similitud para mensajes NO recientes
candidatos_semanticos = []
for i, (msg, emb) in enumerate(self.historial[:-n_recientes]):
if i not in indices_recientes:
sim = cosine_similarity(query_emb, emb)
candidatos_semanticos.append((sim, i, msg))
# Ordenar por similitud y tomar los top K
candidatos_semanticos.sort(reverse=True)
mensajes_semanticos = [
(i, msg)
for sim, i, msg in candidatos_semanticos[:n_semanticos]
if sim > 0.7 # Umbral de relevancia
]
# Combinar y ordenar cronológicamente
todos_indices = set(indices_recientes)
todos_mensajes_con_indice = []
for i, msg in mensajes_semanticos:
if i not in todos_indices:
todos_mensajes_con_indice.append((i, msg))
todos_indices.add(i)
for i in range(len(self.historial) - n_recientes, len(self.historial)):
if i < len(self.historial):
msg, _ = self.historial[i]
todos_mensajes_con_indice.append((i, msg))
# Ordenar por índice (orden cronológico)
todos_mensajes_con_indice.sort(key=lambda x: x[0])
return [msg for _, msg in todos_mensajes_con_indice]
def chatear_con_memoria_semantica(
self,
mensaje: str,
sistema_prompt: str = "Eres un asistente útil."
) -> str:
"""Responde usando selección semántica de contexto."""
# Seleccionar contexto relevante
mensajes_relevantes = self.seleccionar_relevantes(mensaje)
# Construir mensajes para la API
messages = [{"role": "system", "content": sistema_prompt}]
messages.extend(mensajes_relevantes)
messages.append({"role": "user", "content": mensaje})
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
temperature=0.7,
max_tokens=400
)
respuesta = response.choices[0].message.content
# Agregar al historial
self.agregar_mensaje("user", mensaje)
self.agregar_mensaje("assistant", respuesta)
return respuesta
# Ejemplo:
memoria = MemoriaSemantica(k_mensajes=4)
sistema = "Eres un asistente técnico de Python."
respuestas = []
mensajes = [
"Mi nombre es Carlos y trabajo con FastAPI",
"¿Cómo manejo errores en FastAPI?",
"¿Qué es Pydantic?",
"¿Cuál es la diferencia entre async y sync en Python?",
"Volviendo al tema de errores en FastAPI, ¿cómo hago un error handler global?"
]
for msg in mensajes:
resp = memoria.chatear_con_memoria_semantica(msg, sistema)
print(f"\nUsuario: {msg}")
print(f"Bot: {resp[:150]}...")
print(f"Mensajes en historial: {len(memoria.historial)}")Estrategia: Context Pruning (Podado de Contexto)
def construir_contexto_inteligente(
historial: list[dict],
query_actual: str,
max_tokens: int = 4000,
siempre_incluir_n: int = 4
) -> list[dict]:
"""
Construye el contexto óptimo para el LLM usando múltiples estrategias:
1. Siempre incluir los N mensajes más recientes
2. Si hay espacio, incluir mensajes relevantes según keywords
3. Agregar resumen de mensajes excluidos si existe espacio
Args:
historial: Lista completa de mensajes {role, content}
query_actual: El mensaje actual del usuario
max_tokens: Presupuesto de tokens para el contexto
siempre_incluir_n: N mensajes recientes siempre incluidos
Returns:
Lista de mensajes optimizada para la llamada al LLM
"""
tokens_usados = 0
contexto_seleccionado = []
# Paso 1: Incluir siempre los N mensajes más recientes
recientes = historial[-siempre_incluir_n:]
for msg in recientes:
tokens_msg = len(msg["content"].split()) * 1.3
tokens_usados += tokens_msg
contexto_seleccionado = list(recientes)
if tokens_usados >= max_tokens:
return contexto_seleccionado
# Paso 2: Intentar incluir mensajes relevantes adicionales
# Calcular keywords de la query actual
keywords = set(query_actual.lower().split()) - {"de", "el", "la", "los", "las", "en", "que", "por", "a", "y"}
mensajes_anteriores = historial[:-siempre_incluir_n]
candidatos_con_relevancia = []
for i, msg in enumerate(mensajes_anteriores):
content_lower = msg["content"].lower()
relevancia = sum(1 for kw in keywords if kw in content_lower)
tokens_msg = len(msg["content"].split()) * 1.3
candidatos_con_relevancia.append((relevancia, i, msg, tokens_msg))
# Ordenar por relevancia
candidatos_con_relevancia.sort(reverse=True)
# Agregar mensajes relevantes hasta agotar el presupuesto
mensajes_adicionales = []
for relevancia, i, msg, tokens_msg in candidatos_con_relevancia:
if relevancia == 0:
break # No más relevantes
if tokens_usados + tokens_msg <= max_tokens:
mensajes_adicionales.append((i, msg))
tokens_usados += tokens_msg
# Ordenar mensajes adicionales cronológicamente y combinar
mensajes_adicionales.sort(key=lambda x: x[0])
if mensajes_adicionales:
# Agregar indicador de que hay contexto previo omitido
if len(historial) > len(recientes) + len(mensajes_adicionales):
contexto_seleccionado = [
{"role": "system", "content": "[Nota: Hay mensajes anteriores omitidos por espacio de contexto]"}
]
else:
contexto_seleccionado = []
contexto_seleccionado.extend([msg for _, msg in mensajes_adicionales])
contexto_seleccionado.extend(recientes)
return contexto_seleccionadoConversación Multi-Turn con Perfil Persistente
class AsistenteConPerfil:
"""
Asistente conversacional que aprende del usuario a lo largo de la conversación.
El perfil se construye automáticamente y personaliza las respuestas.
"""
def __init__(self, nombre_asistente: str, dominio: str):
self.nombre_asistente = nombre_asistente
self.dominio = dominio
self.historial: list[dict] = []
self.perfil: dict = {
"nombre": "",
"nivel_tecnico": "desconocido", # principiante/intermedio/avanzado
"preferencias": [],
"temas_frecuentes": [],
"problemas_resueltos": []
}
self._resumen = ""
def _system_prompt(self) -> str:
"""Genera el system prompt personalizado."""
prompt = f"""Eres {self.nombre_asistente}, un asistente experto en {self.dominio}.
PERFIL DEL USUARIO:
- Nombre: {self.perfil.get('nombre', 'desconocido')}
- Nivel técnico: {self.perfil.get('nivel_tecnico', 'desconocido')}
- Preferencias: {', '.join(self.perfil.get('preferencias', [])[:3])}
Adapta tu lenguaje al nivel técnico del usuario.
Si el nivel es principiante: usa analogías simples, evita jerga.
Si es intermedio: usa términos técnicos con pequeñas explicaciones.
Si es avanzado: va directo al punto, asume conocimiento base."""
if self._resumen:
prompt += f"\n\nCONTEXTO DE CONVERSACIÓN PREVIA:\n{self._resumen}"
return prompt
def _actualizar_perfil(self, usuario_msg: str, asistente_resp: str):
"""Actualiza el perfil basándose en el intercambio."""
# Detectar nivel técnico por vocabulario usado
palabras_avanzadas = ["async", "asyncio", "dependency injection", "SOLID", "microservices", "kubernetes", "latency", "throughput"]
palabras_basicas = ["¿qué es", "no entiendo", "para qué sirve", "explicame", "como funciona"]
msg_lower = usuario_msg.lower()
if any(w in msg_lower for w in palabras_avanzadas):
self.perfil["nivel_tecnico"] = "avanzado"
elif any(w in msg_lower for w in palabras_basicas) and self.perfil["nivel_tecnico"] == "desconocido":
self.perfil["nivel_tecnico"] = "principiante"
elif self.perfil["nivel_tecnico"] == "desconocido":
self.perfil["nivel_tecnico"] = "intermedio"
# Extraer nombre si no lo tenemos
if not self.perfil["nombre"]:
if "soy " in msg_lower:
partes = usuario_msg.split()
for i, palabra in enumerate(partes):
if palabra.lower() == "soy" and i + 1 < len(partes):
nombre_candidato = partes[i+1].replace(",", "").replace(".", "")
if nombre_candidato[0].isupper():
self.perfil["nombre"] = nombre_candidato
break
def chatear(self, mensaje: str) -> str:
"""Procesa un turno de conversación."""
# Agregar mensaje del usuario
self.historial.append({"role": "user", "content": mensaje})
# Construir mensajes para la API
messages = [{"role": "system", "content": self._system_prompt()}]
# Incluir últimos 8 mensajes (4 turnos)
messages.extend(self.historial[-8:])
# Llamar al LLM
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
temperature=0.7,
max_tokens=500
)
respuesta = response.choices[0].message.content
# Agregar respuesta al historial
self.historial.append({"role": "assistant", "content": respuesta})
# Actualizar perfil
self._actualizar_perfil(mensaje, respuesta)
# Comprimir historial si es muy largo
if len(self.historial) > 16:
mensajes_a_resumir = self.historial[:-8]
conv_str = "\n".join([f"{m['role'].upper()}: {m['content'][:100]}" for m in mensajes_a_resumir])
self._resumen = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"Resume en 3-4 oraciones los puntos clave:\n{conv_str}"}],
temperature=0,
max_tokens=150
).choices[0].message.content
self.historial = self.historial[-8:]
return respuesta
# Demo:
asistente = AsistenteConPerfil("CodeBot", "programación Python y FastAPI")
print("=== CONVERSACIÓN DEMO ===\n")
turnos = [
"Hola, soy María y estoy aprendiendo Python",
"¿Qué es una función decoradora?",
"¿Puedes darme un ejemplo simple?",
"Ahora quiero aprender sobre async. ¿Es difícil?",
"¿Y cómo uso async en FastAPI?"
]
for turno in turnos:
print(f"María: {turno}")
respuesta = asistente.chatear(turno)
print(f"Bot: {respuesta[:200]}...")
print(f"[Perfil: nivel={asistente.perfil['nivel_tecnico']}, nombre={asistente.perfil['nombre']}]\n")Troubleshooting
Problema 1: Context overflow silencioso
Síntoma: Las respuestas empiezan a ser incoherentes o el modelo "olvida" información de principios de la conversación.
Causa: El historial supera el context window sin advertencia.
Solución:
def verificar_tokens_antes_de_llamar(messages: list[dict], limite: int = 120000) -> bool:
"""Verifica que los mensajes no excedan el límite antes de llamar."""
total = sum(len(m["content"].split()) * 1.3 for m in messages)
if total > limite:
print(f"⚠ ADVERTENCIA: {total:.0f} tokens estimados supera el límite de {limite}")
return False
return TrueProblema 2: Pérdida de coherencia por summarization agresiva
Síntoma: El resumen pierde detalles importantes (nombre del usuario, decisiones específicas).
Solución:
def resumir_preservando_criticos(historial: list[dict]) -> str:
"""Resumen que preserva explícitamente los datos críticos."""
conv_str = "\n".join([f"{m['role'].upper()}: {m['content']}" for m in historial])
return client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"""Resume esta conversación (máx 5 oraciones).
CRÍTICO: DEBES preservar exactamente:
- Nombre del usuario (si se mencionó)
- Cualquier dato numérico (IDs, versiones, cantidades)
- Decisiones tomadas o compromisos hechos
- Errores específicos mencionados
Conversación:
{conv_str}"""}],
temperature=0,
max_tokens=200
).choices[0].message.contentProblema 3: Embeddings costosos para historial largo
Síntoma: Calcular embeddings de 100+ mensajes es caro y lento.
Solución: Pre-calcular y cachear embeddings:
import hashlib
from functools import lru_cache
@lru_cache(maxsize=500)
def get_embedding_cacheado(texto: str) -> tuple:
"""Embedding cacheado usando LRU cache (limitado en memoria)."""
emb = get_embedding(texto)
return tuple(emb) # tuple es hashable para lru_cacheEjercicios
Ejercicio 1: Implementar memory window adaptativa
El tamaño de la ventana de memoria debe adaptarse según la complejidad del tema:
- Preguntas simples: últimos 4 mensajes
- Preguntas complejas que refieren a contexto previo: últimos 8 mensajes
- Preguntas que hacen referencia explícita a algo mencionado antes: buscar en todo el historial
Ver solución
def construir_contexto_adaptativo(
historial: list[dict],
query: str
) -> list[dict]:
"""Ventana de contexto adaptativa según la query."""
# Detectar referencias al pasado
referencias_pasado = ["antes dijiste", "mencionaste que", "hablamos de", "como dije", "recuerdas que"]
tiene_referencia = any(r in query.lower() for r in referencias_pasado)
# Detectar complejidad
palabras_complejas = ["comparar", "analizar", "diferencia", "explicar", "relacionar"]
es_compleja = any(p in query.lower() for p in palabras_complejas)
if tiene_referencia:
# Buscar en todo el historial con keywords
keywords = set(query.lower().split()) - {"de", "el", "la", "los"}
candidatos = []
for i, msg in enumerate(historial[:-4]):
content = msg["content"].lower()
score = sum(1 for kw in keywords if kw in content)
if score > 0:
candidatos.append((score, i, msg))
candidatos.sort(reverse=True)
extras = [msg for _, _, msg in candidatos[:3]]
return extras + historial[-4:]
elif es_compleja:
return historial[-8:]
else:
return historial[-4:]Ejercicio 2: Persistencia de sesiones
Implementa un sistema que guarde el estado de la conversación (historial + resumen + perfil) en un archivo JSON, y lo pueda cargar para continuar la conversación en otra sesión.
Ver solución
import json
from pathlib import Path
class AsistentePersistente(AsistenteConPerfil):
def guardar_sesion(self, archivo: str):
"""Guarda el estado actual en JSON."""
estado = {
"historial": self.historial,
"perfil": self.perfil,
"resumen": self._resumen,
"timestamp": time.time()
}
with open(archivo, "w") as f:
json.dump(estado, f, ensure_ascii=False, indent=2)
print(f"Sesión guardada en {archivo}")
def cargar_sesion(self, archivo: str):
"""Carga una sesión previa."""
path = Path(archivo)
if not path.exists():
print(f"No se encontró sesión en {archivo}")
return
with open(archivo) as f:
estado = json.load(f)
self.historial = estado["historial"]
self.perfil = estado["perfil"]
self._resumen = estado["resumen"]
edad = (time.time() - estado["timestamp"]) / 3600
print(f"Sesión cargada ({edad:.1f}h de antigüedad, {len(self.historial)} mensajes)")
# Uso:
import time
asistente = AsistentePersistente("Bot", "Python")
asistente.chatear("Hola, soy Roberto, trabajo con Django")
asistente.guardar_sesion("sesion_roberto.json")
# En otra sesión:
asistente2 = AsistentePersistente("Bot", "Python")
asistente2.cargar_sesion("sesion_roberto.json")
resp = asistente2.chatear("¿Recuerdas mi nombre?")
print(resp) # Debería recordar que es RobertoResumen
- Memoria selectiva: Los últimos N mensajes siempre incluidos. Simple y efectivo para conversaciones cortas.
- Summarization del historial: Comprimir mensajes antiguos cuando el historial crece. Preservar explícitamente datos críticos.
- Relevance-based selection: Usar embeddings para seleccionar mensajes por similitud semántica con la query actual.
- Context pruning: Combinar recientes + relevantes respetando el presupuesto de tokens.
- Perfil de usuario: Construir y mantener un perfil que personaliza las respuestas sin ocupar mucho contexto.