Der Aufbau eines Multi-Agenten-Systems mit CrewAI wird interessanter, wenn unterschiedliche Agenten verschiedene Modelle nutzen können.
Ein Researcher profitiert möglicherweise von einem schnellen, kostengünstigen Modell, ein Analyst benötigt eventuell ein stärkeres Reasoning-Modell, und ein Writer braucht ein Modell, das für hochwertige Long-Form-Generierung optimiert ist. Traditionell bedeutet die Anbindung dieser Agenten an unterschiedliche Anbieter, dass man separate API-Anmeldedaten, Endpunkte, SDKs, Abrechnungssysteme und anbieterspezifische Konfigurationen verwalten muss.
Eine klarere Architektur ist, CrewAI die Agenten und den Workflow managen zu lassen, während CometAPI den Modellzugang verwaltet.
CometAPI stellt einen OpenAI-kompatiblen Endpunkt unter https://api.cometapi.com/v1 bereit, sodass Anwendungen Anfragen an Modelle verschiedener Anbieter über eine gemeinsame API-Schnittstelle leiten können. Die aktuelle Quick-Start-Dokumentation unterstützt zudem die Nutzung des Standard-OpenAI-Python-SDKs, indem der API-Schlüssel und die Basis-URL geändert werden.
In diesem Tutorial baust du einen CrewAI-Workflow mit drei Agenten:
- Gemini 3.7 Flash für Research
- Claude Opus 5 für Analyse
- GPT-5.6 für das finale Schreiben
- Einen CometAPI-API-Schlüssel
- Eine API-Basis-URL
- Pro-Agent Modellkonfiguration
- Begrenzte Fallbacks bei vorübergehenden Fehlern
- CrewAI-Checkpointing für Produktions-Resilienz
- Token- und Ausführungs-Tracking
- Serverseitige Modellvalidierung
Die wichtige Architekturgrenze ist einfach:
CrewAI übernimmt die Agenten-Orchestrierung. CometAPI übernimmt den Modellzugang. Modell-IDs definieren das Routing.
Was ist CrewAI Multi-Agent Model Routing?
CrewAI ist ein Python-Framework zum Erstellen von Agenten, Aufgaben, Crews und Multi-Agent-Workflows. Jeder Agent kann seine eigene LLM-Konfiguration haben, während die Crew koordiniert, wie diese Agenten Aufgaben ausführen und Kontext austauschen.
Die aktuelle LLM-Konfiguration von CrewAI unterstützt explizite Einstellungen für model, api_key und base_url, einschließlich benutzerdefinierter OpenAI-kompatibler Endpunkte.
Das macht eine Multi-Model-Architektur unkompliziert:
CometAPI │ https://api.cometapi.com/v1 │ ┌───────────────────┼───────────────────┐ │ │ │ Researcher Analyst Writer │ │ │ Gemini 3.7 Flash Claude Opus 5 GPT-5.6
Die Agenten bleiben logisch getrennt, aber ihr Modellzugang ist zentralisiert.
Das bedeutet nicht, dass alle Modelle austauschbar sind. Eine OpenAI-kompatible API stellt eine gemeinsame Anfrageschnittstelle bereit; sie garantiert nicht identische Kontextlimits, Tool-Unterstützung, Reasoning-Steuerungen, Ausgabeverhalten, Latenz oder Preise.
Diese Unterscheidung ist wichtig bei der Gestaltung von Produktions-Routing.
Warum CometAPI mit CrewAI verwenden?
Der Hauptvorteil ist nicht, dass CrewAI plötzlich zu einem Multi-Provider-Framework wird. CrewAI unterstützt bereits mehrere LLM-Anbieter.
Der Vorteil ist, dass der Modellzugang hinter einer API-Schicht konsolidiert werden kann.
Ohne eine einheitliche API-Schicht könnte ein Workflow mit drei Agenten so aussehen:
| Agent | Anbieter | Zugangsdaten | Integration |
|---|---|---|---|
| Researcher | Google-API-Schlüssel | Anbieter-spezifisch | |
| Analyst | Anthropic | Anthropic-API-Schlüssel | Anbieter-spezifisch |
| Writer | OpenAI | OpenAI-API-Schlüssel | Anbieter-spezifisch |
Mit CometAPI:
| Agent | Modell | Zugangsdaten | Endpunkt |
|---|---|---|---|
| Researcher | Gemini 3.7 Flash | CometAPI-Key | CometAPI |
| Analyst | Claude Opus 5 | CometAPI-Key | CometAPI |
| Writer | GPT-5.6 | CometAPI-Key | CometAPI |
Die aktuelle Quick-Start-Dokumentation von CometAPI beschreibt seinen Endpunkt als Drop-in-Ersatz für die OpenAI-API-Basis-URL und listet Modelle mehrerer Anbieter über denselben Dienst.
Das verschafft der Anwendung eine nützliche Trennung:
CrewAI
- Definiert Agentenrollen
- Definiert Aufgaben
- Übergibt Kontext
- Steuert die Ausführung
- Verwaltet Agenten-Iterationen
- Handhabt die Crew-Orchestrierung
CometAPI
- Bietet eine gemeinsame Modellzugangsschicht
- Zentralisiert API-Authentifizierung
- Bietet Modellrouting über Modell-IDs
- Gibt der Anwendung einen API-Endpunkt
- Ermöglicht zentrale Sichtbarkeit von Nutzung und Abrechnung
Was baut dieser CrewAI-Workflow?
Das Beispiel erstellt drei sequentielle Agenten.
| CrewAI-Agent | Primäres Modell | Fallback | Rolle |
|---|---|---|---|
| Market Researcher | gemini-3.7-flash | gpt-5.6 | Fakten sammeln und Recherche |
| Product Analyst | claude-opus-5 | gpt-5.6 | Evidenz und Trade-offs synthetisieren |
| Technical Writer | gpt-5.6 | gemini-3.7-flash | Finale Entscheidungs-Memo verfassen |
Dies ist eine Beispiel-Routing-Policy, keine Benchmark-Rangliste.
Das richtige Modell für deinen Agenten hängt ab von:
- Aufgabenkomplexität
- benötigter Kontextlänge
- Tool-Nutzung
- Anforderungen an strukturierte Ausgaben
- Latenz
- Zuverlässigkeit
- Token-Kosten
- Ausgabequalität
- anwendungsspezifischen Evaluationsergebnissen
Eine nützliche Faustregel ist:
Wähle ein Modell passend zur Aufgabe des Agenten, nicht nur zum Anbieter, von dem es stammt.
Welches Modell sollte jeder CrewAI-Agent nutzen?
Für dieses Beispiel folgt die Modellzuweisung einer einfachen Kosten-gegen-Fähigkeiten-Strategie.
Researcher: Gemini 3.7 Flash
Research umfasst häufig die Verarbeitung relativ großer Informationsmengen und die Erstellung eines kompakten Zwischenresultats.
Ein schnelles Modell kann daher für umfangreiche Rechercheaufgaben nützlich sein.
"researcher": "gemini-3.7-flash"
Analyst: Claude Opus 5
Der Analyst hat eine engere, aber stärker reasoning-intensive Rolle. Er erhält die Research-Ausgabe und wandelt sie in eine Empfehlung um.
"analyst": "claude-opus-5"
Writer: GPT-5.6
Der finale Agent konvertiert die Recherche und Analyse in ein entwicklerorientiertes Entscheidungs-Memo.
"writer": "gpt-5.6"
Wichtig ist nicht diese exakte Dreifachzuweisung. Deine Anwendung sollte Kandidatenmodelle anhand repräsentativer Aufgaben evaluieren, bevor die Routing-Policy festgelegt wird.
Was brauchst du vor dem Start?
Du benötigst:
- Python 3.10+
- CrewAI
- OpenAI-Python-SDK-Kompatibilität
python-dotenv- Einen CometAPI-API-Schlüssel
- Die Modell-IDs, die du verwenden willst
Die aktuelle Python-Integration von CometAPI unterstützt die OpenAI-kompatible API, und das offizielle CometAPI-Python-Paket dokumentiert COMETAPI_KEY und COMETAPI_BASE_URL als umgebungsbasierte Konfigurationsoptionen.
Der Standard-Endpunkt ist:
https://api.cometapi.com/v1
Vor der Bereitstellung überprüfe, dass deine ausgewählten Modell-IDs derzeit verfügbar sind und den Endpunkt sowie die Parameter unterstützen, die dein CrewAI-Workload benötigt. Modellkataloge und Preise können sich ändern.
Wie installierst du CrewAI und die Abhängigkeiten?
Lege eine neue Python-Umgebung an:
python -m venv .venv
Aktiviere sie:
source .venv/bin/activate
Unter Windows:
.venv\Scripts\Activate.ps1
Installiere dann die Abhängigkeiten:
pip install "crewai[openai]" openai python-dotenv
Die explizite Verwendung von openai ist beabsichtigt, da die untenstehende Fallback-Implementierung OpenAI-SDK-Ausnahmeklassen direkt importiert.
Für die Produktion sollten die Versionen, die du testest, festgelegt werden, statt sich dauerhaft auf „latest“ zu verlassen.
Zum Beispiel:
crewai==YOUR_TESTED_VERSIONopenai==YOUR_TESTED_VERSIONpython-dotenv==YOUR_TESTED_VERSION
Die LLM-Schicht von CrewAI entwickelt sich aktiv weiter, daher sollten der genaue Konstruktor und die Anbieter-Konfiguration mit der CrewAI-Version, die deine Anwendung nutzt, abgeglichen werden. Die aktuelle CrewAI-Dokumentation unterstützt die Konfiguration eines LLM mit benutzerdefinierter base_url und API-Schlüssel.
Wie konfigurierst du den CometAPI-API-Schlüssel?
Erstelle eine .env-Datei:
COMETAPI_KEY=your_cometapi_keyCOMETAPI_BASE_URL=https://api.cometapi.com/v1
Lade diese Werte in Python:
import osfrom dotenv import load_dotenvload_dotenv()COMETAPI_KEY = os.environ["COMETAPI_KEY"]COMETAPI_BASE_URL = os.getenv( "COMETAPI_BASE_URL", "https://api.cometapi.com/v1",)
Committe .env niemals in Git.
Füge sie zu .gitignore hinzu:
.env.venv/__pycache__/
Der API-Schlüssel sollte ein serverseitiges Credential bleiben. Die aktuelle Schnellstart-Empfehlung von CometAPI rät ebenfalls dazu, den Schlüssel in Umgebungsvariablen statt im Quellcode zu speichern.
Wie verbindest du CrewAI mit CometAPI?
Das LLM-Objekt von CrewAI kann einen Modellnamen, API-Schlüssel und eine benutzerdefinierte Basis-URL erhalten.
Erstelle einen Helfer:
from crewai import LLMdef cometapi_llm(model_id: str) -> LLM: return LLM( model=model_id, base_url=COMETAPI_BASE_URL, api_key=COMETAPI_KEY, timeout=60.0, max_retries=0, )
Das ist vorzuziehen, statt dieselbe Konfiguration in jedem Agenten separat einzubetten.
Jeder Agent benötigt nun nur eine Modell-ID:
research_llm = cometapi_llm("gemini-3.7-flash")analysis_llm = cometapi_llm("claude-opus-5")writing_llm = cometapi_llm("gpt-5.6")
Warum max_retries=0 setzen?
Der Grund ist die Fallback-Kontrolle.
Wenn der zugrunde liegende LLM-Client automatisch erneut versucht und deine Anwendung ebenfalls Fallback implementiert, kann aus einem Fehler mehrere versteckte Anfragen werden, bevor die Fallback-Logik greift.
Für ein Tutorial mit explizitem Routing ist es sauberer, der Anwendung zu überlassen, wann erneut versucht oder das Modell gewechselt wird.
Wie definierst du die Modell-Routing-Policy?
Halte Routing außerhalb deiner Prompts:
PRIMARY_MODELS = { "researcher": "gemini-3.7-flash", "analyst": "claude-opus-5", "writer": "gpt-5.6",}FALLBACK_MODELS = { "researcher": "gpt-5.6", "analyst": "gpt-5.6", "writer": "gemini-3.7-flash",}
Das schafft eine klare Konfigurationsgrenze.
Du kannst die gleiche Zuordnung später in:
- Umgebungskonfiguration
- YAML
- JSON
- eine Datenbank
- Feature Flags
- einen internen Modell-Routing-Service
verschieben, ohne die Agenten-Prompts neu zu schreiben.
Wie baust du die drei CrewAI-Agenten?
Erstelle je ein LLM-Objekt pro Agent.
from crewai import Agentdef build_agents(model_map: dict[str, str]): researcher = Agent( role="Market Researcher", goal="Collect the facts needed to answer the topic", backstory=( "You create concise, source-aware research briefs " "and clearly separate facts from assumptions." ), llm=cometapi_llm(model_map["researcher"]), max_iter=3, allow_delegation=False, ) analyst = Agent( role="Product Analyst", goal="Turn research into a defensible recommendation", backstory=( "You identify evidence, assumptions, risks, " "and trade-offs before making recommendations." ), llm=cometapi_llm(model_map["analyst"]), max_iter=3, allow_delegation=False, ) writer = Agent( role="Technical Writer", goal="Produce a concise technical decision memo", backstory=( "You write clear technical explanations " "without unnecessary marketing language." ), llm=cometapi_llm(model_map["writer"]), max_iter=3, allow_delegation=False, ) return researcher, analyst, writer
Die Modellzuweisung ist nun vollständig unabhängig von der Rollendefinition des Agenten.
Das macht Modellrouting praktikabel.
Wie verbindest du die Agenten mit sequentiellen Aufgaben?
Erstelle drei Aufgaben:
from crewai import Taskdef build_tasks(researcher, analyst, writer): research_task = Task( description=( "Research this topic: {topic}. " "Return the key facts, uncertainties, " "and relevant sources that the analyst should consider." ), expected_output=( "A compact research brief containing facts, " "uncertainties, and source references." ), agent=researcher, ) analysis_task = Task( description=( "Using the research brief, analyze {topic}. " "Identify the strongest conclusion and explain " "the major trade-offs." ), expected_output=( "A decision outline with evidence, " "assumptions, risks, and trade-offs." ), agent=analyst, context=[research_task], ) writing_task = Task( description=( "Write a concise technical decision memo about {topic}. " "State the recommendation early and preserve " "important caveats." ), expected_output="A polished technical decision memo in Markdown.", agent=writer, context=[research_task, analysis_task], ) return research_task, analysis_task, writing_task
Die Abhängigkeitskette ist:
Topic ↓Research ↓Analysis ↓Final memo
Der Analyst erhält die Ausgabe der Research-Aufgabe, während der Writer sowohl den Research- als auch den Analyse-Kontext erhält.
Wie baust du die Crew?
Kombiniere die Agenten und Aufgaben:
from crewai import Crew, Processdef build_crew(model_map: dict[str, str]) -> Crew: researcher, analyst, writer = build_agents(model_map) research_task, analysis_task, writing_task = build_tasks( researcher, analyst, writer, ) return Crew( agents=[researcher, analyst, writer], tasks=[ research_task, analysis_task, writing_task, ], process=Process.sequential, verbose=True, )
Nun ist das Modellrouting vollständig konfigurationsgetrieben.
Die Änderung von:
"researcher": "gemini-3.7-flash"
zu einem anderen unterstützten Modell erfordert keine Anpassung des Research-Prompts oder der Aufgabendefinition.
Wie sollte CrewAI-Modell-Fallback funktionieren?
Hier braucht eine produktionsorientierte Implementierung mehr Sorgfalt.
Ein häufiger Fehler ist:
Any error ↓Switch model
Das ist zu aggressiv.
Beispielsweise sollten diese Fehler im Allgemeinen keinen Modell-Fallback auslösen:
400 Bad Request401 Unauthorized403 Forbidden404 Not Found422 Validation Error
Der Modellwechsel behebt keinen ungültigen API-Schlüssel oder eine fehlerhafte Anfrage.
Fallback ist angemessener bei vorübergehenden Ausfällen wie:
408 Request Timeout429 Rate Limit500 Internal Server Error502 Bad Gateway503 Service Unavailable504 Gateway TimeoutConnection errorTimeout
Die Fallback-Policy sollte daher sein:
Nur bei begrenzten, vorübergehenden Fehlern erneut versuchen oder Modelle wechseln – und nur, wenn das Fallback-Modell denselben Request-Contract unterstützt.
Wie erkennst du wiederholbare Fehler?
Du kannst die Fehlerklassen des OpenAI-SDK verwenden:
from collections.abc import Iteratorfrom openai import ( APIConnectionError, APIStatusError, APITimeoutError,)def exception_chain(error: BaseException) -> Iterator[BaseException]: current: BaseException | None = error seen: set[int] = set() while current is not None and id(current) not in seen: seen.add(id(current)) yield current current = ( current.__cause__ or current.__context__ )def should_fallback(error: BaseException) -> bool: for current in exception_chain(error): if isinstance( current, (APIConnectionError, APITimeoutError), ): return True if isinstance(current, APIStatusError): return ( current.status_code in {408, 429} or current.status_code >= 500 ) return False
Dies schließt absichtlich 400er-Konfigurationsfehler außer 408 und 429 aus.
Solltest du die gesamte Crew oder nur den fehlgeschlagenen Agenten erneut ausführen?
Es gibt zwei unterschiedliche Fallback-Strategien.
Crew-Level-Fallback
Die einfachste Implementierung ist:
Start crew ↓failure ↓change routing ↓run crew again
Das ist leicht zu verstehen, kann aber abgeschlossene Aufgaben wiederholen.
Zum Beispiel:
Research → completedAnalysis → completedWriter → failed
Ein vollständiger kickoff()-Retry kann ausführen:
Research → againAnalysis → againWriter → fallback
Das erhöht:
- Tokenverbrauch
- Latenz
- API-Kosten
- potenzielle Nebenwirkungen
Task-Level-Recovery
Ein Produktions-Workflow sollte stattdessen abgeschlossene Arbeit checkpointen:
Research ↓checkpoint ↓Analysis ↓checkpoint ↓Writer fails ↓retry writer with fallback
CrewAI bietet derzeit Checkpointing, das den Ausführungszustand speichert und eine Ausführung nach einem Fehler fortsetzen lässt. Das dokumentierte Checkpoint-Verhalten überspringt abgeschlossene Aufgaben und setzt die nachgelagerte Arbeit vom gespeicherten Zustand fort.
Das ist die bessere Architektur für teure oder nebenwirkungsproduzierende Workflows.
Wie fügst du CrewAI-Checkpointing hinzu?
Für Produktions-Workflows aktiviere Checkpointing in der Crew:
crew = Crew( agents=[researcher, analyst, writer], tasks=[ research_task, analysis_task, writing_task, ], process=Process.sequential, checkpoint=True, verbose=True,)
Das Checkpointing-System von CrewAI kann den Ausführungszustand nach Abschluss von Aufgaben persistieren und die Crew aus einem Checkpoint wiederherstellen.
Ein wiederhergestellter Lauf kann beispielsweise nutzen:
from crewai import CheckpointConfigresult = crew.kickoff( from_checkpoint=CheckpointConfig( restore_from="./.checkpoints/checkpoint.json", ))
Die genaue Checkpoint-Konfiguration sollte der CrewAI-Version deines Projekts folgen.
Der wichtige Architekturpunkt ist:
Zuerst checkpointen, dann fallbacken.
Dies verhindert, dass ein vorübergehender Modellfehler teure, abgeschlossene Arbeit erneut ausführen muss.
Wie implementierst du ein einfaches begrenztes Fallback?
Für ein Tutorial kannst du dennoch ein einfaches Crew-Level-Fallback demonstrieren.
def run_with_fallback(topic: str): routes = [ PRIMARY_MODELS, { **PRIMARY_MODELS, "writer": FALLBACK_MODELS["writer"], }, { **PRIMARY_MODELS, "analyst": FALLBACK_MODELS["analyst"], "writer": FALLBACK_MODELS["writer"], }, ] last_error = None for attempt, model_map in enumerate(routes, start=1): try: crew = build_crew(model_map) result = crew.kickoff( inputs={"topic": topic} ) return result, model_map except Exception as error: last_error = error if not should_fallback(error): raise if attempt == len(routes): raise print( f"Transient failure on attempt {attempt}. " f"Trying bounded fallback route.", flush=True, ) raise RuntimeError( "Crew execution failed after all fallback routes." ) from last_error
Beachte die wichtige Unterscheidung:
Dies behauptet nicht, dass der fehlgeschlagene Agent identifiziert wurde.
Es ist eine begrenzte Crew-Level-Fallback-Strategie.
Für kleine, zustandslose Workflows kann das akzeptabel sein. Für Produktions-Workflows mit teurer Recherche, Tools oder Nebenwirkungen nutze eine Checkpoint-basierte Wiederherstellung.
Wie trackst du den CrewAI-Tokenverbrauch?
Usage-Tracking sollte Teil der Routing-Schicht sein, nicht ein Nachgedanke.
Am Ende des Laufs inspiziere das CrewAI-Resultat:
result, selected_models = run_with_fallback(topic)print("Selected models:")print(selected_models)print("Final result:")print(result.raw)print("Usage:")print(result.token_usage)
Die genauen verfügbaren Usage-Felder können von der CrewAI-Version und dem Ausführungspfad abhängen, behandle daher das zurückgegebene Ergebnisobjekt als „source of truth“ für die eingesetzte Version.
Ein Produktions-Nutzungsdatensatz sollte idealerweise enthalten:
job_idagentmodelinput_tokensoutput_tokenstotal_tokenslatency_msfallback_usedfallback_reasonstatuscreated_at
So kannst du Fragen beantworten wie:
Welcher Agent verbraucht den größten Teil des Budgets?
Wie oft fällt der Analyst zurück?
Welches Modell hat die höchste Latenz?
Wie viel kostet jeder Workflow?
Wie steuerst du Kosten auf Agentenebene?
Multi-Model-Routing ist am nützlichsten, wenn es tatsächliche Workload-Unterschiede widerspiegelt.
Zum Beispiel:
Researcher→ hohes Volumen→ günstigeres ModellAnalyst→ geringes Volumen→ stärkeres Reasoning-ModellWriter→ mittleres Volumen→ allgemeines Produktionsmodell
Du kannst Kosten auch über die Agentenkonfiguration begrenzen.
Zum Beispiel:
max_iter=3
begrenzt die Iterationsschleife des Agenten. Das sollte nicht als harte Grenze von genau drei API-Aufrufen oder drei Token-Budgets interpretiert werden.
Zusätzliche Steuerungen umfassen:
- Begrenzen des Task-Kontexts
- Zusammenfassen von Zwischenoutputs
- Caching wiederholbarer Research
- Begrenzen der maximalen Eingabegröße
- Begrenzen der maximalen Output-Tokens, wo unterstützt
- Einschränken von Tool-Aufrufen
- Setzen von Budgets pro Benutzer
- Setzen von Budgets pro Workflow
- Tracking der Fallback-Häufigkeit
Wie validierst du Modelle vor der Bereitstellung?
Hardcode keine Modell-IDs auf Dauer.
Ein Modell kann werden:
- nicht verfügbar
- umbenannt
- depreziert
- eingeschränkt
- in Fähigkeit geändert
- im Preis geändert
- inkompatibel mit einem Parameter, den deine Anwendung nutzt
CometAPI stellt einen Modellkatalog-Endpunkt bereit, der programmatisch abgefragt werden kann, während das öffentliche Modellverzeichnis für die menschliche Modellentdeckung genutzt werden kann.
Ein Deployment-Check kann so aussehen:
curl -s \ https://api.cometapi.com/api/models \ -H "Authorization: Bearer $COMETAPI_KEY"
Validiere dann, dass deine konfigurierten Modell-IDs vor der Bereitstellung vorhanden sind.
Zum Beispiel kann dein CI-Prozess prüfen:
gemini-3.7-flash → availableclaude-opus-5 → availablegpt-5.6 → available
Mache Verfügbarkeitsprüfungen nicht zum Ersatz für Anwendungstests. Dass ein Modell im Katalog vorhanden ist, bedeutet nicht, dass jeder Parameter, jedes Tool oder jedes Ausgabeformat, das dein CrewAI-Agent nutzt, unterstützt wird.
Wie sieht das vollständige CrewAI-Beispiel aus?
Hier ist eine konsolidierte Implementierung:
import jsonimport osimport sysfrom collections.abc import Iteratorfrom dotenv import load_dotenvfrom openai import ( APIConnectionError, APIStatusError, APITimeoutError,)from crewai import Agent, Crew, LLM, Process, Taskload_dotenv()COMETAPI_KEY = os.environ["COMETAPI_KEY"]COMETAPI_BASE_URL = os.getenv( "COMETAPI_BASE_URL", "https://api.cometapi.com/v1",)PRIMARY_MODELS = { "researcher": "gemini-3.7-flash", "analyst": "claude-opus-5", "writer": "gpt-5.6",}FALLBACK_MODELS = { "researcher": "gpt-5.6", "analyst": "gpt-5.6", "writer": "gemini-3.7-flash",}def cometapi_llm(model_id: str) -> LLM: return LLM( model=model_id, base_url=COMETAPI_BASE_URL, api_key=COMETAPI_KEY, timeout=60.0, max_retries=0, )def build_crew(model_map: dict[str, str]) -> Crew: researcher = Agent( role="Market Researcher", goal="Collect the facts needed to answer the topic", backstory=( "You create concise, source-aware research briefs " "and distinguish facts from assumptions." ), llm=cometapi_llm(model_map["researcher"]), max_iter=3, allow_delegation=False, ) analyst = Agent( role="Product Analyst", goal="Turn research into a defensible recommendation", backstory=( "You evaluate evidence, assumptions, risks, " "and trade-offs." ), llm=cometapi_llm(model_map["analyst"]), max_iter=3, allow_delegation=False, ) writer = Agent( role="Technical Writer", goal="Produce a concise technical decision memo", backstory=( "You write clear technical explanations " "without unnecessary hype." ), llm=cometapi_llm(model_map["writer"]), max_iter=3, allow_delegation=False, ) research_task = Task( description=( "Research this topic: {topic}. " "Return the key facts, uncertainties, " "and relevant sources." ), expected_output=( "A concise research brief with facts " "and open questions." ), agent=researcher, ) analysis_task = Task( description=( "Using the research brief, analyze {topic}. " "Identify the strongest conclusion and " "explain the major trade-offs." ), expected_output=( "A decision outline with evidence, " "assumptions, risks, and trade-offs." ), agent=analyst, context=[research_task], ) writing_task = Task( description=( "Write a concise technical decision memo " "about {topic}. State the recommendation early " "and preserve important caveats." ), expected_output=( "A polished technical decision memo in Markdown." ), agent=writer, context=[ research_task, analysis_task, ], ) return Crew( agents=[ researcher, analyst, writer, ], tasks=[ research_task, analysis_task, writing_task, ], process=Process.sequential, verbose=True, )def exception_chain( error: BaseException,) -> Iterator[BaseException]: current = error seen: set[int] = set() while current is not None and id(current) not in seen: seen.add(id(current)) yield current current = ( current.__cause__ or current.__context__ )def should_fallback(error: BaseException) -> bool: for current in exception_chain(error): if isinstance( current, ( APIConnectionError, APITimeoutError, ), ): return True if isinstance(current, APIStatusError): return ( current.status_code in {408, 429} or current.status_code >= 500 ) return Falsedef run_with_fallback(topic: str): routes = [ PRIMARY_MODELS, { **PRIMARY_MODELS, "writer": FALLBACK_MODELS["writer"], }, { **PRIMARY_MODELS, "analyst": FALLBACK_MODELS["analyst"], "writer": FALLBACK_MODELS["writer"], }, ] last_error = None for attempt, model_map in enumerate( routes, start=1, ): try: crew = build_crew(model_map) result = crew.kickoff( inputs={ "topic": topic, } ) return result, model_map except Exception as error: last_error = error if not should_fallback(error): raise if attempt == len(routes): raise print( f"Transient failure on attempt " f"{attempt}; trying fallback.", file=sys.stderr, ) raise RuntimeError( "No model route completed the crew." ) from last_errordef main(): topic = ( sys.argv[1] if len(sys.argv) > 1 else ( "Should a small SaaS add " "AI-generated meeting summaries?" ) ) result, selected_models = ( run_with_fallback(topic) ) output = { "selected_models": selected_models, "raw": result.raw, "tasks_output": [ task.raw for task in result.tasks_output ], "token_usage": str( result.token_usage ), } print( json.dumps( output, indent=2, default=str, ) )if __name__ == "__main__": main()
Die wichtige Verbesserung gegenüber der ursprünglichen Version ist, dass der Code nicht mehr fälschlicherweise impliziert, eine Exception identifiziere den exakt fehlgeschlagenen Agenten.
Es ist explizit eine begrenzte Crew-Level-Fallback-Implementierung.
Für die Produktion kombiniere dieselbe Routing-Policy mit CrewAI-Checkpointing.
Wie führst du den CrewAI-Workflow aus?
Speichere die Datei als:
crewai_multi_model.py
Führe dann aus:
python crewai_multi_model.py \ "Should a small SaaS add AI-generated meeting summaries?"
Eine erfolgreiche Antwort enthält Informationen ähnlich:
{ "selected_models": { "researcher": "gemini-3.7-flash", "analyst": "claude-opus-5", "writer": "gpt-5.6" }, "raw": "<final decision memo>", "tasks_output": [ "<research output>", "<analysis output>", "<writing output>" ], "token_usage": "<usage information>"}
Die genaue Antwort und die Nutzungswerte hängen von der Eingabe, dem Modellverhalten, der CrewAI-Version und dem Ausführungspfad ab.
Wenn ein wiederholbarer Fehler die Fallback-Route aktiviert, zeigt das Objekt selected_models die Route, die für diese Crew-Ausführung genutzt wurde.
Wie solltest du Produktions-Modellrouting entwerfen?
Eine Produktions-Routing-Policy sollte mehr als Modellqualität berücksichtigen.
Eine nützliche Entscheidungsfunktion ist:
Model Score =Quality+ Reliability+ Context Fit+ Tool Compatibility- Cost- Latency
Du kannst das auf mehreren Ebenen implementieren.
Kostenbasiertes Routing
Einfache Aufgabe → günstiges ModellKomplexe Aufgabe → Premium-Modell
Latenzbasiertes Routing
Interaktive Anfrage → schnelles ModellHintergrund-Workflow → qualitativ höherwertiges Modell
Zuverlässigkeitsbasiertes Routing
Primärmodell ↓vorübergehender Ausfall ↓Fallback-Modell
Aufgabenbasiertes Routing
Research → Modell AAnalyse → Modell BSchreiben → Modell CCode → Modell D
Der letzte Ansatz ist für CrewAI besonders naheliegend, da das Framework jedem Agenten bereits eine eigene Rolle gibt.
Wie machst du Fallback sicher?
Ein robustes Fallback-System sollte vier Regeln durchsetzen.
Nicht bei Authentifizierungsfehlern fallbacken
Wenn der API-Schlüssel ungültig ist:
401
wird der Modellwechsel das Problem nicht lösen.
Nicht bei fehlerhaften Anfragen fallbacken
Wenn die Anfrage ungültig ist:
400422
korrigiere stattdessen die Anfrage.
Nicht unbegrenzt fallbacken
Setze eine harte Grenze:
MAX_FALLBACK_ATTEMPTS = 2
Ein Fallback-System ohne Grenze kann zu einer teuren Retry-Schleife werden.
Fallback-Modelle müssen request-kompatibel sein
Ein Fallback-Modell muss die Features unterstützen, die dein Agent benötigt.
Wenn der primäre Agent z. B. ein spezifisches Tool oder ein strukturiertes Ausgabeverhalten erfordert, muss das Fallback dasselbe Contract unterstützen.
OpenAI-kompatibel bedeutet nicht Feature-kompatibel.
Was sind die häufigsten CrewAI + CometAPI-Fehler?
| Symptom | Wahrscheinliche Ursache | Lösung |
|---|---|---|
| 401 Unauthorized | Ungültiger oder fehlender API-Key | COMETAPI_KEY prüfen; nicht fallbacken |
| 400 Bad Request | Ungültige Anfrageparameter | Anfrage korrigieren |
| 404 Model Not Found | Veraltete Modell-ID | Aktuellen Modellkatalog prüfen |
| 408 Timeout | Vorübergehendes Timeout | Innerhalb einer begrenzten Policy erneut versuchen |
| 429 Rate Limited | Zu viele Anfragen | Backoff und Retry |
| 500–504 | Vorübergehender Server/Gateway-Fehler | Begrenztes Fallback nutzen |
| Agent wiederholt Retries | Versteckte SDK-Retries | max_retries kontrollieren |
| Abgeschlossene Aufgaben laufen erneut | Vollständiger Crew-Retry | Checkpoint-basierte Wiederherstellung |
| Unterschiedliches Modell verhält sich anders | Unterschiedliche Modellfähigkeiten | Jedes Modell separat testen |
| Unerwarteter CrewAI-Konstruktorfehler | Versionsinkompatibilität | CrewAI-Version pinnen und verifizieren |
Wie trennst du CrewAI-Fehler von Modell-Fehlern?
Diese Unterscheidung ist wichtig beim Debuggen.
Konfigurationsfehler
Missing API keyInvalid model IDInvalid base URLUnsupported parameter
Diese sollten schnell fehlschlagen.
Provider/API-Fehler
401403404429500503
Diese erfordern je nach Status unterschiedliche Behandlung.
Anwendungsfehler
Agent output invalidTool returned malformed dataTask context missingSide effect failed
Diese werden nicht notwendigerweise durch Modellwechsel gelöst.
Ein reifes Agentensystem sollte daher eine getrennte Behandlung für Folgendes haben:
configuration ↓API transport ↓model execution ↓agent logic ↓tool execution ↓application side effects
Das ist deutlich sicherer als ein generisches:
except Exception: use_fallback()
Wie schützt du externe Nebenwirkungen?
Fallback wird erheblich komplizierter, wenn Agenten mehr tun als Text zu generieren.
Stell dir beispielsweise einen Agenten vor, der:
- einen Datenbankeintrag erstellt
- eine E-Mail sendet
- eine externe API aufruft
- ein CRM aktualisiert
Wenn das Modell nach dem erfolgreichen externen Schritt ein Timeout hat, kann ein erneutes Ausführen der gesamten Crew die Aktion duplizieren.
Nutze:
- Idempotency Keys
- Task-Checkpoints
- Transaktionsgrenzen
- Ausführungs-IDs
- Dauerhaften Taskzustand
- Explizite Bestätigung von Nebenwirkungen
Zum Beispiel:
job_id = crew_run_123task_id = writer_456
Speichere diese Kennungen bei externen Operationen, damit ein Retry feststellen kann, ob die Operation bereits erfolgt ist.
Wie überwachst du Multi-Model-CrewAI-Workflows?
Mindestens logge:
workflow_idagentmodeltaskstart_timeend_timelatencystatusfallback_usedfallback_reasoninput_tokensoutput_tokenstotal_tokens
Logge nicht:
API keysprivate credentialsfull sensitive promptsprivate user dataunredacted model output
Für jedes Modell überwache:
Zuverlässigkeit
success ratetimeout rate5xx ratefallback rate
Performance
p50 latencyp95 latencyp99 latency
Kosten
input tokensoutput tokenscost per taskcost per completed workflow
Qualität
task success ratehuman evaluationstructured-output validitytool-call success
So wird Modellrouting von einer hartcodierten Präferenz zu einem beobachtbaren Engineering-System.
Wie wählst du zwischen direkten Provider-APIs und CometAPI?
Die Wahl hängt von deiner Architektur ab.
| Architektur | Zugangsdaten | Modellwechsel | Anbieterintegration | Zentrales Routing |
|---|---|---|---|---|
| Direkte Provider-APIs | Mehrere | Eigene Lösung | Hoch | Nein |
| Einzelner Anbieter | Einer | Begrenzt | Niedrig | Begrenzt |
| CrewAI + CometAPI | Ein CometAPI-Credential | Modell-ID-basiert | Geringer | Ja |
Wenn deine Anwendung nur einen Anbieter und dessen native Fähigkeiten benötigt, kann eine direkte Integration völlig sinnvoll sein.
Wenn deine CrewAI-Anwendung Modelle mehrerer Anbieter braucht und du eine gemeinsame Zugriffsschicht willst, wird CometAPI attraktiver.
Wichtig ist, dass CometAPI CrewAI nicht ersetzt.
Stattdessen:
CrewAIAgent orchestration ↓CometAPIModel access ↓Multiple models
Jede Schicht hat eine andere Verantwortung.
Wie skaliert diese Architektur?
Sobald die Routing-Policy von den Agentendefinitionen getrennt ist, erfordert das Hinzufügen eines weiteren Modells keinen Neuaufbau der gesamten Anwendung.
Zum Beispiel:
PRIMARY_MODELS = { "researcher": "gemini-3.7-flash", "analyst": "claude-opus-5", "writer": "gpt-5.6", "coder": "YOUR_CODE_MODEL",}
Dieselbe Architektur kann dann unterstützen:
Research-AgentAnalyse-AgentCoding-AgentReview-AgentWriting-AgentFact-Checking-Agent
Jeder Agent kann ein anderes Modell haben, während er dieselbe CometAPI-Zugangsschicht teilt.
Der nächste Schritt ist, Routing dynamisch zu machen.
Anstatt:
"analyst": "claude-opus-5"
könntest du schließlich verwenden:
select_model( task="analysis", budget=budget, latency_target=latency_target,)
Das Routing-System kann dann aus freigegebenen Modellen basierend auf den Anforderungen der Anwendung wählen.
Was ist die beste Produktionsarchitektur für CrewAI + CometAPI?
Für einen kleinen Workflow:
User Input ↓CrewAI ↓CometAPI ↓Models
Für die Produktion:
┌───────────────┐ │ Model Catalog │ └───────┬───────┘ │ ▼User → CrewAI → Routing Policy → CometAPI │ │ │ │ │ ├── Gemini │ │ ├── Claude │ │ └── GPT │ │ │ ▼ │ Cost / Quality / │ Latency / Policy │ ▼ Checkpoints │ ▼ Usage Tracking
Die wichtigsten Produktionskomponenten sind:
- Modell-Allowlist
- Pro-Agent-Routing
- Begrenzte Retries
- Task-Checkpointing
- Usage-Tracking
- Kostenkontrollen
- Modell-Kompatibilitätstests
- Observability
- Idempotente Nebenwirkungen
Diese Architektur ist deutlich robuster als einfach ein try/except um crew.kickoff() zu setzen.
Ein CometAPI-Schlüssel, verschiedene Modelle, klarere Agentenrollen
Am nützlichsten ist es, CrewAI und CometAPI als zwei komplementäre Schichten zu betrachten.
CrewAI definiert, was die Agenten tun.
CometAPI definiert, wie diese Agenten auf Modelle zugreifen.
Diese Trennung macht es möglich, einem hochvolumigen Research-Agenten ein schnelles Modell, einem Analyse-Agenten ein stärkeres Reasoning-Modell und dem finalen Writer ein allgemeines Produktionsmodell zuzuweisen, ohne separate Anbieter-Integrationen innerhalb des Workflows zu pflegen.
Die einfachste Implementierung nutzt einen CometAPI-Schlüssel und eine OpenAI-kompatible Basis-URL:
https://api.cometapi.com/v1
Für die Produktion gehe einen Schritt weiter: Halte Modellrouting in der Konfiguration, validiere die Modellverfügbarkeit vor der Bereitstellung, nutze begrenzte Fallbacks nur bei vorübergehenden Fehlern, checkpointiere abgeschlossene Aufgaben und erfasse Modell- und Nutzungsmetadaten für jeden Lauf.
Das liefert ein deutlich belastbareres Muster, als CrewAI einfach mit einem einzigen LLM zu verbinden:
CrewAI orchestriert die Agenten. CometAPI zentralisiert den Modellzugang. Modell-IDs steuern das Routing. Checkpoints schützen abgeschlossene Arbeit. Usage-Tracking kontrolliert Kosten.
Häufig gestellte Fragen
Kann CrewAI mehrere KI-Modelle in derselben Crew nutzen?
Ja. Weise jedem CrewAI-Agenten eine eigene LLM-Konfiguration zu. Jede Konfiguration kann ihr eigenes Modell angeben, während derselbe CometAPI-API-Schlüssel und dieselbe Basis-URL genutzt werden.
Kann CrewAI sich mit einer OpenAI-kompatiblen API verbinden?
Ja. Die LLM-Konfiguration von CrewAI unterstützt eine benutzerdefinierte base_url und einen API-Schlüssel für OpenAI-kompatible Endpunkte.
Mit CometAPI lautet die Basis-URL:
https://api.cometapi.com/v1
Brauche ich separate API-Schlüssel für GPT, Claude und Gemini?
Beim Zugriff auf diese Modelle über CometAPI kann die Anwendung das CometAPI-Credential und den Endpunkt verwenden, statt separate Anbieter-Credentials in jedem CrewAI-Agenten zu implementieren.
Bedeutet ein API-Schlüssel, dass die Modelle identische Fähigkeiten haben?
Nein. Die API-Schnittstelle kann vereinheitlicht sein, während die Modellfähigkeiten unterschiedlich bleiben. Kontextfenster, Tool-Unterstützung, Parameter, Ausgabeverhalten, Latenz und Preise können je nach Modell variieren.
Sollte ich den gesamten CrewAI-Workflow erneut versuchen, wenn ein Modell fehlschlägt?
Nur bei einfachen, zustandslosen Workflows. Ein erneuter Start der gesamten Crew kann abgeschlossene Aufgaben wiederholen und Kosten erhöhen. Für Produktions-Workflows sollten abgeschlossene Aufgaben checkpointiert und nach Möglichkeit vom fehlgeschlagenen Teil aus fortgesetzt werden.
Die aktuelle Checkpoint-Funktionalität von CrewAI ist darauf ausgelegt, den Ausführungszustand zu bewahren und nach Fehlern fortzusetzen.
Sollte jede CrewAI-Exception einen Modell-Fallback auslösen?
Nein. Authentifizierungsfehler, fehlerhafte Anfragen, ungültige Modell-IDs und nicht unterstützte Parameter erfordern in der Regel Konfigurationsänderungen statt eines anderen Modells.
Fallback ist besser für begrenzte, vorübergehende Fehler wie Timeouts, Rate-Limits und temporäre 5xx-Antworten.
Wie tracke ich die Kosten jedes CrewAI-Agenten?
Erfasse für jede Aufgabe den Agentennamen, die Modell-ID, den Tokenverbrauch, die Latenz, den Ausführungsstatus und Informationen zu Fallbacks. Nutze die Daten, um die Kosten pro Agent und Workflow zu berechnen.
Kann ich das Modell, das einem Agenten zugewiesen ist, dynamisch ändern?
Ja. Halte Modell-IDs in einer Routing-Konfiguration, statt sie direkt in Agentendefinitionen einzubetten. Deine Anwendung kann dann Modelle anhand von Kosten, Latenz, Aufgabentyp oder Verfügbarkeit auswählen.
Ist CometAPI ein Ersatz für CrewAI?
Nein. Sie arbeiten auf unterschiedlichen Ebenen. CrewAI orchestriert Agenten und Aufgaben, während CometAPI eine einheitliche Modell-Zugangsschicht bereitstellt.
Wo finde ich die aktuellen CometAPI-Modelle?
Nutze das CometAPI Modellverzeichnis für die menschliche Modellentdeckung und die Modell-API für programmatische Validierung. Die aktuelle Quick-Start-Seite von CometAPI listet über 500 Modelle in den Kategorien Text, Bild, Video und Audio.