@st.cache_data, @st.cache_resource, TTL e invalidación de caché

Por qué necesitamos caché en Streamlit

El modelo de ejecución top-down de Streamlit implica que cada rerun re-ejecuta el script completo de arriba a abajo. Sin mecanismos de caché, todas las operaciones se repiten en cada interacción del usuario, incluyendo las más costosas: lectura de archivos CSV grandes, consultas a bases de datos, llamadas a APIs externas y carga de modelos de machine learning.

import streamlit as st
import pandas as pd

# Sin caché: lee el CSV en CADA interacción del usuario
df = pd.read_csv("datos_grandes.csv")  # Puede tardar 5 segundos

opcion = st.selectbox("Filtrar por región", df["region"].unique())
st.dataframe(df[df["region"] == opcion])

Con @st.cache_data, el CSV se lee una sola vez y el resultado se almacena en caché:

import streamlit as st
import pandas as pd

@st.cache_data
def cargar_datos():
    return pd.read_csv("datos_grandes.csv")

df = cargar_datos()  # Rápido en reruns posteriores: retorna el DataFrame cacheado

opcion = st.selectbox("Filtrar por región", df["region"].unique())
st.dataframe(df[df["region"] == opcion])

El siguiente diagrama resume cuándo usar cada decorador de caché y sus implicaciones de memoria:

flowchart TB
    Call[Función llamada con args] --> Decide{Tipo de retorno}
    Decide -->|DataFrame, list, dict, numpy| CData["@st.cache_data"]
    Decide -->|DB connection, modelo ML, cliente API| CRes["@st.cache_resource"]
    CData --> CDSer[Serializa con pickle]
    CDSer --> CDCopy[Devuelve copia independiente por llamada]
    CDCopy --> CDTTL["TTL, max_entries, show_spinner"]
    CRes --> CRSingleton[Mismo objeto compartido entre sesiones]
    CRSingleton --> CRNoCopy["No serializa, no copia"]
    CRNoCopy --> CRUso["Uso: engines SQL, LLM clients, pipelines"]

@st.cache_data: caché para datos serializables

@st.cache_data es el decorador principal para funciones que retornan datos serializables: DataFrames, arrays NumPy, listas, diccionarios, strings y números. Streamlit serializa el resultado con pickle y guarda una copia en memoria. Cada llamada recibe su propia copia del resultado cacheado, lo que evita mutaciones accidentales entre diferentes partes del script.

La caché se indexa por los argumentos de la función: si se llama a la misma función con argumentos diferentes, se almacenan entradas de caché separadas para cada combinación de argumentos:

import streamlit as st
import pandas as pd
import numpy as np

# TTL en segundos (los datos expiran y se recargan)
@st.cache_data(ttl=3600)  # Caché de 1 hora
def cargar_ventas_desde_api(año: int, region: str) -> pd.DataFrame:
    # Simula llamada a API
    np.random.seed(año)
    return pd.DataFrame({
        "mes": range(1, 13),
        "ventas": np.random.randint(30000, 80000, 12),
        "region": region
    })

# TTL como timedelta
from datetime import timedelta

@st.cache_data(ttl=timedelta(hours=6))
def cargar_dataset_grande(url: str) -> pd.DataFrame:
    return pd.read_csv(url)

# max_entries: limitar el número de entradas en caché
@st.cache_data(max_entries=10, ttl=600)
def procesar_archivo(nombre_archivo: str) -> dict:
    # Costoso: solo se ejecuta una vez por combinación única de argumentos
    return {"resultado": "procesado", "archivo": nombre_archivo}

# Uso en la app
año = st.select_slider("Año", [2022, 2023, 2024, 2025, 2026])
region = st.selectbox("Región", ["Norte", "Sur", "Este"])

df = cargar_ventas_desde_api(año, region)  # Cacheado por (año, region)
st.line_chart(df, x="mes", y="ventas")

@st.cache_resource: singleton para recursos pesados

@st.cache_resource está diseñado para funciones que retornan recursos no serializables que deben existir como instancia única (singleton) compartida entre todas las sesiones y reruns. A diferencia de @st.cache_data, que crea copias independientes para cada llamada, @st.cache_resource devuelve siempre el mismo objeto en memoria.

Los casos de uso típicos son conexiones a bases de datos, modelos de ML cargados en memoria (que pueden ocupar varios GB), clientes de APIs externas y cualquier recurso cuya inicialización sea costosa y cuyo estado deba ser compartido:

import streamlit as st

# Modelo de ML: cargado UNA VEZ, compartido por TODAS las sesiones
@st.cache_resource
def cargar_modelo():
    import joblib
    return joblib.load("modelo_produccion.pkl")

# Conexión a base de datos: singleton, reutilizada
@st.cache_resource
def obtener_conexion_bd():
    import psycopg2
    return psycopg2.connect(
        host=st.secrets["db_host"],
        dbname=st.secrets["db_name"],
        user=st.secrets["db_user"],
        password=st.secrets["db_password"]
    )

# Cliente de API externa: singleton
@st.cache_resource
def cliente_openai():
    from openai import OpenAI
    return OpenAI(api_key=st.secrets["openai_api_key"])

modelo = cargar_modelo()
conn = obtener_conexion_bd()
client = cliente_openai()

Diferencias clave entre @st.cache_data y @st.cache_resource:

| Característica | @st.cache_data | @st.cache_resource | |---|---|---| | Tipo de retorno | Datos serializables (DF, arrays, dicts) | Recursos no serializables (conexiones, modelos) | | Aislamiento | Cada sesión/rerun recibe una copia | Todas las sesiones comparten el mismo objeto | | TTL soportado | Sí | Sí | | Ideal para | Cargar CSVs, APIs de datos, calcular | BD connections, modelos ML, clientes API |

Invalidar caché manualmente

import streamlit as st
import pandas as pd

@st.cache_data(ttl=300)
def cargar_datos_produccion():
    return pd.read_sql("SELECT * FROM ventas", conn)

df = cargar_datos_produccion()
st.dataframe(df)

col1, col2 = st.columns(2)
with col1:
    if st.button("Actualizar datos ahora"):
        st.cache_data.clear()           # Invalida TODA la caché de @st.cache_data
        st.rerun()
with col2:
    if st.button("Limpiar solo esta función"):
        cargar_datos_produccion.clear() # Invalida solo esta función
        st.rerun()

Spinner durante la carga inicial

import streamlit as st
import pandas as pd
import time

@st.cache_data(ttl=3600, show_spinner=False)  # Desactivar el spinner por defecto
def cargar_modelo_grande():
    time.sleep(3)  # Simular carga pesada
    return {"nombre": "ResNet50", "accuracy": 0.9431}

# Spinner personalizado mientras se carga la primera vez
with st.spinner("Cargando modelo (esto solo ocurre la primera vez)..."):
    modelo = cargar_modelo_grande()

st.success(f"Modelo '{modelo['nombre']}' listo (accuracy={modelo['accuracy']:.4f})")

Buenas prácticas con caché

1. Usa tipos inmutables como argumentos: strings, números, tuplas. Evita pasar DataFrames mutables como argumentos.
2. Activa show_spinner=True (por defecto) para informar al usuario durante la primera carga.
3. Establece TTL siempre que los datos puedan cambiar en el tiempo.
4. Llama a .clear() cuando el usuario actualiza datos para forzar la recarga.
5. No cachees funciones con efectos secundarios (escribir en BD, enviar emails).