Architectures multi-outils en production
Architectures multi-outils en production
Un prototype multi-outils fonctionne parce qu’une seule personne l’utilise, connaît ses limites et relance quand il échoue. La production supprime ces trois conditions d’un coup. Ce qui devient alors critique n’est pas la qualité des réponses — elle était déjà là — mais trois propriétés que le prototype n’avait pas : savoir ce qui s’est passé quand un appel échoue, empêcher qu’une boucle emballée vide le budget, et pouvoir modifier un comportement sans redéployer.
Les patterns qui suivent répondent chacun à l’une de ces exigences. Ils ne sont pas à adopter en bloc : prenez l’observabilité en premier, car sans elle vous ne saurez même pas lesquels des autres vous manquent.
Architecture en couches
La séparation en couches n’est pas une élégance d’architecte : elle rend chaque partie remplaçable. L’intérêt concret est qu’on peut changer d’orchestration sans toucher à l’interface, ou ajouter un outil sans réécrire la logique métier. Les quatre couches ci-dessous sont le découpage minimal qui tient à l’usage.
# Couche 1 : Interface - réception des requetes
# Couche 2 : Orchestration - gestion de la boucle et du routage
# Couche 3 : Execution - appels aux outils et fonctions
# Couche 4 : Observabilite - logs, metriques, traces
from openai import OpenAI
from dataclasses import dataclass, field
from datetime import datetime
import json
import time
import logging
logger = logging.getLogger("multi-outils")
@dataclass
class TraceExecution:
"""Trace d'execution pour le monitoring."""
requete_id: str
debut: float = field(default_factory=time.time)
tours: list = field(default_factory=list)
outils_utilises: list = field(default_factory=list)
tokens_total: int = 0
erreurs: list = field(default_factory=list)
duree_totale: float = 0.0Orchestrateur de production
class OrchestreurProduction:
def __init__(self, config: dict):
self.client = OpenAI()
self.config = config
self.tools = self._construire_tools(config)
self.dispatcher = self._construire_dispatcher(config)
def _construire_tools(self, config: dict) -> list:
"""Construit le tableau d'outils depuis la configuration."""
tools = []
if config.get("file_search"):
tools.append({
"type": "file_search",
"vector_store_ids": config["file_search"]["vector_stores"],
"max_num_results": config["file_search"].get("max_results", 10)
})
if config.get("web_search"):
tools.append({
"type": "web_search_preview",
"search_context_size": config["web_search"].get("context_size", "medium")
})
if config.get("code_interpreter"):
tools.append({"type": "code_interpreter"})
for func in config.get("fonctions", []):
tools.append(func)
return tools
def _construire_dispatcher(self, config: dict) -> dict:
"""Mappe les noms de fonctions a leurs implémentations."""
return config.get("implementations", {})
def executer(self, requete: str, fichiers: list = None,
max_tours: int = 10) -> dict:
"""Exécute une requête avec trace complete."""
trace = TraceExecution(
requete_id=f"req_{int(time.time()*1000)}"
)
content = [{"type": "text", "text": requete}]
if fichiers:
for fid in fichiers:
content.append({"type": "input_file", "file_id": fid})
messages = [{"role": "user", "content": content}]
try:
for tour in range(max_tours):
debut_tour = time.time()
response = self.client.responses.create(
model=self.config.get("model", "gpt-5.6-terra"),
input=messages,
tools=self.tools,
instructions=self.config.get("instructions", "")
)
trace.tokens_total += response.usage.total_tokens
# Identifier les appels de fonction
appels = [
item for item in response.output
if item.type == "function_call"
]
# Enregistrer les outils utilisés
for item in response.output:
if hasattr(item, "type"):
trace.outils_utilises.append(item.type)
trace.tours.append({
"numero": tour + 1,
"appels_fonction": len(appels),
"duree": time.time() - debut_tour
})
if not appels:
trace.duree_totale = time.time() - trace.debut
return {
"reponse": response.output_text,
"trace": trace
}
# Exécuter les fonctions personnalisées
resultats = self._executer_fonctions(appels, trace)
messages = response.output + resultats
except Exception as e:
trace.erreurs.append(str(e))
trace.duree_totale = time.time() - trace.debut
logger.error(f"Erreur requête {trace.requete_id}: {e}")
return {"reponse": None, "erreur": str(e), "trace": trace}
trace.duree_totale = time.time() - trace.debut
return {"reponse": "Limite de tours", "trace": trace}
def _executer_fonctions(self, appels: list, trace: TraceExecution) -> list:
"""Exécute les fonctions avec gestion d'erreurs."""
resultats = []
for appel in appels:
try:
func = self.dispatcher.get(appel.name)
if not func:
raise ValueError(f"Fonction inconnue : {appel.name}")
args = json.loads(appel.arguments)
resultat = func(**args)
resultats.append({
"type": "function_call_output",
"call_id": appel.call_id,
"output": json.dumps(resultat)
})
except Exception as e:
trace.erreurs.append(f"{appel.name}: {str(e)}")
resultats.append({
"type": "function_call_output",
"call_id": appel.call_id,
"output": json.dumps({"error": str(e)})
})
return resultatsConfiguration déclarative
Sortir la définition des systèmes du code pour la mettre en configuration change qui peut intervenir. Ajuster un prompt système, retirer un outil, modifier un plafond de coût devient une modification de fichier, relisible en revue et réversible — au lieu d’un déploiement. Sur un système qui évolue chaque semaine, c’est la différence entre itérer et subir.
config_assistant_commercial = {
"model": "gpt-5.6-terra",
"instructions": (
"Tu es un assistant commercial. Tu as accès aux données CRM, "
"a la documentation produit, au web et a l'analyse de données. "
"Priorise les données internes, enrichis avec le web."
),
"file_search": {
"vector_stores": ["vs_fiches_produit", "vs_propositions"],
"max_results": 10
},
"web_search": {
"context_size": "medium"
},
"code_interpreter": True,
"fonctions": [
{
"type": "function",
"name": "creer_opportunite",
"description": "Créer une opportunité dans le CRM",
"parameters": {
"type": "object",
"properties": {
"client": {"type": "string"},
"montant": {"type": "number"},
"probabilite": {"type": "integer"}
},
"required": ["client", "montant", "probabilite"],
"additionalProperties": False
},
"strict": True
}
],
"implementations": {
"creer_opportunite": lambda client, montant, probabilite:
crm_service.creer_opportunite(client, montant, probabilite)
}
}
orchestreur = OrchestreurProduction(config_assistant_commercial)
resultat = orchestreur.executer("Prepare une proposition pour Acme Corp")Observabilité et monitoring
Tableau de bord des métriques
Trois métriques suffisent pour commencer, et elles répondent chacune à une question différente : la latence par outil dit lequel ralentit le système, le taux d’échec par outil dit lequel est fragile, et le coût par requête dit lequel dérape. Une moyenne globale sur les trois ne répond à aucune de ces questions.
class MetriquesMultiOutils:
def __init__(self):
self.requetes = []
def enregistrer(self, trace: TraceExecution):
self.requetes.append({
"id": trace.requete_id,
"timestamp": datetime.now().isoformat(),
"duree": trace.duree_totale,
"tours": len(trace.tours),
"tokens": trace.tokens_total,
"outils": list(set(trace.outils_utilises)),
"erreurs": len(trace.erreurs),
"cout_estime": self._estimer_cout(trace)
})
def _estimer_cout(self, trace: TraceExecution) -> float:
"""Estime le coût en dollars de l'execution."""
# Tarifs approximatifs gpt-5.6-terra : 2 $ / 12 $ par million
cout_input = trace.tokens_total * 0.6 * 0.000002
cout_output = trace.tokens_total * 0.4 * 0.000012
return round(cout_input + cout_output, 4)
def rapport(self) -> dict:
if not self.requetes:
return {}
durees = [r["duree"] for r in self.requetes]
tokens = [r["tokens"] for r in self.requetes]
couts = [r["cout_estime"] for r in self.requetes]
return {
"total_requetes": len(self.requetes),
"duree_moyenne": sum(durees) / len(durees),
"tokens_moyen": sum(tokens) / len(tokens),
"cout_total": sum(couts),
"taux_erreur": sum(1 for r in self.requetes if r["erreurs"] > 0) / len(self.requetes),
"outils_les_plus_utilises": self._outils_frequents()
}
def _outils_frequents(self) -> dict:
compteur = {}
for r in self.requetes:
for outil in r["outils"]:
compteur[outil] = compteur.get(outil, 0) + 1
return dict(sorted(compteur.items(), key=lambda x: -x[1]))Gestion des coûts
class GestionnaireCouts:
def __init__(self, budget_quotidien: float = 50.0):
self.budget = budget_quotidien
self.depenses_jour = 0.0
self.date_reset = datetime.now().date()
def peut_executer(self, cout_estime: float) -> bool:
"""Vérifie si le budget permet l'execution."""
aujourdhui = datetime.now().date()
if aujourdhui > self.date_reset:
self.depenses_jour = 0.0
self.date_reset = aujourdhui
return (self.depenses_jour + cout_estime) <= self.budget
def enregistrer_depense(self, cout: float):
self.depenses_jour += cout
def budget_restant(self) -> float:
return self.budget - self.depenses_jour
# Integration dans l'orchestreur
gestionnaire_couts = GestionnaireCouts(budget_quotidien=100.0)
def executer_avec_budget(orchestreur, requete, **kwargs):
cout_estime = 0.05 # Estimation conservative
if not gestionnaire_couts.peut_executer(cout_estime):
return {"erreur": "Budget quotidien atteint", "restant": 0}
resultat = orchestreur.executer(requete, **kwargs)
cout_reel = resultat["trace"].tokens_total * 0.000002
gestionnaire_couts.enregistrer_depense(cout_reel)
return resultatPattern : routeur intelligent
Quand les cas d’usage se multiplient, la tentation est de tout confier à un unique système doté de tous les outils. C’est le choix qui coûte le plus cher : chaque requête, même triviale, transporte l’intégralité des définitions d’outils. Un routeur qui classe d’abord la demande puis appelle l’orchestrateur spécialisé garde les contextes courts et rend chaque flux mesurable séparément.
class RouteurIntelligent:
def __init__(self):
self.client = OpenAI()
self.orchestreurs = {}
def enregistrer(self, domaine: str, orchestreur: OrchestreurProduction):
self.orchestreurs[domaine] = orchestreur
def router(self, requete: str) -> dict:
"""Détermine le domaine et route vers le bon orchestreur."""
# Classification rapide avec un modèle léger
response = self.client.responses.create(
model="gpt-5.6-terra",
input=f"Classifie cette requête dans un domaine : {requete}\n"
f"Domaines disponibles : {list(self.orchestreurs.keys())}\n"
f"Réponds uniquement avec le nom du domaine.",
)
domaine = response.output_text.strip().lower()
if domaine not in self.orchestreurs:
domaine = "general" # Fallback
logger.info(f"Requête routee vers : {domaine}")
return self.orchestreurs[domaine].executer(requete)
# Configuration
routeur = RouteurIntelligent()
routeur.enregistrer("commercial", OrchestreurProduction(config_assistant_commercial))
routeur.enregistrer("support", OrchestreurProduction(config_support))
routeur.enregistrer("rh", OrchestreurProduction(config_rh))
# Utilisation
resultat = routeur.router("Combien de jours de congés me reste-t-il ?")
# -> Route automatiquement vers l'orchestreur RHChecklist de mise en production
Cette checklist est courte à dessein. Elle ne vise pas l’exhaustivité mais les points dont l’absence se paie en incident plutôt qu’en inconfort — et qu’on ne peut pas ajouter dans l’urgence, le jour où le système est déjà entre les mains des utilisateurs.
checklist = {
"securite": [
"Validation des arguments de chaque fonction",
"Rate limiting par utilisateur",
"Sanitization des entrées et sorties",
"Pas de données sensibles dans les logs"
],
"fiabilite": [
"Retry avec backoff sur les erreurs transitoires",
"Circuit breaker par service externe",
"Timeout par outil et par requête globale",
"Fallback quand un outil est indisponible"
],
"performance": [
"Execution parallèle des appels independants",
"Cache des résultats de recherche fréquents",
"Limite du nombre de tours de boucle",
"Choix du modèle adapte (gpt-5.6-terra vs gpt-5.6-terra)"
],
"observabilite": [
"Logging structure de chaque appel",
"Métriques de latence et de coût",
"Alertes sur le taux d'erreur",
"Trace complete de chaque requête"
],
"couts": [
"Budget quotidien avec coupe-circuit",
"Estimation du coût avant execution",
"Rapport hebdomadaire des dépenses",
"Optimisation du choix de modèle par complexité"
]
}Points clés à retenir
- Décomposez en couches : interface, orchestration, exécution, observabilité
- Utilisez une configuration déclarative pour définir les systèmes
- Tracez chaque exécution pour le monitoring et le debugging
- Implémentez un gestionnaire de coûts avec budget quotidien
- Un routeur intelligent distribue les requêtes vers les bons orchestreurs
Testez vos connaissances
Function calling, recherche, exécution : l’arsenal complet en cinq questions.
1. Que doit contenir un bon schéma de fonction ?
Réponse : Des types stricts et des descriptions opérationnelles par paramètre — c’est sur ce contrat que le modèle décide quand appeler et comment remplir ; la validation rejette le reste.
2. Web Search ou File Search : comment choisir ?
Réponse : Web Search pour l’information publique et récente ; File Search pour vos documents indexés (chunking + ranking intégrés) — et la recherche hybride combine les deux quand la réponse traverse les deux mondes.
3. Que permet Code Interpreter dans l'API ?
Réponse : Exécuter du code dans un bac à sable : traiter CSV/Excel/images, calculer juste, produire des graphiques — le modèle raisonne, le code calcule, le sandbox isole.
4. À quoi sert Tool Search ?
Réponse : À exposer dynamiquement un grand catalogue d’outils : le modèle recherche l’outil pertinent au lieu de recevoir cent schémas d’un coup — le contexte reste léger, les capacités restent vastes.
5. Quelles règles pour une architecture multi-outils en production ?
Réponse : Des outils au périmètre net, la gestion d’erreurs et fallbacks par outil, des appels parallèles quand ils sont indépendants, et l’observabilité de chaque appel — la robustesse se conçoit outil par outil.
Chaque outil a son terrain ; la production les combine — l’architecture finale ci-dessus est votre plan de référence.