Créer un système multi‑agents avec CrewAI devient plus intéressant lorsque différents agents peuvent utiliser des modèles différents.
Un chercheur peut bénéficier d’un modèle rapide et économique, un analyste peut avoir besoin d’un modèle à plus forte capacité de raisonnement, et un rédacteur peut nécessiter un modèle optimisé pour de la génération longue et de haute qualité. Traditionnellement, connecter ces agents à différents fournisseurs implique de gérer des identifiants API séparés, des endpoints, des SDK, des systèmes de facturation et des configurations spécifiques à chaque fournisseur.
Une architecture plus propre consiste à laisser CrewAI gérer les agents et le flux de travail tandis que CometAPI gère l’accès aux modèles.
CometAPI fournit un endpoint compatible OpenAI à l’adresse https://api.cometapi.com/v1, de sorte que les applications peuvent router les requêtes vers des modèles de plusieurs fournisseurs via une interface API commune. Sa documentation de démarrage rapide actuelle prend également en charge l’utilisation du SDK Python OpenAI standard en changeant la clé API et l’URL de base.
Dans ce tutoriel, vous allez construire un flux de travail CrewAI à trois agents avec :
- Gemini 3.7 Flash pour la recherche
- Claude Opus 5 pour l’analyse
- GPT-5.6 pour la rédaction finale
- Une clé API CometAPI
- Une URL de base API
- Une configuration de modèle par agent
- Un repli borné pour les défaillances transitoires
- Le checkpointing de CrewAI pour la reprise en production
- Le suivi de la consommation de jetons et d’exécution
- La validation des modèles côté serveur
La frontière architecturale importante est simple :
CrewAI gère l’orchestration des agents. CometAPI gère l’accès aux modèles. Les IDs de modèles définissent le routage.
Qu’est‑ce que le routage de modèles multi‑agents dans CrewAI ?
CrewAI est un framework Python pour créer des agents, des tâches, des équipes et des flux de travail multi‑agents. Chaque agent peut avoir sa propre configuration LLM, tandis que le Crew coordonne la façon dont ces agents exécutent des tâches et échangent du contexte.
La configuration LLM actuelle de CrewAI prend en charge des paramètres explicites model, api_key et base_url, y compris des endpoints personnalisés compatibles OpenAI.
Cela rend une architecture multi‑modèles simple :
CometAPI │ https://api.cometapi.com/v1 │ ┌───────────────────┼───────────────────┐ │ │ │ Researcher Analyst Writer │ │ │ Gemini 3.7 Flash Claude Opus 5 GPT-5.6
Les agents restent séparés d’un point de vue logique, mais leur accès aux modèles est centralisé.
Cela ne signifie pas que tous les modèles sont interchangeables. Une API compatible OpenAI fournit une interface de requête commune ; elle ne garantit pas des limites de contexte identiques, le support des outils, les contrôles de raisonnement, le comportement de sortie, la latence ou les tarifs.
Cette distinction est importante lors de la conception du routage en production.
Pourquoi utiliser CometAPI avec CrewAI ?
L’avantage principal n’est pas que CrewAI devienne soudainement un framework multi‑fournisseurs. CrewAI prend déjà en charge plusieurs fournisseurs de LLM.
L’avantage est que l’accès aux modèles peut être consolidé derrière une seule couche API.
Sans couche API unifiée, un flux de travail à trois agents pourrait ressembler à ceci :
| Agent | Fournisseur | Identifiant | Intégration |
|---|---|---|---|
| Researcher | Clé API Google | Spécifique fournisseur | |
| Analyst | Anthropic | Clé API Anthropic | Spécifique fournisseur |
| Writer | OpenAI | Clé API OpenAI | Spécifique fournisseur |
Avec CometAPI :
| Agent | Modèle | Identifiant | Endpoint |
|---|---|---|---|
| Researcher | Gemini 3.7 Flash | Clé CometAPI | CometAPI |
| Analyst | Claude Opus 5 | Clé CometAPI | CometAPI |
| Writer | GPT-5.6 | Clé CometAPI | CometAPI |
La documentation de démarrage rapide actuelle de CometAPI décrit son endpoint comme un remplacement direct de l’URL de base de l’API OpenAI et répertorie des modèles de plusieurs fournisseurs via le même service.
Cela donne à l’application une séparation utile :
CrewAI
- Définit les rôles des agents
- Définit les tâches
- Transmet le contexte
- Contrôle l’exécution
- Gère les itérations des agents
- Gère l’orchestration au niveau de l’équipe
CometAPI
- Fournit une couche d’accès aux modèles commune
- Centralise l’authentification API
- Fournit le routage des modèles via les IDs de modèles
- Donne à l’application un endpoint API unique
- Offre une visibilité centralisée sur l’usage et la facturation
Que va construire ce flux de travail CrewAI ?
L’exemple crée trois agents séquentiels.
| Agent CrewAI | Modèle principal | Repli | Rôle |
|---|---|---|---|
| Market Researcher | gemini-3.7-flash | gpt-5.6 | Collecter des faits et faire des recherches |
| Product Analyst | claude-opus-5 | gpt-5.6 | Synthétiser des preuves et des arbitrages |
| Technical Writer | gpt-5.6 | gemini-3.7-flash | Produire la note de décision finale |
Il s’agit d’une politique de routage d’exemple, pas d’un classement de référence.
Le bon modèle pour votre agent dépend de :
- complexité de la tâche
- longueur de contexte requise
- utilisation d’outils
- exigences de sortie structurée
- latence
- fiabilité
- coût en jetons
- qualité de sortie
- résultats d’évaluation spécifiques à l’application
Une règle utile est :
Choisissez un modèle adapté au travail que l’agent effectue, pas simplement au fournisseur dont il provient.
Quel modèle chaque agent CrewAI doit‑il utiliser ?
Pour cet exemple, l’attribution suit une stratégie simple coût vs capacité.
Researcher : Gemini 3.7 Flash
La recherche implique souvent de traiter des quantités relativement importantes d’informations et de produire un résultat intermédiaire compact.
Un modèle rapide peut donc être utile pour des tâches de recherche à fort volume.
"researcher": "gemini-3.7-flash"
Analyst : Claude Opus 5
L’analyste a un rôle plus étroit mais plus intensif en raisonnement. Il reçoit la sortie de la recherche et la transforme en une recommandation.
"analyst": "claude-opus-5"
Writer : GPT-5.6
L’agent final convertit la recherche et l’analyse en une note de décision destinée aux développeurs.
"writer": "gpt-5.6"
L’important n’est pas ces trois affectations exactes. Votre application doit évaluer les modèles candidats sur des tâches représentatives avant de fixer la politique de routage.
De quoi avez‑vous besoin avant de commencer ?
Vous avez besoin de :
- Python 3.10+
- CrewAI
- Compatibilité avec le SDK Python OpenAI
python-dotenv- Une clé API CometAPI
- Les IDs des modèles que vous comptez utiliser
L’intégration Python actuelle de CometAPI prend en charge l’API compatible OpenAI, et le package Python officiel CometAPI documente COMETAPI_KEY et COMETAPI_BASE_URL comme options de configuration basées sur l’environnement.
L’endpoint standard est :
https://api.cometapi.com/v1
Avant le déploiement, vérifiez que les IDs de modèles sélectionnés sont actuellement disponibles et prennent en charge l’endpoint et les paramètres requis par votre charge de travail CrewAI. Les catalogues de modèles et la tarification peuvent évoluer.
Comment installer CrewAI et les dépendances ?
Créez un nouvel environnement Python :
python -m venv .venv
Activez‑le :
source .venv/bin/activate
Sous Windows :
.venv\Scripts\Activate.ps1
Puis installez les dépendances :
pip install "crewai[openai]" openai python-dotenv
Inclure explicitement openai est intentionnel car l’implémentation de repli ci‑dessous importe directement les classes d’exception du SDK OpenAI.
En production, figez les versions que vous testez plutôt que de dépendre indéfiniment des dernières versions flottantes.
Par exemple :
crewai==YOUR_TESTED_VERSIONopenai==YOUR_TESTED_VERSIONpython-dotenv==YOUR_TESTED_VERSION
La couche LLM de CrewAI évolue activement, donc le constructeur exact et la configuration du fournisseur doivent être vérifiés par rapport à la version de CrewAI utilisée par votre application. La documentation actuelle de CrewAI prend en charge la configuration d’un LLM avec un base_url personnalisé et une clé API.
Comment configurer la clé API CometAPI ?
Créez un fichier .env :
COMETAPI_KEY=your_cometapi_keyCOMETAPI_BASE_URL=https://api.cometapi.com/v1
Chargez ces valeurs en 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",)
Ne validez jamais .env dans Git.
Ajoutez‑le à .gitignore :
.env.venv/__pycache__/
La clé API doit rester une crédential côté serveur. Les consignes actuelles de démarrage rapide de CometAPI recommandent également de stocker la clé dans des variables d’environnement plutôt que dans le code source.
Comment connecter CrewAI à CometAPI ?
L’objet LLM de CrewAI peut recevoir un nom de modèle, une clé API et une URL de base personnalisée.
Créez un utilitaire :
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, )
C’est préférable à dupliquer la même configuration séparément dans chaque agent.
Chaque agent n’a alors besoin que d’un ID de modèle :
research_llm = cometapi_llm("gemini-3.7-flash")analysis_llm = cometapi_llm("claude-opus-5")writing_llm = cometapi_llm("gpt-5.6")
Pourquoi définir max_retries=0 ?
La raison est le contrôle du repli.
Si le client LLM sous‑jacent réessaie automatiquement et que votre application implémente également un repli, un échec peut devenir plusieurs requêtes cachées avant que la logique de repli ne s’exécute.
Pour un tutoriel avec un routage explicite, il est plus clair de laisser l’application décider quand réessayer ou changer de modèle.
Comment définir la politique de routage des modèles ?
Gardez le routage en dehors de vos invites :
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",}
Cela crée une frontière de configuration claire.
Vous pouvez ensuite déplacer la même correspondance dans :
- la configuration d’environnement
- YAML
- JSON
- une base de données
- des feature flags
- un service interne de routage de modèles
sans réécrire les invites des agents.
Comment construire les trois agents CrewAI ?
Créez un objet LLM par 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
L’attribution de modèle est désormais complètement indépendante de la définition du rôle de l’agent.
C’est ce qui rend le routage des modèles pratique.
Comment connecter les agents avec des tâches séquentielles ?
Créez trois tâches :
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
La chaîne de dépendances est :
Topic ↓Research ↓Analysis ↓Final memo
L’analyste reçoit la sortie de la tâche de recherche, tandis que le rédacteur reçoit à la fois le contexte de la recherche et de l’analyse.
Comment construire l’équipe (Crew) ?
Combinez les agents et les tâches :
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, )
Le routage des modèles est maintenant entièrement piloté par la configuration.
Changer :
"researcher": "gemini-3.7-flash"
vers un autre modèle pris en charge ne nécessite pas de modifier l’invite ou la définition de la tâche de recherche.
Comment doit fonctionner le repli de modèles CrewAI ?
C’est ici qu’une implémentation orientée production exige plus de soin.
Une erreur courante est :
Any error ↓Switch model
C’est trop agressif.
Par exemple, ces erreurs ne doivent généralement pas déclencher un repli de modèle :
400 Bad Request401 Unauthorized403 Forbidden404 Not Found422 Validation Error
Changer de modèle ne corrigera pas une clé API invalide ou une requête mal formée.
Le repli est plus approprié pour des défaillances temporaires telles que :
408 Request Timeout429 Rate Limit500 Internal Server Error502 Bad Gateway503 Service Unavailable504 Gateway TimeoutConnection errorTimeout
La politique de repli doit donc être :
Réessayer ou changer de modèle uniquement pour des défaillances transitoires bornées et uniquement lorsque le modèle de repli prend en charge le même contrat de requête.
Comment détecter les erreurs réessayables ?
Vous pouvez utiliser les classes d’erreur du SDK OpenAI :
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
Cela exclut délibérément les erreurs de configuration en 4xx autres que 408 et 429.
Faut‑il réessayer l’ensemble de l’équipe ou seulement l’agent en échec ?
Il existe deux stratégies de repli différentes.
Repli au niveau de l’équipe
L’implémentation la plus simple est :
Start crew ↓failure ↓change routing ↓run crew again
C’est facile à comprendre, mais cela peut répéter des tâches déjà terminées.
Par exemple :
Research → completedAnalysis → completedWriter → failed
Un nouvel appel kickoff() complet peut exécuter :
Research → againAnalysis → againWriter → fallback
Cela augmente :
- la consommation de jetons
- la latence
- le coût API
- les effets de bord potentiels
Reprise au niveau de la tâche
Un flux de travail de production doit plutôt checkpoint-er le travail terminé :
Research ↓checkpoint ↓Analysis ↓checkpoint ↓Writer fails ↓retry writer with fallback
CrewAI fournit actuellement un checkpointing qui enregistre l’état d’exécution et permet à un run de reprendre après un échec. Le comportement de checkpoint documenté saute les tâches terminées et continue le travail en aval à partir de l’état sauvegardé.
C’est la meilleure architecture pour des flux de travail coûteux ou produisant des effets de bord.
Comment ajouter le checkpointing CrewAI ?
Pour les flux de travail de production, activez le checkpointing sur l’équipe :
crew = Crew( agents=[researcher, analyst, writer], tasks=[ research_task, analysis_task, writing_task, ], process=Process.sequential, checkpoint=True, verbose=True,)
Le système de checkpointing de CrewAI peut persister l’état d’exécution après l’achèvement des tâches et restaurer l’équipe depuis un checkpoint.
Par exemple, une exécution restaurée peut utiliser :
from crewai import CheckpointConfigresult = crew.kickoff( from_checkpoint=CheckpointConfig( restore_from="./.checkpoints/checkpoint.json", ))
La configuration exacte des checkpoints doit suivre la version de CrewAI utilisée par votre projet.
Le point architectural important est :
Checkpoint d’abord, repli ensuite.
Cela empêche qu’une défaillance temporaire du modèle n’oblige à réexécuter un travail terminé coûteux.
Comment implémenter un repli simple et borné ?
Pour un tutoriel, vous pouvez toujours démontrer un repli simple au niveau de l’équipe.
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
Notez la distinction importante :
Ceci ne prétend pas que l’agent en échec a été identifié.
Il s’agit d’une stratégie de repli borné au niveau de l’équipe.
Pour de petits flux de travail sans état, cela peut être acceptable. Pour des flux de travail de production avec des recherches coûteuses, des outils ou des effets de bord, utilisez une reprise basée sur les checkpoints.
Comment suivre la consommation de jetons CrewAI ?
Le suivi d’usage doit faire partie de la couche de routage, pas un ajout après coup.
À la fin de l’exécution, inspectez le résultat CrewAI :
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)
Les champs d’usage exacts disponibles peuvent dépendre de la version de CrewAI et du chemin d’exécution, donc considérez l’objet de résultat retourné comme source de vérité pour la version que vous déployez.
Un enregistrement de consommation en production devrait idéalement contenir :
job_idagentmodelinput_tokensoutput_tokenstotal_tokenslatency_msfallback_usedfallback_reasonstatuscreated_at
Cela vous permet de répondre à des questions telles que :
Quel agent consomme le plus de budget ?
À quelle fréquence l’analyste recourt‑il au repli ?
Quel modèle a la latence la plus élevée ?
Combien coûte chaque flux de travail ?
Comment contrôler les coûts au niveau de l’agent ?
Le routage multi‑modèles est le plus utile lorsqu’il reflète les différences réelles de charge.
Par exemple :
Researcher→ high volume→ lower-cost modelAnalyst→ low volume→ stronger reasoning modelWriter→ medium volume→ general-purpose production model
Vous pouvez également contraindre le coût via la configuration de l’agent.
Par exemple :
max_iter=3
borne la boucle d’itération de l’agent. Cela ne doit pas être interprété comme une limite stricte de exactement trois appels API ou trois budgets de jetons.
D’autres contrôles incluent :
- limiter le contexte de la tâche
- résumer les sorties intermédiaires
- mettre en cache les recherches répétables
- limiter la taille d’entrée maximale
- limiter le nombre maximal de jetons de sortie lorsque pris en charge
- restreindre les appels d’outils
- définir des budgets par utilisateur
- définir des budgets par flux de travail
- suivre la fréquence des replis
Comment valider les modèles avant le déploiement ?
Ne codez pas en dur des IDs de modèles pour toujours.
Un modèle peut devenir :
- indisponible
- renommé
- obsolète
- restreint
- modifié en capacité
- modifié en tarification
- incompatible avec un paramètre utilisé par votre application
CometAPI fournit un endpoint de catalogue de modèles interrogeable par programme, tandis que son répertoire public de modèles peut être utilisé pour la découverte manuelle.
Un contrôle de déploiement peut ressembler à :
curl -s \ https://api.cometapi.com/api/models \ -H "Authorization: Bearer $COMETAPI_KEY"
Ensuite, validez que vos IDs de modèles configurés existent avant le déploiement.
Par exemple, votre CI peut vérifier :
gemini-3.7-flash → availableclaude-opus-5 → availablegpt-5.6 → available
Ne substituez pas les vérifications de disponibilité aux tests applicatifs. La présence d’un modèle dans un catalogue ne signifie pas que chaque paramètre, outil ou format de sortie utilisé par votre agent CrewAI est pris en charge.
À quoi ressemble l’exemple CrewAI complet ?
Voici une implémentation consolidée :
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()
L’amélioration importante par rapport à la version d’origine est que le code n’implique plus à tort qu’une exception identifie l’agent exact en échec.
Il s’agit explicitement d’une implémentation de repli borné au niveau de l’équipe.
En production, combinez la même politique de routage avec le checkpointing de CrewAI.
Comment exécuter le flux de travail CrewAI ?
Enregistrez le fichier sous :
crewai_multi_model.py
Puis exécutez :
python crewai_multi_model.py \ "Should a small SaaS add AI-generated meeting summaries?"
Une réponse réussie contiendra des informations similaires à :
{ "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>"}
La réponse exacte et les valeurs d’usage dépendent de l’entrée, du comportement du modèle, de la version de CrewAI et du chemin d’exécution.
Si une erreur réessayable active la voie de repli, l’objet selected_models montre la voie utilisée pour cette exécution de l’équipe.
Comment concevoir le routage des modèles en production ?
Une politique de routage de production doit considérer plus que la qualité du modèle.
Une fonction de décision utile est :
Model Score =Quality+ Reliability+ Context Fit+ Tool Compatibility- Cost- Latency
Vous pouvez l’implémenter à plusieurs niveaux.
Routage basé sur le coût
Simple task → economical modelComplex task → premium model
Routage basé sur la latence
Interactive request → fast modelBackground workflow → higher-quality model
Routage basé sur la fiabilité
Primary model ↓transient failure ↓fallback model
Routage basé sur la tâche
Research → Model AAnalysis → Model BWriting → Model CCode → Model D
Cette dernière approche est particulièrement naturelle pour CrewAI car le framework donne déjà à chaque agent un rôle distinct.
Comment rendre le repli sûr ?
Un système de repli robuste doit appliquer quatre règles.
Ne pas faire de repli sur des erreurs d’authentification
Si la clé API est invalide :
401
changer de modèle ne résoudra pas le problème.
Ne pas faire de repli sur des requêtes mal formées
Si la requête est invalide :
400422
corrigez plutôt la requête.
Ne pas faire de repli indéfiniment
Définissez une limite stricte :
MAX_FALLBACK_ATTEMPTS = 2
Un système de repli sans limite peut devenir une boucle de réessais coûteuse.
Rendre les modèles de repli compatibles avec la requête
Un modèle de repli doit prendre en charge les fonctionnalités requises par votre agent.
Par exemple, si l’agent principal nécessite un outil spécifique ou un comportement de sortie structuré, le repli doit prendre en charge le même contrat.
Compatible OpenAI ne signifie pas compatible en fonctionnalités.
Quelles sont les erreurs les plus courantes CrewAI + CometAPI ?
| Symptôme | Cause probable | Correctif |
|---|---|---|
| 401 Non autorisé | Clé API invalide ou manquante | Vérifier COMETAPI_KEY ; pas de repli |
| 400 Mauvaise requête | Paramètres de requête invalides | Corriger la requête |
| 404 Modèle introuvable | ID de modèle obsolète | Vérifier le catalogue de modèles |
| 408 Délai d’attente | Délai de requête temporaire | Réessayer avec politique bornée |
| 429 Limite de débit | Trop de requêtes | Backoff et réessayer |
| 500–504 | Défaillance serveur/passerelle | Utiliser un repli borné |
| L’agent réessaie à répétition | Réessais cachés du SDK | Contrôler max_retries |
| Tâches terminées réexécutées | Réessai de toute l’équipe | Utiliser la reprise via checkpoints |
| Comportements différents par modèle | Capacités de modèles différentes | Tester chaque modèle indépendamment |
| Erreur de constructeur CrewAI inattendue | Décalage de version | Geler et vérifier la version CrewAI |
Comment séparer les erreurs CrewAI des erreurs de modèle ?
Cette distinction est importante lors du débogage.
Erreurs de configuration
Missing API keyInvalid model IDInvalid base URLUnsupported parameter
Celles‑ci doivent échouer rapidement.
Erreurs du fournisseur/de l’API
401403404429500503
Celles‑ci nécessitent une gestion différente selon le statut.
Erreurs applicatives
Agent output invalidTool returned malformed dataTask context missingSide effect failed
Celles‑ci ne sont pas nécessairement résolues en changeant de modèle.
Un système d’agents mature devrait donc avoir une gestion séparée pour :
configuration ↓API transport ↓model execution ↓agent logic ↓tool execution ↓application side effects
C’est bien plus sûr qu’un générique :
except Exception: use_fallback()
Comment protéger les effets de bord externes ?
Le repli devient nettement plus compliqué lorsque les agents font plus que générer du texte.
Par exemple, imaginez un agent qui :
- crée un enregistrement de base de données
- envoie un e‑mail
- appelle une API externe
- met à jour un CRM
Si le modèle expire après que l’action externe a réussi, relancer toute l’équipe peut dupliquer l’action.
Utilisez :
- des clés d’idempotence
- des checkpoints de tâches
- des frontières transactionnelles
- des IDs d’exécution
- un état de tâche durable
- une confirmation explicite des effets de bord
Par exemple :
job_id = crew_run_123task_id = writer_456
Stockez ces identifiants avec les opérations externes pour qu’un réessai puisse déterminer si l’opération a déjà eu lieu.
Comment superviser des flux de travail CrewAI multi‑modèles ?
Au minimum, journalisez :
workflow_idagentmodeltaskstart_timeend_timelatencystatusfallback_usedfallback_reasoninput_tokensoutput_tokenstotal_tokens
Ne journalisez pas :
API keysprivate credentialsfull sensitive promptsprivate user dataunredacted model output
Pour chaque modèle, surveillez :
Fiabilité
success ratetimeout rate5xx ratefallback rate
Performance
p50 latencyp95 latencyp99 latency
Coût
input tokensoutput tokenscost per taskcost per completed workflow
Qualité
task success ratehuman evaluationstructured-output validitytool-call success
Cela transforme le routage des modèles d’une préférence codée en dur en un système d’ingénierie observable.
Comment choisir entre des APIs directes de fournisseur et CometAPI ?
Le choix dépend de votre architecture.
| Architecture | Identifiants | Changement de modèle | Intégration fournisseur | Routage centralisé |
|---|---|---|---|---|
| APIs directes | Multiples | Personnalisé | Élevée | Non |
| Fournisseur unique | Un | Limité | Faible | Limité |
| CrewAI + CometAPI | Une clé CometAPI | Basé sur l’ID modèle | Plus faible | Oui |
Si votre application n’a besoin que d’un seul fournisseur et de ses capacités natives, une intégration directe peut être tout à fait raisonnable.
Si votre application CrewAI a besoin de modèles de plusieurs fournisseurs et que vous voulez une seule couche d’accès, CometAPI devient plus attractif.
L’important est que CometAPI ne remplace pas CrewAI.
Au lieu de cela :
CrewAIAgent orchestration ↓CometAPIModel access ↓Multiple models
Chaque couche a une responsabilité différente.
Comment cette architecture s’adapte‑t‑elle ?
Une fois la politique de routage séparée des définitions d’agent, l’ajout d’un autre modèle ne nécessite pas de reconstruire l’application entière.
Par exemple :
PRIMARY_MODELS = { "researcher": "gemini-3.7-flash", "analyst": "claude-opus-5", "writer": "gpt-5.6", "coder": "YOUR_CODE_MODEL",}
La même architecture peut alors prendre en charge :
Agent de rechercheAgent d’analyseAgent de développementAgent de revueAgent de rédactionAgent de vérification des faits
Chaque agent peut avoir un modèle différent tout en partageant la même couche d’accès CometAPI.
L’étape suivante est de rendre le routage dynamique.
Au lieu de :
"analyst": "claude-opus-5"
vous pourriez éventuellement utiliser :
select_model( task="analysis", budget=budget, latency_target=latency_target,)
Le système de routage peut alors choisir parmi des modèles approuvés en fonction des exigences de l’application.
Quelle est la meilleure architecture de production pour CrewAI + CometAPI ?
Pour un petit flux de travail :
User Input ↓CrewAI ↓CometAPI ↓Models
Pour la production :
┌───────────────┐ │ Model Catalog │ └───────┬───────┘ │ ▼User → CrewAI → Routing Policy → CometAPI │ │ │ │ │ ├── Gemini │ │ ├── Claude │ │ └── GPT │ │ │ ▼ │ Cost / Quality / │ Latency / Policy │ ▼ Checkpoints │ ▼ Usage Tracking
Les composants clés de production sont :
- Liste d’autorisation des modèles
- Routage par agent
- Réessais bornés
- Checkpointing des tâches
- Suivi d’usage
- Contrôles de coût
- Tests de compatibilité des modèles
- Observabilité
- Effets de bord idempotents
Cette architecture est bien plus robuste que d’ajouter simplement un try/except autour de crew.kickoff().
Une seule clé CometAPI, des modèles différents, des rôles d’agent plus clairs
La façon la plus utile de penser à CrewAI et CometAPI ensemble est comme deux couches complémentaires.
CrewAI définit ce que les agents font.
CometAPI définit comment ces agents accèdent aux modèles.
Cette séparation permet d’assigner un modèle rapide à un agent de recherche à fort volume, un modèle plus fort en raisonnement à un agent d’analyse, et un modèle généraliste de production au rédacteur final, sans maintenir des intégrations de fournisseurs séparées au sein du flux de travail.
L’implémentation la plus simple utilise une clé CometAPI et une URL de base compatible OpenAI :
https://api.cometapi.com/v1
En production, faites un pas de plus : conservez le routage des modèles dans la configuration, validez la disponibilité des modèles avant le déploiement, utilisez un repli borné uniquement pour les défaillances transitoires, checkpoint‑ez les tâches terminées, et enregistrez les métadonnées de modèle et d’usage pour chaque exécution.
Cela vous donne un schéma bien plus durable que de simplement connecter CrewAI à un seul LLM :
CrewAI orchestre les agents. CometAPI centralise l’accès aux modèles. Les IDs de modèles contrôlent le routage. Les checkpoints protègent le travail terminé. Le suivi d’usage contrôle le coût.
Foire aux questions
CrewAI peut‑il utiliser plusieurs modèles d’IA dans la même équipe ?
Oui. Assignez une configuration LLM différente à chaque agent CrewAI. Chaque configuration peut spécifier son propre modèle tout en utilisant la même clé API CometAPI et la même URL de base.
CrewAI peut‑il se connecter à une API compatible OpenAI ?
Oui. La configuration LLM de CrewAI prend en charge un base_url personnalisé et une clé API pour les endpoints compatibles OpenAI.
Avec CometAPI, l’URL de base est :
https://api.cometapi.com/v1
Ai‑je besoin de clés API séparées pour GPT, Claude et Gemini ?
En accédant à ces modèles via CometAPI, l’application peut utiliser la crédential et l’endpoint CometAPI plutôt que d’implémenter des identifiants de fournisseur séparés dans chaque agent CrewAI.
Une seule clé API signifie‑t‑elle que les modèles ont des capacités identiques ?
Non. L’interface API peut être unifiée alors que les capacités des modèles restent différentes. Les fenêtres de contexte, le support des outils, les paramètres, le comportement de sortie, la latence et la tarification peuvent varier selon les modèles.
Dois‑je réessayer tout le flux de travail CrewAI lorsqu’un modèle échoue ?
Uniquement pour des flux de travail simples et sans état. Un réessai de toute l’équipe peut répéter des tâches terminées et augmenter le coût. Pour les flux de travail de production, checkpoint‑ez les tâches terminées et reprenez à partir de la partie en échec lorsque c’est possible.
La fonctionnalité de checkpointing actuelle de CrewAI est conçue pour préserver l’état d’exécution et reprendre après des échecs.
Chaque exception CrewAI doit‑elle déclencher un repli de modèle ?
Non. L’authentification, les requêtes mal formées, les IDs de modèles invalides et les paramètres non pris en charge nécessitent généralement des changements de configuration plutôt qu’un modèle différent.
Le repli est mieux réservé aux défaillances transitoires bornées telles que les timeouts, les limites de débit et les réponses 5xx temporaires.
Comment suivre le coût de chaque agent CrewAI ?
Enregistrez le nom de l’agent, l’ID du modèle, la consommation de jetons, la latence, le statut d’exécution et les informations de repli pour chaque tâche. Utilisez ces données pour calculer le coût par agent et par flux de travail.
Puis‑je changer dynamiquement le modèle assigné à un agent ?
Oui. Conservez les IDs de modèles dans une configuration de routage plutôt que de les intégrer directement dans les définitions d’agents. Votre application peut alors choisir les modèles en fonction du coût, de la latence, du type de tâche ou de la disponibilité.
CometAPI remplace‑t‑il CrewAI ?
Non. Ils opèrent à des couches différentes. CrewAI orchestre les agents et les tâches, tandis que CometAPI fournit une couche d’accès aux modèles unifiée.
Où puis‑je trouver les modèles CometAPI actuels ?
Utilisez le répertoire des modèles CometAPI pour la découverte humaine des modèles et l’API des modèles pour la validation programmatique. La page de démarrage rapide actuelle de CometAPI répertorie plus de 500 modèles couvrant les catégories texte, image, vidéo et audio.