Gestion d'erreurs et fallbacks — Outils API : Search, Files et Code Interpreter

Gestion d’erreurs et fallbacks

Le function calling introduit une difficulté que le code classique ne connaît pas : quand votre fonction échoue, il y a deux destinataires de l’erreur. Votre application, comme d’habitude — et le modèle, qui attend un résultat pour poursuivre. L’oublier produit le symptôme le plus déroutant de la mise en production : une conversation qui s’interrompt sans explication, parce qu’une exception a été avalée côté serveur et que le modèle attend toujours.

La règle qui découle de là structure toute la leçon : renvoyez toujours un function_call_output, même en cas d’échec. Un modèle informé qu’une API est indisponible peut proposer une alternative, demander une précision ou expliquer la situation à l’utilisateur. Un modèle qui ne reçoit rien ne peut rien faire.

Renvoyer les erreurs au modèle

La forme de l’erreur compte autant que le fait de la renvoyer. « Erreur » tout court n’apprend rien au modèle ; un objet JSON qui distingue le type d’erreur, sa nature transitoire ou définitive et la marche à suivre lui permet de décider. C’est un point contre-intuitif : on écrit ici des messages d’erreur destinés à être lus par une machine qui raisonne, pas journalisés.

import json
from openai import OpenAI

client = OpenAI()

def executer_fonction(name: str, arguments: str, call_id: str) -> dict:
    try:
        args = json.loads(arguments)
        resultat = dispatcher[name](**args)
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps(resultat)
        }
    except KeyError:
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({
                "error": "fonction_inconnue",
                "message": f"La fonction '{name}' n'existe pas"
            })
        }
    except json.JSONDecodeError:
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({
                "error": "arguments_invalides",
                "message": "Les arguments JSON sont malformes"
            })
        }
    except Exception as e:
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({
                "error": "erreur_interne",
                "message": str(e)
            })
        }

Retry avec backoff exponentiel

Le retry ne vaut que pour ce qui a une chance d’aboutir au deuxième essai : coupure réseau, 429, indisponibilité passagère. Réessayer une requête refusée pour authentification invalide ou paramètre incorrect ne fait que retarder l’échec en multipliant les appels. D’où le tri explicite dans le code ci-dessous — c’est lui qui fait le travail, le backoff n’étant que la politesse qu’on doit au service d’en face.

import time
import random

def executer_avec_retry(func, args, max_retries=3, base_delay=1.0):
    """Retry avec backoff exponentiel et jitter."""
    for tentative in range(max_retries + 1):
        try:
            return func(**args)
        except (ConnectionError, TimeoutError) as e:
            if tentative == max_retries:
                raise
            delay = base_delay * (2 ** tentative) + random.uniform(0, 0.5)
            time.sleep(delay)
        except ValueError:
            # Erreur de données : pas de retry
            raise

def executer_fonction_robuste(name, arguments, call_id):
    args = json.loads(arguments)
    try:
        resultat = executer_avec_retry(dispatcher[name], args)
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps(resultat)
        }
    except Exception as e:
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({
                "error": type(e).__name__,
                "message": str(e),
                "retries_effectues": 3
            })
        }

Timeouts par fonction

Un timeout unique pour toutes les fonctions est toujours mal réglé : trop court, il tue les traitements lourds ; trop long, il fait attendre l’utilisateur sur une simple consultation de cache. Calez chaque valeur sur la durée observée en conditions normales, avec une marge — et non sur le pire cas imaginable, qui reviendrait à n’avoir aucun timeout.

import signal
from functools import wraps

TIMEOUTS = {
    "rechercher_client": 5,
    "generer_rapport": 30,
    "envoyer_email": 10,
}

def avec_timeout(func, timeout_seconds):
    """Exécute une fonction avec un timeout."""
    import concurrent.futures
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
        future = executor.submit(func)
        try:
            return future.result(timeout=timeout_seconds)
        except concurrent.futures.TimeoutError:
            raise TimeoutError(
                f"Fonction interrompue apres {timeout_seconds}s"
            )

def executer_avec_timeout(name, arguments, call_id):
    args = json.loads(arguments)
    timeout = TIMEOUTS.get(name, 10)  # 10s par defaut

    try:
        resultat = avec_timeout(
            lambda: dispatcher[name](**args),
            timeout
        )
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps(resultat)
        }
    except TimeoutError as e:
        return {
            "type": "function_call_output",
            "call_id": call_id,
            "output": json.dumps({
                "error": "timeout",
                "message": str(e),
                "suggestion": "Réessayez avec des critères plus précis"
            })
        }

Fallback entre fonctions

Le fallback est un arbitrage, pas un réflexe : servir une donnée de cache vieille de six heures vaut mieux que rien pour un tableau de bord, et beaucoup moins bien que rien pour un solde bancaire. Quand vous dégradez, dites-le — dans le résultat renvoyé au modèle, pour qu’il puisse en avertir l’utilisateur au lieu de présenter une donnée de secours comme une donnée fraîche.

async def rechercher_produit(terme: str) -> dict:
    """Recherche avec fallback : API principale -> cache -> base locale."""
    # Tentative 1 : API catalogue
    try:
        return await api_catalogue.rechercher(terme)
    except Exception:
        pass

    # Tentative 2 : cache Redis
    try:
        cached = await redis_client.get(f"produit:{terme}")
        if cached:
            resultat = json.loads(cached)
            resultat["_source"] = "cache"
            resultat["_avertissement"] = "Données potentiellement obsoletes"
            return resultat
    except Exception:
        pass

    # Tentative 3 : base locale SQLite
    try:
        row = db.execute(
            "SELECT * FROM produits WHERE nom LIKE ?",
            (f"%{terme}%",)
        ).fetchone()
        if row:
            return {
                "nom": row["nom"],
                "prix": row["prix"],
                "_source": "base_locale",
                "_avertissement": "Données de la dernière synchronisation"
            }
    except Exception:
        pass

    return {
        "error": "indisponible",
        "message": "Aucune source de données accessible pour cette recherche"
    }

Circuit breaker

Le circuit breaker répond à ce que le retry ne sait pas traiter : un service durablement hors service. Après N échecs, il coupe et renvoie l’erreur immédiatement, sans même essayer, puis laisse repasser une requête de temps en temps pour tester le retour à la normale. Le gain est double — vos utilisateurs cessent d’attendre un timeout à chaque requête, et le service en difficulté n’est pas maintenu à terre par vos réessais.

from datetime import datetime, timedelta

class CircuitBreaker:
    def __init__(self, seuil_echecs=5, duree_ouverture=60):
        self.seuil = seuil_echecs
        self.duree = duree_ouverture
        self.echecs = 0
        self.dernier_echec = None
        self.etat = "ferme"  # ferme, ouvert, semi-ouvert

    def peut_executer(self) -> bool:
        if self.etat == "ferme":
            return True
        if self.etat == "ouvert":
            if datetime.now() - self.dernier_echec > timedelta(seconds=self.duree):
                self.etat = "semi-ouvert"
                return True
            return False
        return True  # semi-ouvert : laisser passer un essai

    def enregistrer_succes(self):
        self.echecs = 0
        self.etat = "ferme"

    def enregistrer_echec(self):
        self.echecs += 1
        self.dernier_echec = datetime.now()
        if self.echecs >= self.seuil:
            self.etat = "ouvert"

# Un circuit breaker par service externe
breakers = {
    "api_meteo": CircuitBreaker(seuil_echecs=3, duree_ouverture=30),
    "api_catalogue": CircuitBreaker(seuil_echecs=5, duree_ouverture=60),
}

Boucle robuste complète

Le pattern complet rassemble les cinq mécanismes précédents. Une précaution le rend réellement sûr, et elle est facile à omettre : borner le nombre d’itérations. Un modèle qui rappelle en boucle une fonction dont le résultat ne le satisfait pas est rare, mais sans plafond il consomme le budget jusqu’à épuisement — et c’est le genre d’incident qu’on découvre sur la facture.

def boucle_robuste(prompt, tools, max_tours=10):
    messages = prompt
    for tour in range(max_tours):
        response = client.responses.create(
            model="gpt-5.6-terra",
            input=messages,
            tools=tools
        )

        appels = [i for i in response.output if i.type == "function_call"]
        if not appels:
            return response.output_text

        resultats = []
        for appel in appels:
            resultat = executer_fonction_robuste(
                appel.name, appel.arguments, appel.call_id
            )
            resultats.append(resultat)

        messages = response.output + resultats

    return "Workflow interrompu : limite de tours atteinte"

Points clés à retenir

Renvoyez toujours un function_call_output, même en cas d’erreur
Structurez les erreurs en JSON pour que le modèle les exploite
Retentez uniquement les erreurs transitoires avec backoff exponentiel
Implémentez des fallbacks entre sources de données
Un circuit breaker protège les services défaillants