Multi-Tool-Architekturen in der Produktion
Auf dieser Seite
Aktualisiert am 2. November 2026 basierend auf der OpenAI-Dokumentation: Das Tool Code Interpreter erfordert nun einen
container-Parameter (in allen Beispielen hinzugefügt), Texteingabeteile werden alsinput_textformatiert, und die Websuche verwendet das Toolweb_search(web_search_previewist nun ein veraltetes Tool ohne Domänenfilter). Generierte Dateien werden mit den Funktionenfichiers_generesundtelechargeraus Lektion 10 abgerufen. Quellen: Code Interpreter · Websuche.
Multi-Tool-Architekturen in der Produktion
Ein Prototyp einer Multi-Tool-Architektur funktioniert, weil eine einzelne Person ihn nutzt, seine Grenzen kennt und ihn bei Fehlern neu startet. Die Produktion entfernt alle drei dieser Bedingungen gleichzeitig. Was kritisch wird, ist nicht die Antwortqualität – diese war bereits vorhanden – sondern drei Eigenschaften, die dem Prototypen fehlten: Die Kenntnis dessen, was passiert, wenn ein Aufruf fehlschlägt, die Verhinderung eines ausufernden Schleifenlaufs, der das Budget erschöpft, und die Möglichkeit, das Verhalten zu ändern, ohne neu zu deployen.
Die nachfolgenden Muster adressieren jeweils eines dieser Anforderungen. Nehmen Sie nicht alle auf einmal auf: Beginnen Sie mit Observability, denn ohne sie werden Sie nicht einmal wissen, welche der anderen Ihnen fehlen.
Schichtenarchitektur
Die Trennung von Anliegen in Schichten ist keine architektonische Eleganz: Sie macht jeden Teil austauschbar. Der konkrete Vorteil besteht darin, dass Sie die Orchestrierung ändern können, ohne die Schnittstelle zu berühren, oder ein Tool hinzufügen können, ohne die Geschäftslogik neu zu schreiben. Die vier Schichten unten sind die minimale Aufteilung, die in der Praxis standhält.
# Schicht 1: Schnittstelle - Anfrageempfang
# Schicht 2: Orchestrierung - Schleifenverwaltung und Routing
# Schicht 3: Ausführung - Tool- und Funktionsaufrufe
# Schicht 4: Observability - Logs, Metriken, 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:
"""Ausführungs-Trace für die Überwachung."""
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.0Produktions-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:
"""Erstellt das Tool-Array aus der Konfiguration."""
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:
"""Ordnet Funktionsnamen ihren Implementierungen zu."""
return config.get("implementations", {})
def executer(self, requete: str, fichiers: list = None,
max_tours: int = 10) -> dict:
"""Führt eine Anfrage mit vollständiger Verfolgung aus."""
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
# Identifizierung von Funktionsaufrufen
appels = [
item for item in response.output
if item.type == "function_call"
]
# Protokollierung der verwendeten Tools
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
}
# Ausführung benutzerdefinierter Funktionen
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"Anfragefehler {trace.requete_id}: {e}")
return {"reponse": None, "erreur": str(e), "trace": trace}
trace.duree_totale = time.time() - trace.debut
return {"reponse": "Schleifengrenze erreicht", "trace": trace}
def _executer_fonctions(self, appels: list, trace: TraceExecution) -> list:
"""Führt Funktionen mit Fehlerbehandlung aus."""
resultats = []
for appel in appels:
try:
func = self.dispatcher.get(appel.name)
if not func:
raise ValueError(f"Unbekannte Funktion: {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 resultatsDeklarative Konfiguration
Die Verlagerung von Systemdefinitionen aus dem Code in die Konfiguration ändert, wer eingreifen kann. Das Anpassen eines Systemprompts, das Entfernen eines Tools oder das Ändern einer Kostenobergrenze wird zu einer Dateiänderung – überprüfbar und reversibel – statt zu einem Deployment. Bei einem System, das sich wöchentlich entwickelt, ist das der Unterschied zwischen Iterieren und Reagieren.
config_assistant_commercial = {
"model": "gpt-5.6-terra",
"instructions": (
"Sie sind ein Vertriebsassistent. Sie haben Zugriff auf CRM-Daten, "
"Produktdokumentation, das Web und Datenanalysen. "
"Priorisieren Sie interne Daten und ergänzen Sie diese mit Webdaten."
),
"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": "Erstellen Sie eine Gelegenheit im 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("Bereiten Sie ein Angebot für Acme Corp vor")Observability und Monitoring
Metriken-Dashboard
Drei Metriken reichen aus, um zu beginnen, und jede beantwortet eine andere Frage: Die Latenz pro Tool zeigt, welches das System verlangsamt, die Fehlerrate pro Tool zeigt, welches fragil ist, und die Kosten pro Anfrage zeigen, welches sich aus dem Ruder läuft. Ein globaler Durchschnitt über alle drei beantwortet keine dieser Fragen.
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:
"""Schätzt die Kosten in Dollar für die Ausführung."""
# Näherungsweise Raten für gpt-5.6-terra: $2 / $12 pro 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]))Kostenmanagement
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:
"""Überprüft, ob das Budget die Ausführung erlaubt."""
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 in den Orchestrator
gestionnaire_couts = GestionnaireCouts(budget_quotidien=100.0)
def executer_avec_budget(orchestreur, requete, **kwargs):
cout_estime = 0.05 # Konservative Schätzung
if not gestionnaire_couts.peut_executer(cout_estime):
return {"erreur": "Tägliches Budget erreicht", "restant": 0}
resultat = orchestreur.executer(requete, **kwargs)
cout_reel = resultat["trace"].tokens_total * 0.000002
gestionnaire_couts.enregistrer_depense(cout_reel)
return resultatMuster: Intelligenter Router
Wenn Anwendungsfälle zunehmen, ist die Versuchung groß, alles einem einzigen System anzuvertrauen, das mit jedem Tool ausgestattet ist. Das ist die teuerste Wahl: Jede Anfrage, selbst banale, trägt den vollständigen Satz Tool-Definitionen. Ein Router, der die Anfrage zunächst klassifiziert und dann einen spezialisierten Orchestrator aufruft, hält Kontexte kurz und macht jeden Fluss separat messbar.
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:
"""Bestimmt den Bereich und leitet an den richtigen Orchestrator weiter."""
# Schnelle Klassifizierung mit einem leichten Modell
response = self.client.responses.create(
model="gpt-5.6-terra",
input=f"Klassifizieren Sie diese Anfrage in einen Bereich: {requete}\n"
f"Verfügbare Bereiche: {list(self.orchestreurs.keys())}\n"
f"Antworten Sie nur mit dem Bereichsnamen.",
)
domaine = response.output_text.strip().lower()
if domaine not in self.orchestreurs:
domaine = "general" # Fallback
logger.info(f"Anfrage geroutet an: {domaine}")
return self.orchestreurs[domaine].executer(requete)
# Konfiguration
routeur = RouteurIntelligent()
routeur.enregistrer("commercial", OrchestreurProduction(config_assistant_commercial))
routeur.enregistrer("support", OrchestreurProduction(config_support))
routeur.enregistrer("rh", OrchestreurProduction(config_rh))
# Nutzung
resultat = routeur.router("Wie viele Urlaubstage habe ich noch?")
# -> Routet automatisch an den HR-OrchestratorProduktions-Checkliste
Diese Checkliste ist absichtlich kurz. Sie zielt nicht auf Vollständigkeit ab, sondern deckt die Punkte ab, deren Fehlen Sie in Incidents kostet, statt nur Unbehagen zu verursachen – und die Sie nicht in Eile hinzufügen können, wenn das System bereits in den Händen der Nutzer ist.
checklist = {
"securite": [
"Validierung der Argumente für jede Funktion",
"Rate Limiting pro Benutzer",
"Sensibilisierung von Eingaben und Ausgaben",
"Keine sensiblen Daten in Logs"
],
"fiabilite": [
"Wiederholung mit Backoff bei vorübergehenden Fehlern",
"Circuit Breaker pro externem Dienst",
"Timeout pro Tool und pro globaler Anfrage",
"Fallback, wenn ein Tool nicht verfügbar ist"
],
"performance": [
"Parallele Ausführung unabhängiger Aufrufe",
"Caching häufiger Suchergebnisse",
"Begrenzung der Anzahl der Schleifeniterationen",
"Auswahl des für die Komplexität angepassten Modells (gpt-5.6-terra vs gpt-5.6-terra)"
],
"observabilite": [
"Strukturierte Protokollierung für jeden Aufruf",
"Latenz- und Kostenmetriken",
"Alarme bei Fehlerrate",
"Vollständiger Trace jeder Anfrage"
],
"couts": [
"Tägliches Budget mit Circuit Breaker",
"Kostenschätzung vor der Ausführung",
"Wöchentliches Ausgabenbericht",
"Optimierung der Modellwahl nach Komplexität"
]
}Kernaussagen
- Zerlegen Sie in Schichten: Schnittstelle, Orchestrierung, Ausführung, Observability
- Verwenden Sie deklarative Konfiguration zur Definition von Systemen
- Verfolgen Sie jede Ausführung für Monitoring und Debugging
- Implementieren Sie einen Kostenmanager mit einem täglichen Budget
- Ein intelligenter Router verteilt Anfragen an die richtigen Orchestratoren
Testen Sie Ihr Wissen
Funktionsaufrufe, Suche, Ausführung: das gesamte Arsenal in fünf Fragen.
1. Was sollte ein gutes Funktionsschema enthalten?
Antwort: Strenge Typen und operationale Beschreibungen pro Parameter – dies ist der Vertrag, den das Modell verwendet, um zu entscheiden, wann es aufruft und wie es füllt; die Validierung lehnt den Rest ab.
2. Websuche oder Dateisuche: Wie wählt man?
Antwort: Websuche für öffentliche, aktuelle Informationen; Dateisuche für Ihre indizierten Dokumente (Chunking + eingebaute Ranking) – und hybride Suche kombiniert beide, wenn die Antwort beide Welten spannt.
3. Was ermöglicht Code Interpreter in der API?
Antwort: Code in einer Sandbox ausführen: CSV/Excel/Bilder verarbeiten, präzise berechnen, Diagramme erstellen – das Modell denkt, der Code berechnet, die Sandbox isoliert.
4. Wozu dient Tool Search?
Antwort: Um ein großes Tool-Katalog dynamisch freizugeben: Das Modell sucht nach dem relevanten Tool, statt hundert Schemas auf einmal zu erhalten – der Kontext bleibt leicht, die Fähigkeiten bleiben vast.
5. Welche Regeln gelten für eine Multi-Tool-Architektur in der Produktion?
Antwort: Tools mit klarem Umfang, Fehlerbehandlung und Fallbacks pro Tool, parallele Aufrufe, wenn unabhängig, und Observability für jeden Aufruf – Robustheit wird Tool für Tool entworfen.
Jedes Tool hat sein Terrain; die Produktion kombiniert sie – der obige Endarchitekturplan ist Ihr Referenzplan.