A notificação chegou às 3 da manhã. Seu servidor de processamento de dados, que deveria estar rodando tranquilamente, bateu no teto de CPU. Uma cascata de tarefas falhou, o relatório de BI ficou comprometido e a sua única opção foi acordar e ‘dar um jeito’.
Se essa cena te soa familiar, você sabe que o agendamento de tarefas vai muito além de apenas “rodar um cron”. É uma arte, e a inteligência artificial está aqui para te dar um pincel novo. Pense no agendamento manual como um jogo de Jenga, onde cada tarefa é um bloco que deve ser posicionado com cuidado para que a torre (o servidor) não desabe. Se o seu objetivo é a perfeição, a IA, com algoritmos genéticos, nos permite simular milênios de evolução em segundos, encontrando a solução mais robusta e equilibrada.
Neste artigo, vamos explorar como aplicar essa técnica, inspirada na seleção natural, para resolver um problema comum: como agendar tarefas de forma inteligente para evitar sobrecarga e maximizar a eficiência do seu servidor.
O desafio NP-Difícil do agendamento inteligente
O agendamento de dezenas de tarefas, cada uma com suas restrições de tempo e recursos, não é um problema para um cérebro humano — é um pesadelo. Em termos técnicos, estamos lidando com um Problema de Otimização Combinatória, ou, na linguagem da ciência da computação, um problema NP-difícil. Isso significa que a quantidade de combinações possíveis cresce exponencialmente, tornando a busca pela solução perfeita (por força bruta) impraticável até mesmo para os supercomputadores mais potentes.
É nesse cenário que a natureza nos dá a resposta. Em vez de tentar todas as combinações, podemos evoluir uma população de soluções candidatas, descartando as ruins e aprimorando as boas ao longo de gerações.
Seu desafio: Suponha que você tenha uma lista de tarefas, cada uma com as seguintes características:
- Duração estimada (em minutos).
- Consumo de CPU e RAM (em percentual do total).
- Faixa horária permitida para execução.
Seu objetivo é agendar essas tarefas ao longo de um período (por exemplo, 24 horas), garantindo que:
- O agendamento total seja o mais curto e eficiente possível.
- O servidor nunca ultrapasse os 100% de CPU ou RAM em nenhum momento.
- As tarefas sejam executadas dentro de sua janela permitida.
Por que um Algoritmo Genético? A lógica da evolução
O Algoritmo Genético (GA) é uma técnica de otimização inspirada na seleção natural. Ele não busca a solução perfeita de forma exaustiva. Em vez disso, usa a “sabedoria da multidão” (da população de soluções) e a “lei da selva” (seleção natural) para convergir, de forma eficiente, para uma solução quase ótima.
Ele é uma excelente escolha para este problema porque:
- Representação Intuitiva: É fácil de representar uma solução, como um simples vetor de horários de início.
- Adaptação: Ele tolera soluções imperfeitas e busca melhorias iterativas, sem a necessidade de um modelo matemático complexo e rígido.
- Robustez: Pode lidar com restrições complicadas (como o limite de 100% de CPU/RAM) de forma flexível, usando penalidades.
A estrutura da nossa solução
Para implementar nosso GA, vamos precisar de três componentes principais:
- Representação do Indivíduo: Como uma solução candidata (um agendamento) é codificada.
- Função de Fitness: O “juiz” que avalia a qualidade de cada agendamento.
- Operadores Genéticos: Funções de seleção, crossover (recombinação) e mutação que evoluem a população.
Representação do problema e do indivíduo
Cada tarefa será um dicionário, e um indivíduo (solução candidata) será um vetor de horários de início.
# Dados de exemplo
tarefas = [
{"id": "tarefa1", "duracao": 60, "cpu": 50, "ram": 30, "janela": [0, 600]},
{"id": "tarefa2", "duracao": 30, "cpu": 30, "ram": 20, "janela": [100, 700]},
{"id": "tarefa3", "duracao": 45, "cpu": 60, "ram": 40, "janela": [200, 900]},
{"id": "tarefa4", "duracao": 90, "cpu": 40, "ram": 30, "janela": [0, 1440]},
{"id": "tarefa5", "duracao": 30, "cpu": 30, "ram": 20, "janela": [0, 1440]}
]Um indivíduo será algo como: [120, 30, 450, 200, 500], onde cada valor representa o horário de início da respectiva tarefa em minutos.
A função de fitness: O coração da solução
A Função de Fitness é o nosso juiz, que decide se um agendamento é bom ou ruim. Nosso objetivo é minimizar o fitness score. A função penaliza severamente qualquer agendamento que viole as regras do jogo e recompensa agendamentos mais curtos e eficientes.
import numpy as np
CPU_MAX = 100
RAM_MAX = 100
PERIODO_TOTAL = 1440 # 24h em minutos
def avaliar(individuo, tarefas):
"""
Avalia a qualidade de um agendamento (indivíduo).
Retorna uma pontuação de fitness, onde menor é melhor.
"""
uso_cpu = np.zeros(PERIODO_TOTAL)
uso_ram = np.zeros(PERIODO_TOTAL)
penalidade = 0
for idx, inicio in enumerate(individuo):
tarefa = tarefas[idx]
fim = inicio + tarefa["duracao"]
# Penalidade por agendar fora da janela de tempo permitida
if inicio < tarefa["janela"][0] or fim > tarefa["janela"][1]:
penalidade += 1000
# Mapeia o uso de recursos ao longo do tempo
# Garante que não ultrapasse o período total
segmento_fim = min(fim, PERIODO_TOTAL)
for t in range(inicio, segmento_fim):
uso_cpu[t] += tarefa["cpu"]
uso_ram[t] += tarefa["ram"]
# Penalidades por ultrapassar os limites do servidor
penalidade += np.sum(np.maximum(0, uso_cpu - CPU_MAX)) * 10
penalidade += np.sum(np.maximum(0, uso_ram - RAM_MAX)) * 10
# O "makespan" é o tempo que leva para a última tarefa terminar.
# Um makespan menor significa uma solução mais compacta.
makespan = 0
if individuo:
ultima_tarefa_idx = np.argmax(individuo)
makespan = individuo[ultima_tarefa_idx] + tarefas[ultima_tarefa_idx]["duracao"]
return penalidade + makespanO motor da evolução: Algoritmo Genético simples
Aqui está o código completo do nosso motor de otimização. Ele cria uma população de agendamentos aleatórios e, em cada geração, seleciona os melhores, os recombina (crossover), e introduz pequenas mutações para continuar explorando o espaço de soluções.
import random
POPULACAO_TAMANHO = 50
NUM_GERACOES = 100
TAXA_MUTACAO = 0.2
TAXA_ELITISMO = 0.2
def gerar_individuo(tarefas):
"""
Cria um agendamento inicial aleatório, respeitando as janelas de tempo.
"""
return [random.randint(t["janela"][0], t["janela"][1] - t["duracao"]) for t in tarefas]
def crossover(pai1, pai2):
"""
Combina dois agendamentos parentais para criar um novo.
"""
corte = random.randint(1, len(pai1) - 2)
filho = pai1[:corte] + pai2[corte:]
return filho
def mutacao(individuo, tarefas):
"""
Faz uma pequena alteração aleatória no agendamento.
"""
i = random.randint(0, len(tarefas) - 1)
tarefa = tarefas[i]
individuo[i] = random.randint(tarefa["janela"][0], tarefa["janela"][1] - tarefa["duracao"])
return individuo
def algoritmo_genetico(tarefas):
"""
Executa o algoritmo genético para encontrar o melhor agendamento.
"""
populacao = [gerar_individuo(tarefas) for _ in range(POPULACAO_TAMANHO)]
num_elite = int(POPULACAO_TAMANHO * TAXA_ELITISMO)
for geracao in range(NUM_GERACOES):
# Avalia e ordena a população por fitness (menor é melhor)
populacao.sort(key=lambda ind: avaliar(ind, tarefas))
# Elitismo: os melhores indivíduos sobrevivem para a próxima geração
nova_populacao = populacao[:num_elite]
while len(nova_populacao) < POPULACAO_TAMANHO:
# Seleção: escolhe pais para crossover (os melhores têm mais chance)
p1, p2 = random.sample(populacao[:POPULACAO_TAMANHO // 2], 2)
# Crossover: cria um novo filho
filho = crossover(p1, p2)
# Mutação: pequena chance de alterar o filho
if random.random() < TAXA_MUTACAO:
filho = mutacao(filho, tarefas)
nova_populacao.append(filho)
populacao = nova_populacao
if geracao % 10 == 0:
melhor_fitness = avaliar(populacao[0], tarefas)
print(f"Geração {geracao:3d} - Melhor fitness: {melhor_fitness:7.2f}")
# Retorna o melhor indivíduo encontrado
populacao.sort(key=lambda ind: avaliar(ind, tarefas))
return populacao[0]Visualizando a solução otimizada
Depois de executar o algoritmo, temos o nosso agendamento otimizado. Para entender o resultado, nada melhor do que uma visualização. Um diagrama de Gantt é a ferramenta perfeita para isso. Ele nos mostra a linha do tempo de cada tarefa, permitindo ver visualmente o quão bem o nosso algoritmo distribuiu a carga.
import matplotlib.pyplot as plt
# Executa o algoritmo
melhor_agendamento = algoritmo_genetico(tarefas)
print("\nMelhor agendamento encontrado:", melhor_agendamento)
print("Melhor fitness:", avaliar(melhor_agendamento, tarefas))
def visualizar_agendamento(agendamento, tarefas):
cores = ["#1f77b4", "#ff7f0e", "#2ca02c", "#d62728", "#9467bd"]
plt.style.use('seaborn-v0_8-whitegrid')
fig, ax = plt.subplots(figsize=(12, 6))
for i, inicio in enumerate(agendamento):
duracao = tarefas[i]["duracao"]
ax.barh(i, duracao, left=inicio, color=cores[i % len(cores)], edgecolor='black', height=0.6)
ax.text(inicio + 2, i, tarefas[i]["id"], va='center', ha='left', color='white', fontweight='bold')
# Adicionando visualizações para os limites de recursos
uso_cpu_plot = np.zeros(PERIODO_TOTAL)
uso_ram_plot = np.zeros(PERIODO_TOTAL)
for idx, inicio in enumerate(agendamento):
tarefa = tarefas[idx]
fim = inicio + tarefa["duracao"]
segmento_fim = min(fim, PERIODO_TOTAL)
for t in range(inicio, segmento_fim):
uso_cpu_plot[t] += tarefa["cpu"]
uso_ram_plot[t] += tarefa["ram"]
fig_uso, ax_uso = plt.subplots(figsize=(12, 4))
ax_uso.plot(uso_cpu_plot, label="Uso de CPU (%)", color="#1f77b4")
ax_uso.plot(uso_ram_plot, label="Uso de RAM (%)", color="#ff7f0e")
ax_uso.axhline(y=CPU_MAX, color='r', linestyle='--', label="Limite de CPU/RAM")
ax_uso.set_xlabel("Tempo (minutos)")
ax_uso.set_ylabel("Uso de Recursos (%)")
ax_uso.set_title("Uso de Recursos ao Longo do Tempo")
ax_uso.legend()
ax_uso.set_ylim(0, 150)
ax_uso.grid(True)
ax.set_xlabel("Tempo (minutos)")
ax.set_ylabel("Tarefas")
ax.set_title("Agendamento Otimizado de Tarefas")
ax.set_yticks(range(len(tarefas)))
ax.set_yticklabels([t["id"] for t in tarefas])
ax.grid(axis='x', linestyle='--', alpha=0.7)
plt.tight_layout()
plt.show()
visualizar_agendamento(melhor_agendamento, tarefas)
A imagem acima mostra o agendamento otimizado. Note como as barras se encaixam de forma inteligente, como peças de um quebra-cabeça, garantindo que nenhuma “linha vertical” (momento no tempo) exceda a capacidade do servidor. O gráfico de uso de recursos, por sua vez, confirma que nosso agendamento respeita os limites de CPU e RAM.
Conclusão: O futuro do agendamento inteligente
A otimização de agendamento de tarefas com IA é uma aplicação poderosa e prática de algoritmos genéticos. Com um modelo simples, conseguimos distribuir tarefas em uma linha do tempo respeitando restrições reais de recursos computacionais. Essa abordagem é extremamente útil para equipes de engenharia de dados, DevOps ou BI que lidam com execução crônica de tarefas pesadas, como as do Apache Airflow.
O potencial é imenso. Imagine um scheduler de Airflow que não só evita a sobrecarga, mas também se adapta em tempo real a novas tarefas ou falhas de execução. Esse é o futuro do agendamento inteligente, onde os algoritmos não apenas resolvem problemas, mas aprendem a antecipá-los.
Além deste problema, o algoritmo genético pode ser usado para encontrar uma combinação de hiperparâmetros para uma rede neural, como descrevo neste artigo.
Se você gostou da ideia, que tal começar a integrar essa lógica no seu fluxo de trabalho? Compartilhe, comente e me diga como você usaria essa ideia no seu ambiente!