Architectures multi-outils en production
Architectures multi-outils en production
Passer d’un prototype a un systeme de production avec plusieurs outils exige une architecture pensee pour la fiabilite, la performance et l’observabilite. Cette lecon couvre les patterns eprouves pour deployer des applications multi-outils a grande echelle.
Architecture en couches
Un systeme multi-outils en production se decompose en quatre couches :
# Couche 1 : Interface - reception 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 implementations."""
return config.get("implementations", {})
def executer(self, requete: str, fichiers: list = None,
max_tours: int = 10) -> dict:
"""Execute une requete 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.3"),
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 utilises
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
}
# Executer les fonctions personnalisees
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 requete {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:
"""Execute 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 declarative
Definissez vos systemes multi-outils via une configuration :
config_assistant_commercial = {
"model": "gpt-5.3",
"instructions": (
"Tu es un assistant commercial. Tu as acces aux donnees CRM, "
"a la documentation produit, au web et a l'analyse de donnees. "
"Priorise les donnees 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": "Creer une opportunite 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")Observabilite et monitoring
Tableau de bord des metriques
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 cout en dollars de l'execution."""
# Tarifs approximatifs GPT-5.3
cout_input = trace.tokens_total * 0.6 * 0.000001
cout_output = trace.tokens_total * 0.4 * 0.000003
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 couts
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:
"""Verifie 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
Pour les systemes avec de nombreux cas d’usage, un routeur dirige les requetes vers le bon orchestreur :
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:
"""Determine le domaine et route vers le bon orchestreur."""
# Classification rapide avec un modele leger
response = self.client.responses.create(
model="o4-mini",
input=f"Classifie cette requete dans un domaine : {requete}\n"
f"Domaines disponibles : {list(self.orchestreurs.keys())}\n"
f"Reponds uniquement avec le nom du domaine.",
)
domaine = response.output_text.strip().lower()
if domaine not in self.orchestreurs:
domaine = "general" # Fallback
logger.info(f"Requete 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 conges me reste-t-il ?")
# -> Route automatiquement vers l'orchestreur RHChecklist de mise en production
Avant de deployer un systeme multi-outils :
checklist = {
"securite": [
"Validation des arguments de chaque fonction",
"Rate limiting par utilisateur",
"Sanitization des entrees et sorties",
"Pas de donnees sensibles dans les logs"
],
"fiabilite": [
"Retry avec backoff sur les erreurs transitoires",
"Circuit breaker par service externe",
"Timeout par outil et par requete globale",
"Fallback quand un outil est indisponible"
],
"performance": [
"Execution parallele des appels independants",
"Cache des resultats de recherche frequents",
"Limite du nombre de tours de boucle",
"Choix du modele adapte (o4-mini vs gpt-5.3)"
],
"observabilite": [
"Logging structure de chaque appel",
"Metriques de latence et de cout",
"Alertes sur le taux d'erreur",
"Trace complete de chaque requete"
],
"couts": [
"Budget quotidien avec coupe-circuit",
"Estimation du cout avant execution",
"Rapport hebdomadaire des depenses",
"Optimisation du choix de modele par complexite"
]
}Points cles a retenir
- Decomposez en couches : interface, orchestration, execution, observabilite
- Utilisez une configuration declarative pour definir les systemes
- Tracez chaque execution pour le monitoring et le debugging
- Implementez un gestionnaire de couts avec budget quotidien
- Un routeur intelligent distribue les requetes vers les bons orchestreurs