Comment surveiller les tâches d’IA planifiées pour qu’elles ne tournent pas en boucle?

Une tâche d’IA planifiée est un programme qui appelle un modèle d’IA selon un horaire : une revue de presse le matin, un résumé nocturne des demandes de soutien, une vérification horaire d’une boîte de réception. Une fois qu’elle tourne, personne ne la surveille. Si un bogue la fait s’exécuter chaque minute au lieu de chaque jour, ou si un agent qu’elle contient continue d’appeler des outils sans jamais terminer, elle peut se répéter pendant des heures et dépenser de l’argent bien réel avant que quelqu’un y jette un œil.

Surveiller, c’est répondre à trois questions sans avoir à penser à les poser : la tâche s’est-elle exécutée? S’est-elle bien comportée? Combien a-t-elle coûté? Les réponses viennent d’un journal que vous pouvez lire, de limites que la tâche vérifie avant de démarrer et d’alertes qui vous joignent quand quelque chose cloche.

Les quatre parties d’un dispositif de surveillance

Un jeton est un petit morceau de texte; les API d’IA facturent selon le nombre de jetons envoyés et reçus, si bien que le nombre de jetons est la mesure la plus directe des dépenses.

Un exemple concret : un garde-fou autour de n’importe quelle tâche

Le script ci-dessous enveloppe une fonction de tâche, run_job, qui représente votre propre code qui appelle le modèle. Elle renvoie le nombre d’étapes effectuées et le nombre de jetons utilisés. Le garde-fou fait le reste. Le prix et les limites sont fictifs; fixez-les d’après la grille tarifaire de votre fournisseur et votre propre budget.

import json, os, time, urllib.request
from datetime import datetime, timezone
from pathlib import Path

LOG = Path("runs.jsonl")
STOP = Path("STOP")                    # créez ce fichier pour bloquer toutes les exécutions
MAX_RUNS_PER_DAY = 3
MAX_STEPS = 10                         # transmis à la tâche; les agents doivent le respecter
MONTHLY_BUDGET = 20.00                 # dollars, à titre d’exemple
PRICE_PER_M_TOKENS = 5.00              # prix moyen fictif; utilisez le vrai prix de votre modèle
ALERT_URL = os.environ.get("ALERT_WEBHOOK")   # un webhook de clavardage ou de courriel que vous contrôlez

def alert(text):
    print("ALERTE :", text)
    if ALERT_URL:
        data = json.dumps({"text": text}).encode()
        urllib.request.urlopen(urllib.request.Request(
            ALERT_URL, data=data, headers={"Content-Type": "application/json"}), timeout=10)

def past_runs():
    if not LOG.exists():
        return []
    return [json.loads(line) for line in LOG.read_text().splitlines() if line]

def guarded_run(run_job):
    now = datetime.now(timezone.utc)
    runs = past_runs()
    today = [r for r in runs if r["start"][:10] == now.date().isoformat()]
    month_cost = sum(r.get("cost", 0) for r in runs if r["start"][:7] == now.strftime("%Y-%m"))

    if STOP.exists():
        return print("Arrêt d’urgence activé; aucune exécution.")
    if len(today) >= MAX_RUNS_PER_DAY:
        return alert(f"Plafond atteint : {len(today)} exécutions aujourd’hui. Vérifiez l’horaire.")
    if month_cost >= MONTHLY_BUDGET:
        return alert(f"Budget mensuel épuisé : {month_cost:.2f} $. Tâche suspendue.")

    entry = {"start": now.isoformat(), "status": "ok"}
    t0 = time.time()
    try:
        steps, tokens = run_job(max_steps=MAX_STEPS)
        entry.update(steps=steps, tokens=tokens, cost=tokens / 1e6 * PRICE_PER_M_TOKENS)
        if steps >= MAX_STEPS:
            entry["status"] = "hit_step_limit"
            alert(f"La tâche a utilisé ses {MAX_STEPS} étapes; elle tourne peut-être en boucle.")
    except Exception as e:
        entry.update(status="error", error=str(e)[:300])
        alert(f"Échec de la tâche : {e}")
    entry["seconds"] = round(time.time() - t0, 1)
    with LOG.open("a") as f:
        f.write(json.dumps(entry) + "\n")
    if month_cost + entry.get("cost", 0) > 0.8 * MONTHLY_BUDGET:
        alert("Plus de 80 % du budget mensuel d’IA est utilisé.")

Remarquez ce que le garde-fou vérifie avant d’appeler le modèle : l’arrêt d’urgence, le nombre d’exécutions et le budget. Une boucle causée par un mauvais horaire frappe le plafond à sa quatrième tentative de la journée, pas à sa quatre-centième.

Surveiller l’exécution qui n’a pas eu lieu

Les boucles font du bruit; le silence est la panne la plus discrète. Un horaire peut s’arrêter sans erreur. La documentation de GitHub, par exemple, note que les flux de travail planifiés peuvent être retardés quand son service est occupé, surtout au début de chaque heure, que certaines tâches en file d’attente peuvent être abandonnées et que, dans les dépôts publics, ils sont désactivés après 60 jours sans activité.

Ajoutez donc une deuxième vérification, distincte, qui s’exécute une fois par jour et vous alerte si runs.jsonl ne contient aucune ligne au statut ok dans les 26 dernières heures. Ces deux heures de plus absorbent les retards normaux sans masquer un véritable arrêt. Les ingénieurs appellent cela un dispositif de l’homme mort (dead man’s switch) : l’alarme sonne quand un signal attendu manque.

Des limites au niveau du planificateur

Si votre tâche tourne sur GitHub Actions, deux réglages ajoutent une deuxième couche de protection qui fonctionne même quand votre code se comporte mal. timeout-minutes annule une tâche qui dure trop longtemps; sans lui, GitHub permet 360 minutes. concurrency ne laisse s’exécuter qu’une seule tâche à la fois dans un groupe nommé, pour qu’une exécution lente ne puisse pas s’empiler derrière elle-même.

on:
  schedule:
    - cron: "17 6 * * *"     # 06:17 UTC chaque jour, loin du début de l'heure
  workflow_dispatch:
concurrency:
  group: ai-digest
jobs:
  digest:
    if: vars.AI_JOB_ENABLED == 'true'   # une variable du dépôt comme arrêt d'urgence
    runs-on: ubuntu-latest
    timeout-minutes: 10

Comme l’exécuteur repart à neuf chaque fois, stockez runs.jsonl dans un endroit qui persiste, comme une petite base de données ou un fichier que votre tâche renvoie dans le dépôt par un commit. Pour tout arrêter, mettez la variable à false, ou désactivez le flux de travail dans l’onglet Actions ou avec gh workflow disable.

Quelles limites chaque routine devrait-elle avoir?

Chaque routine a besoin d’un plafond sur la fréquence à laquelle elle peut s’exécuter, d’un plafond sur la quantité de travail qu’une exécution peut faire et d’un plafond sur ce qu’elle peut dépenser. Pour la fréquence, utilisez un nombre quotidien d’exécutions légèrement supérieur à l’horaire prévu. Pour chaque exécution, utilisez une limite d’étapes pour les agents, une longueur de sortie maximale pour les appels simples et un délai maximal. Pour les dépenses, utilisez à la fois un budget mensuel dans votre propre code et la limite de dépenses ou l’alerte budgétaire que la plupart des fournisseurs d’API offrent dans les paramètres du compte, pour qu’un bogue dans votre code ne puisse pas la contourner.

Ajoutez un arrêt d’urgence que vous pouvez atteindre depuis votre téléphone, comme une variable du dépôt ou un fichier sur un disque partagé, et testez-le avant d’en avoir besoin. Enfin, passez le journal en revue chaque semaine pendant le premier mois. Un nombre de jetons, des durées d’exécution et des coûts stables sont le signe qu’une routine est en bonne santé; une hausse soudaine mérite un coup d’œil même si aucune alerte ne s’est déclenchée.