El Proyecto | Sobre nosotros | Contribuir | Donaciones | Licencia
HOME Mis cursos Finalizados
◄ Anterior
◄

Colas de Prioridad

Siguiente ►
►

Todas las colas que llevamos construidas comparten una regla: sale el más antiguo. Pero en TaskFlow hay momentos en que esa regla es injusta al revés: una tarea de prioridad 1 ("el servidor está caído") no puede esperar detrás de veinte tareas rutinarias que llegaron antes. Necesitamos una cola donde desencolar signifique "dame el más prioritario", no "dame el más antiguo". Ese TDA es la cola de prioridad, y cierra el segundo puente del módulo 3: la PilaConMinimo podía consultar el elemento mínimo en O(1), pero no extraerlo; hoy aprenderemos a extraerlo. Compararemos dos implementaciones honestas —lista no ordenada y lista ordenada, reutilizando el insertar_ordenado del módulo 2—, entenderemos sus costes enfrentados, y usaremos la herramienta profesional de Python, heapq, como caja negra, con especial atención a un detalle que separa el código correcto del código traicionero: los empates y la estabilidad.

Contenido

  1. El TDA cola de prioridad
  2. Implementación 1: lista no ordenada
  3. Implementación 2: lista ordenada (reutilizando insertar_ordenado)
  4. Tabla comparativa y el anuncio del montículo
  5. heapq: la caja negra profesional
  6. Empates y estabilidad: el truco del contador
  7. TaskFlow: la bandeja de urgencias

El TDA cola de prioridad

El contrato se parece al de la cola FIFO, pero cambia la promesa central:

Operación Cola FIFO Cola de prioridad
encolar(elemento) Entra por el final Entra "donde corresponda"
desencolar() Sale el más antiguo Sale el más prioritario
frente() Consulta el más antiguo Consulta el más prioritario
esta_vacia() / tamano() Igual Igual

En TaskFlow, prioridad 1 es la máxima, así que "el más prioritario" es el de menor número de prioridad: nuestra cola de prioridad es de mínimos (min-priority queue). Es un convenio frecuente (piensa en "prioridad 1" en soporte técnico) y encaja de fábrica con las herramientas de Python, que también trabajan con mínimos.

Fíjate en que el TDA no dice nada de cómo se logra: solo promete que desencolar devuelve el mínimo. Como siempre desde el módulo 1, un mismo contrato admite implementaciones con costes muy distintos — y esta vez la tensión entre ellas es especialmente instructiva: se puede pagar al entrar o pagar al salir, pero con listas se paga.

Implementación 1: lista no ordenada

La estrategia perezosa: encolar es soltar el elemento al final, sin orden ninguno; desencolar es buscar el mínimo y sacarlo.

class ColaPrioridadNoOrdenada:
    """Encolar O(1); desencolar O(n): busca el mínimo cada vez."""

    def __init__(self):
        self._elementos = []          # pares (prioridad, tarea), sin orden

    def encolar(self, prioridad, tarea):
        self._elementos.append((prioridad, tarea))     # O(1): al final y listo

    def desencolar(self):
        if self.esta_vacia():
            raise IndexError("desencolar sobre una cola vacía")
        # Buscar la posición del mínimo: recorrido completo, O(n)
        mejor = 0
        for i in range(1, len(self._elementos)):
            if self._elementos[i][0] < self._elementos[mejor][0]:
                mejor = i
        return self._elementos.pop(mejor)[1]           # pop(i) también O(n)

    def frente(self):
        if self.esta_vacia():
            raise IndexError("frente sobre una cola vacía")
        return min(self._elementos, key=lambda par: par[0])[1]   # O(n)

    def esta_vacia(self):
        return len(self._elementos) == 0

    def tamano(self):
        return len(self._elementos)

Análisis: encolar es O(1) —imbatible—, pero desencolar recorre toda la lista para localizar el mínimo (O(n)) y encima lo extrae con pop(mejor), que desplaza los elementos posteriores (otra vez O(n), viejo conocido del módulo 1). ¿Cuándo compensa? Cuando se encola muchísimo y se desencola muy poco: por ejemplo, si acumulas miles de candidatos pero solo extraerás unos pocos.

Implementación 2: lista ordenada (reutilizando insertar_ordenado)

La estrategia previsora: mantener la colección siempre ordenada por prioridad, de modo que el mínimo esté listo en el frente. Cuando en el módulo 2 escribimos insertar_ordenado sobre la ListaEnlazada, anunciamos que era "un puente hacia las colas de prioridad". Este es el momento de cruzarlo: si la lista enlazada se mantiene ordenada ascendentemente por prioridad, el más urgente está en la cabeza, y extraerlo es el eliminar_del_principio O(1) de la lección 04-02.

class ColaPrioridadOrdenada:
    """Encolar O(n) (inserción ordenada); desencolar O(1) (la cabeza)."""

    def __init__(self):
        self._elementos = ListaEnlazada()   # ordenada ascendente por prioridad

    def encolar(self, prioridad, tarea):
        # insertar_ordenado del módulo 2, guardando pares (prioridad, tarea):
        # recorre hasta el primer nodo con prioridad MAYOR y se inserta antes.
        nuevo = Nodo((prioridad, tarea))
        if (self._elementos.cabeza is None
                or prioridad < self._elementos.cabeza.dato[0]):
            nuevo.siguiente = self._elementos.cabeza       # nueva cabeza
            self._elementos.cabeza = nuevo
            if nuevo.siguiente is None:
                self._elementos.cola = nuevo
        else:
            actual = self._elementos.cabeza
            # Avanzar mientras el siguiente exista y no sea menos urgente.
            # OJO al <=: los empates quedan DETRÁS de los ya existentes
            # (esto preserva el orden de llegada entre iguales: estabilidad).
            while (actual.siguiente is not None
                   and actual.siguiente.dato[0] <= prioridad):
                actual = actual.siguiente
            nuevo.siguiente = actual.siguiente
            actual.siguiente = nuevo
            if nuevo.siguiente is None:
                self._elementos.cola = nuevo
        self._elementos.tamano += 1

    def desencolar(self):
        if self.esta_vacia():
            raise IndexError("desencolar sobre una cola vacía")
        return self._elementos.eliminar_del_principio()[1]   # O(1)

    def frente(self):
        if self.esta_vacia():
            raise IndexError("frente sobre una cola vacía")
        return self._elementos.cabeza.dato[1]                # O(1)

    def esta_vacia(self):
        return self._elementos.tamano == 0

    def tamano(self):
        return self._elementos.tamano

Ahora los costes se invierten: encolar recorre la lista hasta encontrar el hueco (O(n)), pero desencolar y frente son O(1). ¿Cuándo compensa? Cuando desencolas y consultas mucho más de lo que encolas — por ejemplo, un panel que consulta constantemente "¿cuál es la siguiente urgencia?".

Detalle fino que hemos dejado comentado en el código: el <= del bucle hace que una tarea nueva con la misma prioridad que otras existentes se coloque detrás de ellas. Entre iguales, orden de llegada: guardaremos este concepto —estabilidad— para la sección de empates, porque con heapq habrá que ganárselo a pulso.

Tabla comparativa y el anuncio del montículo

Implementación encolar desencolar frente ¿Cuándo elegirla?
Lista no ordenada O(1) O(n) O(n) Muchas inserciones, pocas extracciones
Lista ordenada (insertar_ordenado) O(n) O(1) O(1) Pocas inserciones, muchas extracciones/consultas
Montículo (heap) — módulo 6 O(log n) O(log n) O(1) El caso general: mezcla de ambas

La tabla revela un patrón que verás una y otra vez en estructuras de datos: dos soluciones simétricas que pagan O(n) en operaciones opuestas, y una tercera estructura más sofisticada que equilibra: ni O(1) ni O(n), sino O(log n) en ambas. Para hacerte una idea de lo que significa: con n = 1.000.000, O(n) son un millón de pasos y O(log n) son unos 20. Esa estructura es el montículo (heap), un árbol con una propiedad de orden muy astuta... y por eso vive en el módulo 6 (lección 06-07), después de que hayamos aprendido árboles. Aquí no vamos a abrirlo: vamos a usarlo.

heapq: la caja negra profesional

Python trae el montículo de serie en el módulo heapq. Lo usaremos como caja negra: sabemos qué promete (mínimo fuera, O(log n) por operación) sin mirar todavía cómo lo consigue. Es exactamente la disciplina TDA del módulo 1 aplicada a una biblioteca real.

heapq no define una clase: ofrece funciones que operan sobre una list normal a la que mantienen la propiedad de montículo:

import heapq

pendientes = []                       # una list corriente hará de montículo

heapq.heappush(pendientes, 3)         # encolar: O(log n)
heapq.heappush(pendientes, 1)
heapq.heappush(pendientes, 2)

print(pendientes[0])                  # 1 → el mínimo SIEMPRE está en [0] (frente)
print(heapq.heappop(pendientes))      # 1 → desencolar: extrae el mínimo, O(log n)
print(heapq.heappop(pendientes))      # 2
print(heapq.heappop(pendientes))      # 3

Reglas de la caja negra:

  • heappush(lista, elemento) = encolar; heappop(lista) = desencolar (el mínimo); lista[0] = frente; len(lista) = tamano.
  • La lista solo debe tocarse con funciones de heapq (o leerse en [0]). Un append o un sort por tu cuenta rompe la propiedad interna y el mínimo deja de estar garantizado.
  • Curiosidad que conecta con el módulo 6: aunque el interior de pendientes es una list aparentemente desordenada (¡imprímela!), su estructura codifica un árbol. Que un array pueda "ser" un árbol es una de las ideas bonitas que allí desvelaremos.

¿Y si los elementos no son números sueltos? heapq compara los elementos entre sí con <. Con tuplas, Python compara componente a componente, así que la costumbre es encolar tuplas cuyo primer campo sea la prioridad: (prioridad, tarea). Y aquí aparece una trampa seria.

Empates y estabilidad: el truco del contador

Intentemos encolar tareas (nuestros dict) con su prioridad:

import heapq

urgencias = []
t1 = {"id": 1, "titulo": "Reiniciar servidor", "prioridad": 1, "estado": "pendiente"}
t2 = {"id": 2, "titulo": "Avisar a clientes", "prioridad": 1, "estado": "pendiente"}

heapq.heappush(urgencias, (t1["prioridad"], t1))
heapq.heappush(urgencias, (t2["prioridad"], t2))   # ¡TypeError!

TypeError: '<' not supported between instances of 'dict' and 'dict'. ¿Por qué? Ambas tuplas empatan en el primer campo (1 == 1), así que Python pasa a comparar el segundo... y los dict no saben compararse con <. El empate rompe el programa.

Y aunque los elementos fueran comparables, quedaría un problema más sutil: entre dos tareas de prioridad 1, ¿cuál debe salir primero? Lo justo —y lo que espera cualquier usuario— es la que llegó antes: a igual prioridad, FIFO. Esa propiedad se llama estabilidad, y heapq por sí solo no la garantiza (el orden interno del montículo no recuerda llegadas).

El truco canónico resuelve ambas cosas a la vez: encolar tuplas de tres campos, (prioridad, contador, tarea), donde contador es un entero que crece con cada inserción:

  • Si las prioridades empatan, se compara el contador, que nunca empata → jamás se llega a comparar los dict (adiós TypeError).
  • El contador menor corresponde a la llegada más antigua → los empates salen por orden de llegada (estabilidad garantizada).
import heapq
from itertools import count

urgencias = []
contador = count()        # count() genera 0, 1, 2, ... uno nuevo en cada next()

heapq.heappush(urgencias, (t1["prioridad"], next(contador), t1))
heapq.heappush(urgencias, (t2["prioridad"], next(contador), t2))   # ahora sí

prioridad, _, tarea = heapq.heappop(urgencias)
print(tarea["titulo"])    # "Reiniciar servidor": llegó antes que su empate

Este patrón (prioridad, contador, elemento) es tan estándar que lo encontrarás tal cual en la documentación oficial de Python y en código de producción. Memorízalo como se memoriza un idiom.

TaskFlow: la bandeja de urgencias

Empaquetemos el patrón en la pieza de TaskFlow de esta lección: la bandeja de urgencias, donde se vuelcan tareas de cualquier procedencia y siempre se atiende la más urgente (recuerda: prioridad 1 = máxima), con empates resueltos por llegada:

import heapq
from itertools import count

class BandejaUrgencias:
    """Cola de prioridad de tareas de TaskFlow sobre heapq.

    desencolar() devuelve la tarea de menor número de prioridad;
    a igual prioridad, la que se encoló antes (estable).
    """

    def __init__(self):
        self._monticulo = []
        self._contador = count()

    def encolar(self, tarea):
        entrada = (tarea["prioridad"], next(self._contador), tarea)
        heapq.heappush(self._monticulo, entrada)          # O(log n)

    def desencolar(self):
        if self.esta_vacia():
            raise IndexError("desencolar sobre una bandeja vacía")
        return heapq.heappop(self._monticulo)[2]          # O(log n); [2] = tarea

    def frente(self):
        if self.esta_vacia():
            raise IndexError("frente sobre una bandeja vacía")
        return self._monticulo[0][2]                      # O(1)

    def esta_vacia(self):
        return len(self._monticulo) == 0

    def tamano(self):
        return len(self._monticulo)


# --- Uso ---
bandeja = BandejaUrgencias()
bandeja.encolar({"id": 4, "titulo": "Actualizar documentación",
                 "prioridad": 3, "estado": "pendiente"})
bandeja.encolar({"id": 5, "titulo": "Servidor caído",
                 "prioridad": 1, "estado": "pendiente", "asignada_a": "ana"})
bandeja.encolar({"id": 6, "titulo": "Cliente sin acceso",
                 "prioridad": 1, "estado": "pendiente", "asignada_a": "luis"})
bandeja.encolar({"id": 7, "titulo": "Revisar estilos CSS",
                 "prioridad": 2, "estado": "pendiente"})

while not bandeja.esta_vacia():
    tarea = bandeja.desencolar()
    print(f"prioridad {tarea['prioridad']}: {tarea['titulo']}")

# prioridad 1: Servidor caído          ← empate a 1: sale la que llegó antes
# prioridad 1: Cliente sin acceso
# prioridad 2: Revisar estilos CSS
# prioridad 3: Actualizar documentación

La salida resume la lección entera: manda la prioridad, no la llegada (la tarea 5 adelanta a la 4); y entre empates manda la llegada (la 5 antes que la 6). El montículo hace ambas cosas en O(log n) por operación, y en el módulo 6 abriremos por fin la caja para ver el árbol que lo hace posible.

Errores Comunes y Consejos

  • Encolar (prioridad, tarea) sin contador: funciona en las pruebas... hasta el primer empate con dicts, y entonces TypeError en producción. Usa siempre (prioridad, contador, elemento) cuando el elemento no sea comparable (y aunque lo sea: ganas estabilidad).
  • Confundir el sentido de la prioridad: heapq extrae el mínimo. Con nuestro convenio (1 = máxima) encaja directo; si tu convenio fuera "número mayor = más urgente", tendrías que encolar (-prioridad, contador, tarea). Documenta el convenio del proyecto y no lo mezcles.
  • Tocar la lista del montículo por fuera: un lista.append(x) o lista.sort() sobre el montículo rompe su invariante interno silenciosamente: no falla en el momento, falla después, devolviendo mínimos incorrectos. La lista del montículo es territorio exclusivo de heapq (por eso BandejaUrgencias la esconde tras _).
  • Reordenar tras cambiar la prioridad de una tarea ya encolada: mutar tarea["prioridad"] no recoloca su entrada en el montículo. El patrón profesional es encolar una entrada nueva y marcar la vieja como invalidada (o reconstruir la cola si es pequeña). Trabajaremos esta idea en los proyectos del módulo 8.
  • Consejo: si dudas entre lista ordenada, no ordenada o heapq, cuenta operaciones: ¿dominan las inserciones, las extracciones o van a la par? La tabla comparativa de esta lección es literalmente una guía de decisión; tenla a mano.

Ejercicios

Ejercicio 1: predecir con empates

Sin ejecutar, indica en qué orden salen los títulos si se encolan en una BandejaUrgencias estas tareas (en este orden) y luego se desencola todo: ("Backup", 2), ("Incendio", 1), ("Informe", 2), ("Rescate", 1). Justifica cada posición con la regla correspondiente (prioridad o estabilidad).

Ejercicio 2: desencolar estable en la lista NO ordenada

La ColaPrioridadNoOrdenada de esta lección no es estable: ante un empate, devuelve el primero que encuentra, lo cual solo es correcto por accidente. Modifica desencolar para que la búsqueda del mínimo use < estricto (no <=) al comparar, y razona por qué eso —dado que append añade al final— garantiza que entre empates sale el más antiguo.

Ejercicio 3: fusionar dos bandejas

Los equipos "web" y "móvil" de TaskFlow mantienen bandejas de urgencias separadas y se fusionan en un solo equipo. Escribe una función fusionar(bandeja_a, bandeja_b) que devuelva una BandejaUrgencias nueva con todas las tareas de ambas, conservando el orden por prioridad. ¿Se conserva también la estabilidad entre bandejas? Razona la respuesta.

Soluciones

Solución 1: Orden de salida: Incendio, Rescate, Backup, Informe.

  1. Incendio (prioridad 1): mínimo global.
  2. Rescate (prioridad 1): empata con Incendio, pero Incendio llegó antes (contador menor) → estabilidad.
  3. Backup (prioridad 2): menor que 2 ya no queda; empata con Informe y llegó antes.
  4. Informe (prioridad 2): último por estabilidad.

Solución 2:

def desencolar(self):
    if self.esta_vacia():
        raise IndexError("desencolar sobre una cola vacía")
    mejor = 0
    for i in range(1, len(self._elementos)):
        if self._elementos[i][0] < self._elementos[mejor][0]:   # < ESTRICTO
            mejor = i
    return self._elementos.pop(mejor)[1]

Razonamiento: append coloca cada elemento nuevo detrás de los anteriores, así que dentro de la lista los empates están en orden de llegada. La búsqueda con < estricto solo cambia de candidato ante un elemento estrictamente mejor; ante un empate se queda con el que ya tenía, que por recorrer de izquierda a derecha es el de índice menor = el más antiguo. Con <= ocurriría lo contrario: se quedaría el último empate, invirtiendo el orden de llegada.

Solución 3:

def fusionar(bandeja_a, bandeja_b):
    resultado = BandejaUrgencias()
    for bandeja in (bandeja_a, bandeja_b):
        while not bandeja.esta_vacia():
            resultado.encolar(bandeja.desencolar())
    return resultado

El orden por prioridad se conserva siempre: cada tarea se reencola con su prioridad y el montículo nuevo las ordenará. La estabilidad se conserva dentro de cada bandeja (desencolamos en orden estable y reencolamos en ese mismo orden, con contadores nuevos crecientes), pero entre bandejas no hay garantía histórica: todas las tareas de bandeja_a reciben contadores menores que las de bandeja_b, de modo que en los empates entre bandejas siempre gana la bandeja a, no la tarea que realmente llegó antes en el tiempo. Para una fusión históricamente justa habría que haber guardado una marca de tiempo global en cada tarea. Nota la mejora respecto a versiones ingenuas: la función consume las bandejas originales; si quisieras conservarlas, deberías reencolar también en las originales (rotación) o exponer una copia.

Conclusión

La cola de prioridad cambia la promesa del desencolar: ya no sale el más antiguo, sino el más urgente — en TaskFlow, el de menor número de prioridad. Hemos visto el dilema de las implementaciones con lista (no ordenada: encolar O(1) / desencolar O(n); ordenada con insertar_ordenado: encolar O(n) / desencolar O(1)), y hemos usado la solución equilibrada de la biblioteca estándar, heapq, con el idiom (prioridad, contador, tarea) que evita el TypeError de los empates y garantiza estabilidad: a igual urgencia, orden de llegada. El montículo que hay dentro de heapq —y por qué logra O(log n)— es cita pendiente para la lección 06-07, cuando sepamos de árboles. Nos queda un puente del módulo 3 por cruzar, el primero de todos: aquel HistorialConLimite que necesitaba eficiencia por los dos extremos a la vez. La estructura que lo consigue —la cola doble o deque— y la joya de la biblioteca estándar que la implementa, collections.deque, nos esperan en la próxima lección.

◄ Anterior
◄
Siguiente ►
►
El Proyecto | Sobre nosotros | Contribuir | Donaciones | Licencia
© Copyright 2026. Todos los derechos reservados