Multi-tool architectures in production
On this page
Updated November 2, 2026 based on OpenAI documentation: the Code Interpreter tool now requires a
containerparameter (added to all examples), text input parts are typed asinput_text, and web search uses theweb_searchtool (web_search_previewis now a legacy tool without domain filters). Generated files are retrieved using thefichiers_generesandtelechargerfunctions from Lesson 10. Sources: Code Interpreter · Web search.
Multi-tool architectures in production
A multi-tool prototype works because a single person uses it, knows its limits, and restarts it when it fails. Production removes all three conditions at once. What becomes critical isn’t answer quality—that was already there—but three properties the prototype lacked: knowing what happened when a call fails, preventing a runaway loop from draining your budget, and being able to modify behavior without redeploying.
The patterns below each address one of these requirements. Don’t adopt them all at once: start with observability, because without it you won’t even know which of the others you’re missing.
Layered architecture
Separating concerns into layers isn’t architectural elegance: it makes each part replaceable. The concrete benefit is that you can change orchestration without touching the interface, or add a tool without rewriting business logic. The four layers below are the minimal breakdown that holds up in practice.
# Layer 1: Interface - request reception
# Layer 2: Orchestration - loop management and routing
# Layer 3: Execution - tool and function calls
# Layer 4: Observability - logs, metrics, 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:
"""Execution trace for 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.0Production orchestrator
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:
"""Builds the tool array from 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",
"search_context_size": config["web_search"].get("context_size", "medium")
})
if config.get("code_interpreter"):
tools.append({"type": "code_interpreter", "container": {"type": "auto"}})
for func in config.get("fonctions", []):
tools.append(func)
return tools
def _construire_dispatcher(self, config: dict) -> dict:
"""Maps function names to their implementations."""
return config.get("implementations", {})
def executer(self, requete: str, fichiers: list = None,
max_tours: int = 10) -> dict:
"""Executes a request with complete tracing."""
trace = TraceExecution(
requete_id=f"req_{int(time.time()*1000)}"
)
content = [{"type": "input_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
# Identify function calls
appels = [
item for item in response.output
if item.type == "function_call"
]
# Record tools used
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
}
# Execute custom functions
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:
"""Executes functions with error handling."""
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 resultatsDeclarative configuration
Moving system definitions out of code and into configuration changes who can intervene. Adjusting a system prompt, removing a tool, or modifying a cost ceiling becomes a file change—reviewable and reversible—instead of a deployment. On a system that evolves weekly, that’s the difference between iterating and reacting.
config_assistant_commercial = {
"model": "gpt-5.6-terra",
"instructions": (
"You are a sales assistant. You have access to CRM data, "
"product documentation, the web, and data analysis. "
"Prioritize internal data, enrich with web data."
),
"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": "Create an opportunity in the 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 a proposal for Acme Corp")Observability and monitoring
Metrics dashboard
Three metrics are enough to start, and each answers a different question: latency per tool tells you which one is slowing the system down, failure rate per tool tells you which one is fragile, and cost per request tells you which one is spiraling. A global average across all three answers none of these 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:
"""Estimates the cost in dollars of the execution."""
# Approximate rates for gpt-5.6-terra: $2 / $12 per 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]))Cost management
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:
"""Checks if the budget allows 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 into the orchestrator
gestionnaire_couts = GestionnaireCouts(budget_quotidien=100.0)
def executer_avec_budget(orchestreur, requete, **kwargs):
cout_estime = 0.05 # Conservative estimate
if not gestionnaire_couts.peut_executer(cout_estime):
return {"erreur": "Daily budget reached", "restant": 0}
resultat = orchestreur.executer(requete, **kwargs)
cout_reel = resultat["trace"].tokens_total * 0.000002
gestionnaire_couts.enregistrer_depense(cout_reel)
return resultatPattern: intelligent router
As use cases multiply, the temptation is to hand everything to a single system equipped with every tool. That’s the most expensive choice: every request, even trivial ones, carries the full set of tool definitions. A router that classifies the request first, then calls a specialized orchestrator, keeps contexts short and makes each flow measurable separately.
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:
"""Determines the domain and routes to the correct orchestrator."""
# Fast classification with a lightweight model
response = self.client.responses.create(
model="gpt-5.6-terra",
input=f"Classify this request into a domain: {requete}\n"
f"Available domains: {list(self.orchestreurs.keys())}\n"
f"Respond only with the domain name.",
)
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))
# Usage
resultat = routeur.router("How many vacation days do I have left?")
# -> Automatically routes to the HR orchestratorProduction checklist
This checklist is intentionally short. It doesn’t aim for exhaustiveness but covers the points whose absence costs you in incidents rather than discomfort—and that you can’t add in a hurry the day the system is already in users’ hands.
checklist = {
"securite": [
"Validate arguments for each function",
"Rate limiting per user",
"Sanitize inputs and outputs",
"No sensitive data in logs"
],
"fiabilite": [
"Retry with backoff on transient errors",
"Circuit breaker per external service",
"Timeout per tool and per global request",
"Fallback when a tool is unavailable"
],
"performance": [
"Parallel execution of independent calls",
"Cache frequent search results",
"Limit the number of loop iterations",
"Choose the model adapted to complexity (gpt-5.6-terra vs gpt-5.6-terra)"
],
"observabilite": [
"Structured logging for each call",
"Latency and cost metrics",
"Alerts on error rate",
"Complete trace of each request"
],
"couts": [
"Daily budget with circuit breaker",
"Cost estimation before execution",
"Weekly spending report",
"Model choice optimization by complexity"
]
}Key takeaways
- Decompose into layers: interface, orchestration, execution, observability
- Use declarative configuration to define systems
- Trace every execution for monitoring and debugging
- Implement a cost manager with a daily budget
- An intelligent router distributes requests to the right orchestrators
Test your knowledge
Function calling, search, execution: the full arsenal in five questions.
1. What should a good function schema contain?
Answer: Strict types and operational descriptions per parameter—this is the contract the model uses to decide when to call and how to fill; validation rejects the rest.
2. Web Search or File Search: how to choose?
Answer: Web Search for public, recent information; File Search for your indexed documents (chunking + built-in ranking)—and hybrid search combines both when the answer spans both worlds.
3. What does Code Interpreter allow in the API?
Answer: Execute code in a sandbox: process CSV/Excel/images, calculate accurately, produce charts—the model reasons, the code calculates, the sandbox isolates.
4. What is Tool Search for?
Answer: To dynamically expose a large tool catalog: the model searches for the relevant tool instead of receiving a hundred schemas at once—the context stays light, capabilities stay vast.
5. What rules apply to a multi-tool architecture in production?
Answer: Tools with clear scope, error handling and fallbacks per tool, parallel calls when independent, and observability for every call—robustness is designed tool by tool.
Each tool has its terrain; production combines them—the final architecture above is your reference plan.