Cap 6: Historial y Contexto de Conversación

El problema del historial en memoria

En el capítulo 4 guardamos el historial en un dict de Python. Esto tiene problemas:

• Se pierde cuando reinicias el bot
• Puede crecer indefinidamente consumiendo RAM
• No persiste entre deployments

La solución: SQLite para persistencia ligera sin servidor de base de datos.

storage/history.py con SQLite

import sqlite3
import json
from pathlib import Path
from datetime import datetime, timedelta

DB_PATH = Path("bot_history.db")
MAX_MESSAGES_PER_USER = 30
SESSION_TIMEOUT_HOURS = 24  # reset automático si inactivo por 24h

def _get_conn() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn

def init_db():
    with _get_conn() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS messages (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER NOT NULL,
                role TEXT NOT NULL,
                content TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        conn.execute("""
            CREATE INDEX IF NOT EXISTS idx_user_id ON messages(user_id, created_at)
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS user_settings (
                user_id INTEGER PRIMARY KEY,
                system_prompt TEXT,
                model TEXT,
                last_active TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

def get_history(user_id: int) -> list[dict]:
    """Retorna historial reciente del usuario, respetando timeout de sesión."""
    cutoff = datetime.utcnow() - timedelta(hours=SESSION_TIMEOUT_HOURS)

    with _get_conn() as conn:
        rows = conn.execute("""
            SELECT role, content FROM messages
            WHERE user_id = ? AND created_at > ?
            ORDER BY created_at DESC
            LIMIT ?
        """, (user_id, cutoff.isoformat(), MAX_MESSAGES_PER_USER)).fetchall()

    # Retornar en orden cronológico (de más antiguo a más reciente)
    return [{"role": r["role"], "content": r["content"]} for r in reversed(rows)]

def add_message(user_id: int, role: str, content: str):
    with _get_conn() as conn:
        conn.execute(
            "INSERT INTO messages (user_id, role, content) VALUES (?, ?, ?)",
            (user_id, role, content)
        )
        conn.execute("""
            INSERT INTO user_settings (user_id, last_active)
            VALUES (?, CURRENT_TIMESTAMP)
            ON CONFLICT(user_id) DO UPDATE SET last_active = CURRENT_TIMESTAMP
        """, (user_id,))

def clear_history(user_id: int):
    with _get_conn() as conn:
        conn.execute("DELETE FROM messages WHERE user_id = ?", (user_id,))

def get_stats(user_id: int) -> dict:
    with _get_conn() as conn:
        total = conn.execute(
            "SELECT COUNT(*) as n FROM messages WHERE user_id = ?", (user_id,)
        ).fetchone()["n"]
        last = conn.execute(
            "SELECT MAX(created_at) as t FROM messages WHERE user_id = ?", (user_id,)
        ).fetchone()["t"]
    return {"total_messages": total, "last_active": last}

Inicializar la DB al arrancar el bot:

# bot.py
from storage.history import init_db

def main():
    init_db()  # crear tablas si no existen
    app = Application.builder().token(TELEGRAM_TOKEN).build()
    # ...

Gestión de la ventana de contexto

Claude tiene un límite de tokens. Con mensajes largos, el historial puede exceder el límite. Estrategias:

1. Límite de mensajes (ya implementado)

LIMIT ? # últimos N mensajes

2. Resumen automático cuando el historial crece

from services.claude_api import ask_claude

async def resumir_historial(user_id: int) -> str:
    """Condensa el historial en un resumen para ahorrar tokens."""
    history = get_history(user_id)
    if len(history) < 10:
        return ""

    prompt_resumen = [
        {"role": "user", "content":
            f"Resume esta conversación en 3-5 puntos clave, "
            f"preservando contexto técnico importante:\n\n"
            + "\n".join(f"{m['role']}: {m['content'][:200]}" for m in history)
        }
    ]

    resumen = ask_claude(prompt_resumen)
    # Reemplazar historial con el resumen
    clear_history(user_id)
    add_message(user_id, "user", "[Resumen de conversación previa]")
    add_message(user_id, "assistant", resumen)
    return resumen

System prompts por usuario

Permite que cada usuario personalice el comportamiento de Claude:

# storage/history.py (agregar)

def get_user_system_prompt(user_id: int) -> str | None:
    with _get_conn() as conn:
        row = conn.execute(
            "SELECT system_prompt FROM user_settings WHERE user_id = ?", (user_id,)
        ).fetchone()
    return row["system_prompt"] if row else None

def set_user_system_prompt(user_id: int, prompt: str):
    with _get_conn() as conn:
        conn.execute("""
            INSERT INTO user_settings (user_id, system_prompt)
            VALUES (?, ?)
            ON CONFLICT(user_id) DO UPDATE SET system_prompt = ?
        """, (user_id, prompt, prompt))

# handlers/commands.py

async def cmd_persona(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """
    /persona <descripción> — personalizar el comportamiento de Claude
    Ejemplo: /persona Responde siempre en inglés y sé muy conciso
    """
    if not context.args:
        current = get_user_system_prompt(update.effective_user.id)
        await update.message.reply_text(
            f"System prompt actual:\n_{current or 'Por defecto'}_\n\n"
            "Uso: /persona <nueva descripción>",
            parse_mode="Markdown"
        )
        return

    prompt = " ".join(context.args)
    set_user_system_prompt(update.effective_user.id, prompt)
    await update.message.reply_text(f"Personalidad actualizada: _{prompt}_", parse_mode="Markdown")

Actualizar claude_api.py para usar el system prompt del usuario:

from storage.history import get_user_system_prompt

DEFAULT_SYSTEM = "Eres un asistente de programación experto..."

def ask_claude(messages: list[dict], user_id: int | None = None) -> str:
    system = DEFAULT_SYSTEM
    if user_id:
        custom = get_user_system_prompt(user_id)
        if custom:
            system = f"{DEFAULT_SYSTEM}\n\nInstrucciones adicionales: {custom}"

    response = client.messages.create(
        model=CLAUDE_MODEL,
        max_tokens=2048,
        system=system,
        messages=messages,
    )
    return response.content[0].text

Comando /stats

async def cmd_stats(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Muestra estadísticas del usuario."""
    user_id = update.effective_user.id
    stats = get_stats(user_id)
    history = get_history(user_id)

    await update.message.reply_text(
        f"*Tus estadísticas*\n\n"
        f"Mensajes totales: {stats['total_messages']}\n"
        f"En contexto actual: {len(history)}\n"
        f"Último activo: {stats['last_active'] or 'nunca'}\n",
        parse_mode="Markdown"
    )

Limpieza automática

Purga mensajes viejos para evitar que la DB crezca indefinidamente:

# storage/history.py

def purge_old_messages(days: int = 30):
    """Elimina mensajes más viejos que N días."""
    cutoff = (datetime.utcnow() - timedelta(days=days)).isoformat()
    with _get_conn() as conn:
        deleted = conn.execute(
            "DELETE FROM messages WHERE created_at < ?", (cutoff,)
        ).rowcount
    return deleted

# bot.py — job periódico

async def purge_job(context):
    n = purge_old_messages(days=30)
    logging.info(f"Purge: {n} mensajes eliminados")

app.job_queue.run_repeating(purge_job, interval=86400, first=10)  # cada 24h

Flujo de sesión completo

sequenceDiagram
    participant U as Usuario
    participant BOT as Bot
    participant DB as SQLite
    participant CL as Claude

    U->>BOT: mensaje
    BOT->>DB: get_history(user_id)
    Note over DB: últimos 30 msgs, últimas 24h
    DB-->>BOT: historial

    BOT->>CL: historial + nuevo mensaje
    CL-->>BOT: respuesta

    BOT->>DB: add_message(user, texto)
    BOT->>DB: add_message(assistant, respuesta)
    BOT->>U: respuesta