La creazione di un sistema multi-agente con CrewAI diventa più interessante quando agenti diversi possono usare modelli diversi.
Un ricercatore può beneficiare di un modello veloce ed economico, un analista può aver bisogno di un modello con capacità di ragionamento più forti e uno scrittore può richiedere un modello ottimizzato per la generazione long-form di alta qualità. Tradizionalmente, collegare questi agenti a provider diversi significa gestire credenziali API, endpoint, SDK, sistemi di fatturazione e configurazioni specifiche per provider separati.
Un’architettura più pulita è lasciare che CrewAI gestisca gli agenti e il workflow mentre CometAPI gestisce l’accesso ai modelli.
CometAPI fornisce un endpoint compatibile con OpenAI su https://api.cometapi.com/v1, quindi le applicazioni possono instradare le richieste a modelli di più provider attraverso un’interfaccia API comune. La sua documentazione di quick start attuale supporta anche l’uso dell’SDK Python standard di OpenAI cambiando la chiave API e la base URL.
In questo tutorial, costruirai un workflow CrewAI con tre agenti e:
- Gemini 3.7 Flash per la ricerca
- Claude Opus 5 per l’analisi
- GPT-5.6 per la stesura finale
- Una chiave API CometAPI
- Una base URL API
- Configurazione del modello per agente
- Fallback limitato per errori transitori
- Checkpointing di CrewAI per il ripristino in produzione
- Tracciamento dell’uso di token ed esecuzioni
- Validazione dei modelli lato server
Il confine architetturale importante è semplice:
CrewAI gestisce l’orchestrazione degli agenti. CometAPI gestisce l’accesso ai modelli. Gli ID modello definiscono il routing.
Che cos’è il routing dei modelli multi‑agente in CrewAI?
CrewAI è un framework Python per creare agenti, task, crew e workflow multi-agente. Ogni agente può avere la propria configurazione LLM, mentre il Crew coordina come tali agenti eseguono i task e si scambiano contesto.
L’attuale configurazione LLM di CrewAI supporta impostazioni esplicite di model, api_key e base_url, inclusi endpoint personalizzati compatibili con OpenAI.
Questo rende lineare un’architettura multi‑modello:
CometAPI │ https://api.cometapi.com/v1 │ ┌───────────────────┼───────────────────┐ │ │ │ Researcher Analyst Writer │ │ │ Gemini 3.7 Flash Claude Opus 5 GPT-5.6
Gli agenti restano separati da un punto di vista logico, ma il loro accesso ai modelli è centralizzato.
Questo è diverso dal dire che tutti i modelli siano intercambiabili. Un’API compatibile con OpenAI fornisce un’interfaccia di richiesta comune; non garantisce identici limiti di contesto, supporto agli strumenti, controlli di ragionamento, comportamento dell’output, latenza o prezzi.
Questa distinzione conta quando si progetta il routing in produzione.
Perché usare CometAPI con CrewAI?
Il vantaggio principale non è che CrewAI diventi improvvisamente un framework multi‑provider. CrewAI supporta già più provider LLM.
Il vantaggio è che l’accesso ai modelli può essere consolidato dietro un unico strato API.
Senza uno strato API unificato, un workflow a tre agenti potrebbe essere così:
| Agente | Provider | Credenziale | Integrazione |
|---|---|---|---|
| Ricercatore | Chiave API Google | Specifica del provider | |
| Analista | Anthropic | Chiave API Anthropic | Specifica del provider |
| Scrittore | OpenAI | Chiave API OpenAI | Specifica del provider |
Con CometAPI:
| Agente | Modello | Credenziale | Endpoint |
|---|---|---|---|
| Ricercatore | Gemini 3.7 Flash | Chiave CometAPI | CometAPI |
| Analista | Claude Opus 5 | Chiave CometAPI | CometAPI |
| Scrittore | GPT-5.6 | Chiave CometAPI | CometAPI |
La documentazione di quick start attuale di CometAPI descrive il suo endpoint come sostituzione drop‑in della base URL dell’API OpenAI e elenca modelli di più provider tramite lo stesso servizio.
Questo offre all’applicazione una separazione utile:
CrewAI
- Definisce i ruoli degli agenti
- Definisce i task
- Passa il contesto
- Controlla l’esecuzione
- Gestisce le iterazioni degli agenti
- Gestisce l’orchestrazione a livello di crew
CometAPI
- Fornisce uno strato comune di accesso ai modelli
- Centralizza l’autenticazione API
- Fornisce il routing ai modelli attraverso gli ID modello
- Offre un unico endpoint API all’applicazione
- Fornisce visibilità centralizzata su usage e billing
Cosa costruirà questo workflow CrewAI?
L’esempio crea tre agenti sequenziali.
| Agente CrewAI | Modello primario | Fallback | Ruolo |
|---|---|---|---|
| Ricercatore di mercato | gemini-3.7-flash | gpt-5.6 | Raccogliere fatti e fare ricerca |
| Analista di prodotto | claude-opus-5 | gpt-5.6 | Sintetizzare evidenze e trade‑off |
| Redattore tecnico | gpt-5.6 | gemini-3.7-flash | Produrre il memo decisionale finale |
Questa è una policy di routing di esempio, non una classifica benchmark.
Il modello giusto per il tuo agente dipende da:
- complessità del task
- lunghezza di contesto richiesta
- uso di strumenti
- requisiti di output strutturato
- latenza
- affidabilità
- costo in token
- qualità dell’output
- risultati di valutazioni specifiche dell’applicazione
Una regola utile è:
Scegli un modello in base al lavoro che l’agente svolge, non semplicemente al provider di provenienza.
Quale modello dovrebbe usare ciascun agente CrewAI?
Per questo esempio, l’assegnazione dei modelli segue una strategia semplice costo‑vs‑capacità.
Ricercatore: Gemini 3.7 Flash
La ricerca spesso implica l’elaborazione di quantità relativamente ampie di informazioni e la produzione di un risultato intermedio compatto.
Un modello veloce può quindi essere utile per compiti di ricerca ad alto volume.
"researcher": "gemini-3.7-flash"
Analista: Claude Opus 5
L’analista ha un ruolo più ristretto ma più intensivo in termini di ragionamento. Riceve l’output della ricerca e lo trasforma in una raccomandazione.
"analyst": "claude-opus-5"
Scrittore: GPT-5.6
L’agente finale converte la ricerca e l’analisi in un memo decisionale per sviluppatori.
"writer": "gpt-5.6"
La parte importante non sono queste tre assegnazioni esatte. La tua applicazione dovrebbe valutare i modelli candidati su task rappresentativi prima di fissare la policy di routing.
Cosa serve prima di iniziare?
Ti servono:
- Python 3.10+
- CrewAI
- Compatibilità con l’SDK Python di OpenAI
python-dotenv- Una chiave API CometAPI
- Gli ID dei modelli che intendi usare
L’integrazione Python attuale di CometAPI supporta l’API compatibile con OpenAI, e il pacchetto Python ufficiale di CometAPI documenta COMETAPI_KEY e COMETAPI_BASE_URL come opzioni di configurazione basate su variabili d’ambiente.
L’endpoint standard è:
https://api.cometapi.com/v1
Prima del deploy, verifica che gli ID modello selezionati siano attualmente disponibili e supportino l’endpoint e i parametri richiesti dal tuo carico di lavoro CrewAI. I cataloghi dei modelli e i prezzi possono cambiare.
Come installare CrewAI e le dipendenze?
Crea un nuovo ambiente Python:
python -m venv .venv
Attivalo:
source .venv/bin/activate
Su Windows:
.venv\Scripts\Activate.ps1
Poi installa le dipendenze:
pip install "crewai[openai]" openai python-dotenv
Usare esplicitamente openai è intenzionale perché l’implementazione di fallback qui sotto importa direttamente le classi di eccezioni dell’SDK OpenAI.
In produzione, blocca le versioni che testi invece di affidarti indefinitamente alle ultime versioni floating.
Ad esempio:
crewai==YOUR_TESTED_VERSIONopenai==YOUR_TESTED_VERSIONpython-dotenv==YOUR_TESTED_VERSION
Il layer LLM di CrewAI è in rapida evoluzione, quindi il costruttore esatto e la configurazione del provider dovrebbero essere verificati rispetto alla versione di CrewAI usata dalla tua applicazione. La documentazione attuale di CrewAI supporta la configurazione di un LLM con base_url personalizzata e chiave API.
Come configurare la chiave API di CometAPI?
Crea un file .env:
COMETAPI_KEY=your_cometapi_keyCOMETAPI_BASE_URL=https://api.cometapi.com/v1
Carica questi valori 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",)
Non fare mai commit di .env su Git.
Aggiungilo a .gitignore:
.env.venv/__pycache__/
La chiave API dovrebbe rimanere una credenziale lato server. Le linee guida di quick start di CometAPI raccomandano allo stesso modo di archiviare la chiave in variabili d’ambiente anziché nel codice sorgente.
Come collegare CrewAI a CometAPI?
L’oggetto LLM di CrewAI può ricevere un nome di modello, una chiave API e una base URL personalizzata.
Crea un helper:
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, )
Questo è preferibile rispetto a incorporare la stessa configurazione separatamente in ogni agente.
Ora ogni agente ha bisogno solo di un ID modello:
research_llm = cometapi_llm("gemini-3.7-flash")analysis_llm = cometapi_llm("claude-opus-5")writing_llm = cometapi_llm("gpt-5.6")
Perché impostare max_retries=0?
Il motivo è il controllo del fallback.
Se il client LLM sottostante ritenta automaticamente e la tua applicazione implementa anche un fallback, un singolo errore può diventare diverse richieste nascoste prima che la logica di fallback venga eseguita.
Per un tutorial con routing esplicito, è più pulito lasciare che sia l’applicazione a decidere quando ritentare o cambiare modello.
Come definire la policy di routing dei modelli?
Mantieni il routing fuori dai prompt:
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",}
Questo crea un chiaro confine di configurazione.
Potrai poi spostare la stessa mappatura in:
- configurazione d’ambiente
- YAML
- JSON
- un database
- feature flag
- un servizio interno di model routing
senza riscrivere i prompt degli agenti.
Come costruire i tre agenti CrewAI?
Crea un oggetto LLM per agente.
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’assegnazione dei modelli è ora completamente indipendente dalla definizione del ruolo dell’agente.
Questo è ciò che rende pratico il model routing.
Come collegare gli agenti con task sequenziali?
Crea tre task:
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 catena di dipendenze è:
Topic ↓Research ↓Analysis ↓Final memo
L’analista riceve l’output del task di ricerca, mentre lo scrittore riceve sia il contesto di ricerca sia quello di analisi.
Come costruire la crew?
Combina agenti e task:
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, )
Ora il routing dei modelli è interamente guidato dalla configurazione.
Cambiare:
"researcher": "gemini-3.7-flash"
con un altro modello supportato non richiede modifiche al prompt o alla definizione del task di ricerca.
Come dovrebbe funzionare il fallback dei modelli in CrewAI?
Qui un’implementazione orientata alla produzione richiede più attenzione.
Un errore comune è:
Any error ↓Switch model
Questo è troppo aggressivo.
Ad esempio, questi errori in genere non dovrebbero innescare un fallback del modello:
400 Bad Request401 Unauthorized403 Forbidden404 Not Found422 Validation Error
Cambiare modello non risolverà una chiave API non valida o una richiesta mal formata.
Il fallback è più appropriato per guasti temporanei come:
408 Request Timeout429 Rate Limit500 Internal Server Error502 Bad Gateway503 Service Unavailable504 Gateway TimeoutConnection errorTimeout
La policy di fallback dovrebbe quindi essere:
Ritentare o cambiare modello solo per errori transitori e limitati, e solo quando il modello di fallback supporta lo stesso contratto di richiesta.
Come rilevare errori ritentabili?
Puoi usare le classi di errore dell’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
Questo esclude deliberatamente gli errori di configurazione di livello 400 diversi da 408 e 429.
Dovresti ritentare l’intera crew o solo l’agente che è fallito?
Esistono due diverse strategie di fallback.
Fallback a livello di crew
L’implementazione più semplice è:
Start crew ↓failure ↓change routing ↓run crew again
È facile da capire, ma può ripetere i task già completati.
Ad esempio:
Research → completedAnalysis → completedWriter → failed
Un retry completo di kickoff() può eseguire:
Research → againAnalysis → againWriter → fallback
Questo aumenta:
- uso di token
- latenza
- costo API
- potenziali effetti collaterali
Ripristino a livello di task
Un workflow di produzione dovrebbe invece fare checkpoint del lavoro completato:
Research ↓checkpoint ↓Analysis ↓checkpoint ↓Writer fails ↓retry writer with fallback
CrewAI fornisce attualmente un checkpointing che salva lo stato di esecuzione e permette a una run di riprendere dopo un errore. Il comportamento di checkpoint documentato salta i task completati e continua il lavoro a valle dallo stato salvato.
Questa è l’architettura migliore per workflow costosi o con effetti collaterali.
Come aggiungere il checkpointing di CrewAI?
Per workflow di produzione, abilita il checkpointing sulla crew:
crew = Crew( agents=[researcher, analyst, writer], tasks=[ research_task, analysis_task, writing_task, ], process=Process.sequential, checkpoint=True, verbose=True,)
Il sistema di checkpointing di CrewAI può persistere lo stato di esecuzione dopo il completamento dei task e ripristinare la crew da un checkpoint.
Ad esempio, una run ripristinata può usare:
from crewai import CheckpointConfigresult = crew.kickoff( from_checkpoint=CheckpointConfig( restore_from="./.checkpoints/checkpoint.json", ))
La configurazione esatta del checkpoint dovrebbe seguire la versione di CrewAI usata dal tuo progetto.
Il punto architetturale importante è:
Checkpoint prima, fallback poi.
Questo impedisce che un guasto temporaneo del modello costringa a rieseguire costoso lavoro già completato.
Come implementare un semplice fallback limitato?
Per un tutorial, puoi comunque dimostrare un fallback semplice a livello di crew.
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
Nota la distinzione importante:
Questo non afferma di aver identificato l’esatto agente fallito.
È una strategia di fallback limitata a livello di crew.
Per piccoli workflow stateless, può essere accettabile. Per workflow di produzione con ricerca costosa, strumenti o effetti collaterali, usa un ripristino basato su checkpoint.
Come tracciare l’uso dei token in CrewAI?
Il tracciamento dell’uso dovrebbe far parte del layer di routing, non un ripensamento.
Alla fine della run, ispeziona il risultato di 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)
I campi di usage esatti disponibili possono dipendere dalla versione di CrewAI e dal percorso di esecuzione, quindi tratta l’oggetto result restituito come fonte di verità per la versione che distribuisci.
Un record di usage di produzione dovrebbe idealmente contenere:
job_idagentmodelinput_tokensoutput_tokenstotal_tokenslatency_msfallback_usedfallback_reasonstatuscreated_at
Questo ti permette di rispondere a domande come:
Quale agente consuma la maggior parte del budget?
Con quale frequenza l’analista ricorre al fallback?
Quale modello ha la latenza più alta?
Quanto costa ogni workflow?
Come controllare i costi a livello di agente?
Il routing multi‑modello è più utile quando riflette le differenze reali del carico di lavoro.
Ad esempio:
Ricerca → alto volume → modello a costo inferioreAnalisi → basso volume → modello con ragionamento più forteScrittura → volume medio → modello di produzione generico
Puoi anche vincolare i costi tramite configurazione dell’agente.
Ad esempio:
max_iter=3
limita il loop di iterazione dell’agente. Non dovrebbe essere interpretato come un limite rigido di esattamente tre chiamate API o tre budget di token.
Controlli aggiuntivi includono:
- limitare il contesto del task
- riassumere gli output intermedi
- mettere in cache ricerche ripetibili
- limitare la dimensione massima dell’input
- limitare i token massimi di output dove supportato
- restringere le chiamate agli strumenti
- impostare budget per utente
- impostare budget per workflow
- tracciare la frequenza del fallback
Come validare i modelli prima del deploy?
Non fissare per sempre gli ID modello nel codice.
Un modello può diventare:
- non disponibile
- rinominato
- deprecato
- soggetto a restrizioni
- cambiato in capacità
- cambiato nel prezzo
- incompatibile con un parametro usato dall’applicazione
CometAPI fornisce un endpoint del catalogo dei modelli che può essere interrogato programmaticamente, mentre la sua directory pubblica dei modelli può essere usata per la scoperta umana.
Un controllo di deploy può essere simile a:
curl -s \ https://api.cometapi.com/api/models \ -H "Authorization: Bearer $COMETAPI_KEY"
Poi valida che gli ID modello configurati esistano prima del deploy.
Ad esempio, il tuo processo CI può verificare:
gemini-3.7-flash → availableclaude-opus-5 → availablegpt-5.6 → available
Non sostituire i controlli di disponibilità con i test dell’applicazione. La presenza di un modello in un catalogo non significa che ogni parametro, strumento o formato di output usato dal tuo agente CrewAI sia supportato.
Com’è l’esempio completo di CrewAI?
Ecco un’implementazione consolidata:
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()
Il miglioramento importante rispetto alla versione originale è che il codice non implica più erroneamente che un’eccezione identifichi l’agente esattamente fallito.
È esplicitamente un’implementazione di fallback limitato a livello di crew.
In produzione, combina la stessa policy di routing con il checkpointing di CrewAI.
Come eseguire il workflow CrewAI?
Salva il file come:
crewai_multi_model.py
Poi esegui:
python crewai_multi_model.py \ "Should a small SaaS add AI-generated meeting summaries?"
Una risposta positiva conterrà informazioni simili a:
{ "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 risposta e i valori di usage esatti dipendono dall’input, dal comportamento dei modelli, dalla versione di CrewAI e dal percorso di esecuzione.
Se un errore ritentabile attiva il percorso di fallback, l’oggetto selected_models mostra la route usata per quell’esecuzione della crew.
Come progettare il routing dei modelli in produzione?
Una policy di routing di produzione dovrebbe considerare più della qualità del modello.
Una funzione decisionale utile è:
Model Score =Quality+ Reliability+ Context Fit+ Tool Compatibility- Cost- Latency
Puoi implementarla a più livelli.
Routing basato sui costi
Task semplice → modello economicoTask complesso → modello premium
Routing basato sulla latenza
Richiesta interattiva → modello veloceWorkflow in background → modello di qualità superiore
Routing basato sull’affidabilità
Modello primario ↓errore transitorio ↓modello di fallback
Routing basato sul task
Ricerca → Modello AAnalisi → Modello BScrittura → Modello CCodice → Modello D
L’ultimo approccio è particolarmente naturale per CrewAI perché il framework fornisce già a ciascun agente un ruolo distinto.
Come rendere sicuro il fallback?
Un sistema di fallback robusto dovrebbe applicare quattro regole.
Non fare fallback su errori di autenticazione
Se la chiave API non è valida:
401
cambiare modello non risolverà il problema.
Non fare fallback su richieste mal formate
Se la richiesta non è valida:
400422
correggi la richiesta invece.
Non fare fallback indefinitamente
Imposta un limite rigido:
MAX_FALLBACK_ATTEMPTS = 2
Un sistema di fallback senza limite può diventare un costoso loop di retry.
Rendi i modelli di fallback compatibili con la richiesta
Un modello di fallback deve supportare le funzionalità richieste dal tuo agente.
Ad esempio, se l’agente primario richiede un tool specifico o un comportamento di output strutturato, il fallback deve supportare lo stesso contratto.
Compatibile con OpenAI non significa compatibile nelle funzionalità.
Quali sono gli errori più comuni CrewAI + CometAPI?
| Sintomo | Causa probabile | Correzione |
|---|---|---|
| 401 Unauthorized | Chiave API mancante o invalida | Controlla COMETAPI_KEY; non fare fallback |
| 400 Bad Request | Parametri di richiesta non validi | Correggi la richiesta |
| 404 Model Not Found | ID modello obsoleto | Controlla il catalogo modelli attuale |
| 408 Timeout | Timeout temporaneo della richiesta | Ritenta con policy limitata |
| 429 Rate Limited | Troppe richieste | Backoff e retry |
| 500–504 | Guasto temporaneo server/gateway | Usa fallback limitato |
| L’agente ritenta ripetutamente | Retry nascosti dell’SDK | Controlla max_retries |
| Task completati rieseguiti | Retry della crew intera | Usa ripristino basato su checkpoint |
| Modello diverso si comporta diversamente | Differenze di capacità tra modelli | Testa ogni modello indipendentemente |
| Errore di costruttore CrewAI inatteso | Mismatch di versione | Fissa e verifica la versione di CrewAI |
Come separare errori di CrewAI da errori del modello?
Questa distinzione è importante durante il debug.
Errori di configurazione
Missing API keyInvalid model IDInvalid base URLUnsupported parameter
Questi dovrebbero fallire rapidamente.
Errori del provider/API
401403404429500503
Richiedono gestioni diverse a seconda dello status.
Errori dell’applicazione
Agent output invalidTool returned malformed dataTask context missingSide effect failed
Non sono necessariamente risolti cambiando modello.
Un sistema di agenti maturo dovrebbe quindi gestire separatamente:
configuration ↓API transport ↓model execution ↓agent logic ↓tool execution ↓application side effects
Questo è molto più sicuro di un generico:
except Exception: use_fallback()
Come proteggere gli effetti collaterali esterni?
Il fallback diventa significativamente più complicato quando gli agenti fanno più che generare testo.
Per esempio, immagina un agente che:
- crea un record nel database
- invia un’email
- chiama un’API esterna
- aggiorna un CRM
Se il modello va in timeout dopo che l’azione esterna è riuscita, rieseguire l’intera crew può duplicare l’azione.
Usa:
- chiavi di idempotenza
- checkpoint dei task
- confini di transazione
- execution ID
- stato durabile dei task
- conferma esplicita degli effetti collaterali
Ad esempio:
job_id = crew_run_123task_id = writer_456
Archivia questi identificatori con le operazioni esterne così che un retry possa determinare se l’operazione è già avvenuta.
Come monitorare workflow CrewAI multi‑modello?
Al minimo, registra:
workflow_idagentmodeltaskstart_timeend_timelatencystatusfallback_usedfallback_reasoninput_tokensoutput_tokenstotal_tokens
Non registrare:
API keysprivate credentialsfull sensitive promptsprivate user dataunredacted model output
Per ciascun modello, monitora:
Affidabilità
success ratetimeout rate5xx ratefallback rate
Performance
p50 latencyp95 latencyp99 latency
Costo
input tokensoutput tokenscost per taskcost per completed workflow
Qualità
task success ratehuman evaluationstructured-output validitytool-call success
Questo trasforma il model routing da una preferenza codificata a un sistema ingegneristico osservabile.
Come scegliere tra API dirette dei provider e CometAPI?
La scelta dipende dalla tua architettura.
| Architettura | Credenziali | Cambio modello | Integrazione provider | Routing centralizzato |
|---|---|---|---|---|
| API dirette provider | Multiple | Personalizzato | Elevata | No |
| Provider unico | Una | Limitato | Bassa | Limitato |
| CrewAI + CometAPI | Una credenziale CometAPI | Basato su ID modello | Inferiore | Sì |
Se la tua applicazione ha bisogno di un solo provider e delle sue capacità native, un’integrazione diretta può essere perfettamente ragionevole.
Se la tua applicazione CrewAI necessita di modelli da più provider e vuoi uno strato di accesso unico, CometAPI diventa più attraente.
Il punto importante è che CometAPI non sostituisce CrewAI.
Invece:
CrewAIOrchestrazione agenti ↓CometAPIAccesso ai modelli ↓Più modelli
Ogni layer ha una responsabilità diversa.
Come scala questa architettura?
Una volta separata la policy di routing dalle definizioni degli agenti, aggiungere un altro modello non richiede di ricostruire l’intera applicazione.
Ad esempio:
PRIMARY_MODELS = { "researcher": "gemini-3.7-flash", "analyst": "claude-opus-5", "writer": "gpt-5.6", "coder": "YOUR_CODE_MODEL",}
La stessa architettura può quindi supportare:
Agente di ricercaAgente di analisiAgente di codingAgente di reviewAgente di scritturaAgente di fact‑checking
Ogni agente può avere un modello diverso condividendo lo stesso strato di accesso CometAPI.
Il passo successivo è rendere dinamico il routing.
Invece di:
"analyst": "claude-opus-5"
potresti in futuro usare:
select_model( task="analysis", budget=budget, latency_target=latency_target,)
Il sistema di routing può quindi scegliere tra modelli approvati in base ai requisiti dell’applicazione.
Qual è la migliore architettura di produzione per CrewAI + CometAPI?
Per un piccolo workflow:
User Input ↓CrewAI ↓CometAPI ↓Models
Per la produzione:
┌───────────────┐ │ Model Catalog │ └───────┬───────┘ │ ▼User → CrewAI → Routing Policy → CometAPI │ │ │ │ │ ├── Gemini │ │ ├── Claude │ │ └── GPT │ │ │ ▼ │ Cost / Quality / │ Latency / Policy │ ▼ Checkpoints │ ▼ Usage Tracking
I componenti chiave di produzione sono:
- Allowlist di modelli
- Routing per agente
- Retry limitati
- Checkpointing dei task
- Tracciamento dell’uso
- Controlli dei costi
- Test di compatibilità dei modelli
- Osservabilità
- Effetti collaterali idempotenti
Quell’architettura è sostanzialmente più robusta che aggiungere semplicemente un try/except intorno a crew.kickoff().
Una sola chiave CometAPI, modelli diversi, ruoli degli agenti più chiari
Il modo più utile di pensare a CrewAI e CometAPI insieme è come due layer complementari.
CrewAI definisce cosa fanno gli agenti.
CometAPI definisce come quegli agenti accedono ai modelli.
Questa separazione rende possibile assegnare un modello veloce a un agente di ricerca ad alto volume, un modello con ragionamento più forte a un agente di analisi e un modello generico a uno scrittore finale senza mantenere integrazioni specifiche dei provider all’interno del workflow.
L’implementazione più semplice usa una sola chiave CometAPI e una sola base URL compatibile con OpenAI:
https://api.cometapi.com/v1
Per la produzione, porta l’architettura un passo oltre: mantieni il routing dei modelli nella configurazione, valida la disponibilità dei modelli prima del deploy, usa fallback limitati solo per errori transitori, fai checkpoint dei task completati e registra metadati su modello e usage per ogni run.
Questo ti dà un pattern molto più durevole che collegare semplicemente CrewAI a un unico LLM:
CrewAI orchestra gli agenti. CometAPI centralizza l’accesso ai modelli. Gli ID modello controllano il routing. I checkpoint proteggono il lavoro completato. Il tracciamento dell’uso controlla i costi.
Domande frequenti
CrewAI può usare più modelli di IA nella stessa crew?
Sì. Assegna una configurazione LLM diversa a ciascun agente CrewAI. Ogni configurazione può specificare il proprio modello usando la stessa chiave API e base URL di CometAPI.
CrewAI può connettersi a un’API compatibile con OpenAI?
Sì. La configurazione LLM di CrewAI supporta una base_url personalizzata e una chiave API per endpoint compatibili con OpenAI.
Con CometAPI, la base URL è:
https://api.cometapi.com/v1
Ho bisogno di chiavi API separate per GPT, Claude e Gemini?
Quando accedi a questi modelli tramite CometAPI, l’applicazione può usare la credenziale e l’endpoint di CometAPI invece di implementare credenziali del provider separate in ogni agente CrewAI.
Una sola chiave API significa che i modelli hanno capacità identiche?
No. L’interfaccia API può essere unificata mentre le capacità dei modelli restano diverse. Finestra di contesto, supporto agli strumenti, parametri, comportamento dell’output, latenza e prezzi possono variare per modello.
Dovrei ritentare l’intero workflow CrewAI quando un modello fallisce?
Solo per workflow semplici e stateless. Un retry dell’intera crew può ripetere task completati e aumentare i costi. Per workflow di produzione, fai checkpoint dei task completati e riprendi dalla parte fallita dove possibile.
La funzionalità di checkpointing attuale di CrewAI è progettata per preservare lo stato di esecuzione e riprendere dopo i fallimenti.
Ogni eccezione di CrewAI dovrebbe attivare il fallback del modello?
No. Autenticazione, richieste mal formate, ID modello non validi e parametri non supportati richiedono generalmente modifiche di configurazione piuttosto che un modello diverso.
Il fallback è meglio riservarlo a guasti transitori limitati, come timeout, rate limit e risposte 5xx temporanee.
Come posso tracciare il costo di ciascun agente CrewAI?
Registra nome dell’agente, ID modello, uso di token, latenza, stato di esecuzione e informazioni sul fallback per ciascun task. Usa i dati risultanti per calcolare il costo per agente e per workflow.
Posso cambiare dinamicamente il modello assegnato a un agente?
Sì. Mantieni gli ID modello in una configurazione di routing anziché incorporarli direttamente nelle definizioni degli agenti. La tua applicazione può quindi scegliere i modelli in base a costo, latenza, tipo di task o disponibilità.
CometAPI sostituisce CrewAI?
No. Operano a layer diversi. CrewAI orchestra agenti e task, mentre CometAPI fornisce uno strato unificato di accesso ai modelli.
Dove posso trovare i modelli attuali di CometAPI?
Usa la directory dei modelli CometAPI per la scoperta umana dei modelli e l’API dei modelli per la validazione programmatica. La pagina di quick start attuale di CometAPI elenca oltre 500 modelli tra categorie testo, immagine, video e audio.