Construir um sistema multiagente com CrewAI fica mais interessante quando agentes diferentes podem usar modelos diferentes.
Um pesquisador pode se beneficiar de um modelo rápido e econômico, um analista pode precisar de um modelo com raciocínio mais forte, e um redator pode exigir um modelo otimizado para geração de textos longos de alta qualidade. Tradicionalmente, conectar esses agentes a provedores diferentes significa gerenciar credenciais de API separadas, endpoints, SDKs, sistemas de cobrança e configuração específica de cada provedor.
Uma arquitetura mais limpa é deixar o CrewAI gerenciar os agentes e o fluxo de trabalho enquanto o CometAPI gerencia o acesso aos modelos.
O CometAPI fornece um endpoint compatível com OpenAI em https://api.cometapi.com/v1, para que aplicativos possam encaminhar solicitações a modelos de vários provedores por meio de uma interface de API comum. Sua documentação atual de início rápido também suporta o uso do SDK Python padrão da OpenAI alterando a chave de API e a base URL.
Neste tutorial, você construirá um fluxo de trabalho CrewAI com três agentes:
- Gemini 3.7 Flash para pesquisa
- Claude Opus 5 para análise
- GPT-5.6 para redação final
- Uma chave de API do CometAPI
- Uma base URL de API
- Configuração de modelo por agente
- Fallback limitado para falhas transitórias
- Checkpointing do CrewAI para recuperação em produção
- Rastreamento de uso de tokens e execução
- Validação de modelo no servidor
A fronteira arquitetural importante é simples:
CrewAI cuida da orquestração dos agentes. CometAPI cuida do acesso aos modelos. IDs de modelo definem o roteamento.
O que é roteamento de modelos multiagente no CrewAI?
CrewAI é um framework Python para criar agentes, tarefas, equipes (crews) e fluxos de trabalho multiagente. Cada agente pode ter sua própria configuração de LLM, enquanto o Crew coordena como esses agentes executam tarefas e trocam contexto.
A configuração atual de LLM do CrewAI suporta definições explícitas de model, api_key e base_url, incluindo endpoints personalizados compatíveis com OpenAI.
Isso torna a arquitetura multimodelo direta:
CometAPI │ https://api.cometapi.com/v1 │ ┌───────────────────┼───────────────────┐ │ │ │ Researcher Analyst Writer │ │ │ Gemini 3.7 Flash Claude Opus 5 GPT-5.6
Os agentes permanecem separados do ponto de vista lógico, mas seu acesso aos modelos é centralizado.
Isso é diferente de dizer que todos os modelos são intercambiáveis. Uma API compatível com OpenAI fornece uma interface de solicitação comum; ela não garante limites de contexto idênticos, suporte a ferramentas, controles de raciocínio, comportamento de saída, latência ou preços.
Essa distinção importa ao projetar o roteamento em produção.
Por que usar CometAPI com CrewAI?
A principal vantagem não é que o CrewAI de repente se torna um framework multiprovedor. O CrewAI já suporta vários provedores de LLM.
A vantagem é que o acesso aos modelos pode ser consolidado por trás de uma única camada de API.
Sem uma camada de API unificada, um fluxo de trabalho de três agentes pode ser assim:
| Agente | Provedor | Credencial | Integração |
|---|---|---|---|
| Pesquisador | Chave de API Google | Específica do provedor | |
| Analista | Anthropic | Chave de API Anthropic | Específica do provedor |
| Redator | OpenAI | Chave de API OpenAI | Específica do provedor |
Com CometAPI:
| Agente | Modelo | Credencial | Endpoint |
|---|---|---|---|
| Pesquisador | Gemini 3.7 Flash | Chave do CometAPI | CometAPI |
| Analista | Claude Opus 5 | Chave do CometAPI | CometAPI |
| Redator | GPT-5.6 | Chave do CometAPI | CometAPI |
A documentação atual de início rápido do CometAPI descreve seu endpoint como um substituto direto para a base URL da API da OpenAI e lista modelos de vários provedores por meio do mesmo serviço.
Isso dá ao aplicativo uma separação útil:
CrewAI
- Define papéis dos agentes
- Define tarefas
- Transmite contexto
- Controla execução
- Gerencia iterações dos agentes
- Lida com a orquestração no nível da equipe
CometAPI
- Fornece uma camada comum de acesso a modelos
- Centraliza autenticação de API
- Proporciona roteamento por meio de IDs de modelo
- Dá ao aplicativo um endpoint único de API
- Oferece visibilidade centralizada de uso e cobrança
O que este fluxo de trabalho CrewAI vai construir?
O exemplo cria três agentes sequenciais.
| Agente CrewAI | Modelo primário | Fallback | Papel |
|---|---|---|---|
| Market Researcher | gemini-3.7-flash | gpt-5.6 | Coletar fatos e pesquisa |
| Product Analyst | claude-opus-5 | gpt-5.6 | Sintetizar evidências e trade-offs |
| Technical Writer | gpt-5.6 | gemini-3.7-flash | Produzir o memorando de decisão final |
Este é um exemplo de política de roteamento, não um ranking de benchmark.
O modelo certo para seu agente depende de:
- complexidade da tarefa
- comprimento de contexto necessário
- uso de ferramentas
- requisitos de saída estruturada
- latência
- confiabilidade
- custo de tokens
- qualidade da saída
- resultados de avaliação específicos do aplicativo
Uma regra útil é:
Escolha um modelo para o trabalho que o agente executa, não simplesmente pelo provedor.
Qual modelo cada agente CrewAI deve usar?
Para este exemplo, a atribuição de modelos segue uma estratégia simples de custo versus capacidade.
Pesquisador: Gemini 3.7 Flash
Pesquisa frequentemente envolve processar quantidades relativamente grandes de informação e produzir um resultado intermediário compacto.
Um modelo rápido pode ser útil para tarefas de pesquisa de alto volume.
"researcher": "gemini-3.7-flash"
Analista: Claude Opus 5
O analista tem um papel mais estreito porém intensivo em raciocínio. Ele recebe a saída da pesquisa e a transforma em uma recomendação.
"analyst": "claude-opus-5"
Redator: GPT-5.6
O agente final converte a pesquisa e a análise em um memorando de decisão voltado para desenvolvedores.
"writer": "gpt-5.6"
A parte importante não são estas três atribuições exatas. Seu aplicativo deve avaliar modelos candidatos em tarefas representativas antes de fixar a política de roteamento.
O que você precisa antes de começar?
Você precisa de:
- Python 3.10+
- CrewAI
- Compatibilidade com o SDK Python da OpenAI
python-dotenv- Uma chave de API do CometAPI
- Os IDs de modelo que você pretende usar
A integração Python atual do CometAPI suporta a API compatível com OpenAI, e o pacote Python oficial do CometAPI documenta COMETAPI_KEY e COMETAPI_BASE_URL como opções de configuração baseadas em ambiente.
O endpoint padrão é:
https://api.cometapi.com/v1
Antes de implantar, verifique se os IDs de modelo selecionados estão disponíveis e suportam o endpoint e os parâmetros exigidos pelo seu workload no CrewAI. Catálogos de modelos e preços podem mudar.
Como instalar CrewAI e dependências?
Crie um novo ambiente Python:
python -m venv .venv
Ative-o:
source .venv/bin/activate
No Windows:
.venv\Scripts\Activate.ps1
Depois instale as dependências:
pip install "crewai[openai]" openai python-dotenv
Usar openai explicitamente é intencional porque a implementação de fallback abaixo importa classes de exceção do SDK da OpenAI diretamente.
Para produção, fixe as versões que você testar em vez de depender indefinidamente de versões flutuantes mais recentes.
Por exemplo:
crewai==YOUR_TESTED_VERSIONopenai==YOUR_TESTED_VERSIONpython-dotenv==YOUR_TESTED_VERSION
A camada de LLM do CrewAI está evoluindo ativamente, então o construtor exato e a configuração de provedor devem ser verificados em relação à versão do CrewAI usada pelo seu aplicativo. A documentação atual do CrewAI suporta configurar um LLM com base_url personalizada e chave de API.
Como configurar a chave de API do CometAPI?
Crie um arquivo .env:
COMETAPI_KEY=your_cometapi_keyCOMETAPI_BASE_URL=https://api.cometapi.com/v1
Carregue esses valores em 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",)
Nunca faça commit do .env no Git.
Adicione-o ao .gitignore:
.env.venv/__pycache__/
A chave de API deve permanecer uma credencial do lado do servidor. A orientação atual de início rápido do CometAPI igualmente recomenda armazenar a chave em variáveis de ambiente em vez de código-fonte.
Como conectar CrewAI ao CometAPI?
O objeto LLM do CrewAI pode receber um nome de modelo, chave de API e base URL personalizada.
Crie um 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, )
Isso é preferível a embutir a mesma configuração separadamente em cada agente.
Cada agente agora precisa apenas de um ID de modelo:
research_llm = cometapi_llm("gemini-3.7-flash")analysis_llm = cometapi_llm("claude-opus-5")writing_llm = cometapi_llm("gpt-5.6")
Por que definir max_retries=0?
A razão é o controle de fallback.
Se o cliente LLM subjacente tenta automaticamente novamente e seu aplicativo também implementa fallback, uma falha pode se tornar várias solicitações ocultas antes que a lógica de fallback seja executada.
Para um tutorial com roteamento explícito, é mais limpo deixar o aplicativo decidir quando tentar novamente ou alternar modelos.
Como definir a política de roteamento de modelos?
Mantenha o roteamento fora dos seus 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",}
Isso cria uma fronteira de configuração clara.
Você pode mover mais tarde o mesmo mapeamento para:
- configuração de ambiente
- YAML
- JSON
- um banco de dados
- feature flags
- um serviço interno de roteamento de modelos
sem reescrever os prompts dos agentes.
Como construir os três agentes do CrewAI?
Crie um objeto LLM por 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
A atribuição do modelo agora é completamente independente da definição de papel do agente.
Isso é o que torna o roteamento de modelos prático.
Como conectar os agentes com tarefas sequenciais?
Crie três tarefas:
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
A cadeia de dependência é:
Topic ↓Research ↓Analysis ↓Final memo
O analista recebe a saída da tarefa de pesquisa, enquanto o redator recebe o contexto de pesquisa e de análise.
Como construir a equipe (Crew)?
Combine os agentes e tarefas:
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, )
Agora o roteamento de modelos é totalmente dirigido por configuração.
Alterar:
"researcher": "gemini-3.7-flash"
para outro modelo suportado não requer mudar o prompt de pesquisa ou a definição de tarefa.
Como deve funcionar o fallback de modelo no CrewAI?
Aqui é onde uma implementação orientada para produção precisa de mais cuidado.
Um erro comum é:
Any error ↓Switch model
Isso é agressivo demais.
Por exemplo, estes erros geralmente não devem acionar fallback de modelo:
400 Bad Request401 Unauthorized403 Forbidden404 Not Found422 Validation Error
Trocar de modelo não vai corrigir uma chave de API inválida ou uma solicitação malformada.
Fallback é mais apropriado para falhas temporárias como:
408 Request Timeout429 Rate Limit500 Internal Server Error502 Bad Gateway503 Service Unavailable504 Gateway TimeoutConnection errorTimeout
A política de fallback deve ser:
Tentar novamente ou trocar de modelo apenas para falhas transitórias e limitadas e somente quando o modelo de fallback suporta o mesmo contrato de solicitação.
Como detectar erros passíveis de retry?
Você pode usar as classes de erro do SDK da 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
Isso exclui deliberadamente erros de configuração na faixa 400, exceto 408 e 429.
Você deve tentar novamente toda a equipe ou apenas o agente que falhou?
Há duas estratégias diferentes de fallback.
Fallback no nível da equipe (crew)
A implementação mais simples é:
Start crew ↓failure ↓change routing ↓run crew again
Isso é fácil de entender, mas pode repetir tarefas concluídas.
Por exemplo:
Research → completedAnalysis → completedWriter → failed
Um retry completo de kickoff() pode executar:
Research → againAnalysis → againWriter → fallback
Isso aumenta:
- uso de tokens
- latência
- custo de API
- possíveis efeitos colaterais
Recuperação no nível da tarefa
Um fluxo de trabalho de produção deve, em vez disso, fazer checkpoint do trabalho concluído:
Research ↓checkpoint ↓Analysis ↓checkpoint ↓Writer fails ↓retry writer with fallback
O CrewAI atualmente fornece checkpointing que salva o estado de execução e permite que uma execução seja retomada após uma falha. O comportamento de checkpoint documentado pula tarefas concluídas e continua o trabalho a jusante a partir do estado salvo.
Esta é a melhor arquitetura para fluxos de trabalho caros ou com produção de efeitos colaterais.
Como adicionar checkpointing no CrewAI?
Para fluxos de trabalho de produção, habilite checkpointing na equipe:
crew = Crew( agents=[researcher, analyst, writer], tasks=[ research_task, analysis_task, writing_task, ], process=Process.sequential, checkpoint=True, verbose=True,)
O sistema de checkpointing do CrewAI pode persistir o estado de execução após a conclusão da tarefa e restaurar a equipe a partir de um ponto de verificação.
Por exemplo, uma execução restaurada pode usar:
from crewai import CheckpointConfigresult = crew.kickoff( from_checkpoint=CheckpointConfig( restore_from="./.checkpoints/checkpoint.json", ))
A configuração exata de checkpoint deve seguir a versão do CrewAI usada pelo seu projeto.
O ponto arquitetural importante é:
Checkpoint primeiro, fallback depois.
Isso evita que uma falha temporária de modelo force a repetição de trabalho caro já concluído.
Como implementar um fallback simples e limitado?
Para um tutorial, você ainda pode demonstrar um fallback simples no nível da equipe.
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
Observe a distinção importante:
Isto não afirma que o agente que falhou foi identificado.
É uma estratégia limitada de fallback no nível da equipe.
Para pequenos fluxos de trabalho sem estado, isso pode ser aceitável. Para fluxos de produção com pesquisa cara, ferramentas ou efeitos colaterais, use recuperação baseada em checkpoint.
Como rastrear o uso de tokens no CrewAI?
O rastreamento de uso deve fazer parte da camada de roteamento, não ser um pensamento tardio.
Ao final da execução, inspecione o resultado do 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)
Os campos exatos de uso disponíveis podem depender da versão do CrewAI e do caminho de execução, portanto trate o objeto de resultado retornado como fonte de verdade para a versão que você implantar.
Um registro de uso de produção idealmente deve conter:
job_idagentmodelinput_tokensoutput_tokenstotal_tokenslatency_msfallback_usedfallback_reasonstatuscreated_at
Isso permite responder perguntas como:
Qual agente está consumindo mais do orçamento?
Com que frequência o analista faz fallback?
Qual modelo tem a maior latência?
Quanto custa cada fluxo de trabalho?
Como controlar custo no nível do agente?
O roteamento multimodelo é mais útil quando reflete diferenças reais de carga de trabalho.
Por exemplo:
Researcher→ high volume→ lower-cost modelAnalyst→ low volume→ stronger reasoning modelWriter→ medium volume→ general-purpose production model
Você também pode restringir o custo por meio da configuração do agente.
Por exemplo:
max_iter=3
limita o loop de iteração do agente. Isso não deve ser interpretado como um limite rígido de exatamente três chamadas de API ou três orçamentos de tokens.
Controles adicionais incluem:
- limitar o contexto da tarefa
- resumir saídas intermediárias
- fazer cache de pesquisas repetíveis
- limitar tamanho máximo de entrada
- limitar tokens máximos de saída onde suportado
- restringir chamadas de ferramentas
- definir orçamentos por usuário
- definir orçamentos por fluxo de trabalho
- rastrear frequência de fallback
Como validar modelos antes da implantação?
Não codifique IDs de modelo para sempre.
Um modelo pode se tornar:
- indisponível
- renomeado
- descontinuado
- restrito
- alterado em capacidade
- alterado em preço
- incompatível com um parâmetro que seu aplicativo usa
O CometAPI fornece um endpoint de catálogo de modelos que pode ser consultado programaticamente, enquanto seu diretório público de modelos pode ser usado para descoberta humana de modelos.
Uma verificação de implantação pode ser:
curl -s \ https://api.cometapi.com/api/models \ -H "Authorization: Bearer $COMETAPI_KEY"
Depois valide que os IDs de modelo configurados existem antes da implantação.
Por exemplo, seu processo de CI pode verificar:
gemini-3.7-flash → availableclaude-opus-5 → availablegpt-5.6 → available
Não faça das verificações de disponibilidade um substituto para testes do aplicativo. Um modelo estar presente no catálogo não significa que todo parâmetro, ferramenta ou formato de saída usado pelo seu agente CrewAI seja suportado.
Como é o exemplo completo do CrewAI?
Aqui está uma implementação consolidada:
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()
A melhoria importante em relação à versão original é que o código não implica erroneamente que uma exceção identifica exatamente o agente que falhou.
É explicitamente uma implementação limitada de fallback no nível da equipe.
Para produção, combine a mesma política de roteamento com checkpointing do CrewAI.
Como executar o fluxo de trabalho do CrewAI?
Salve o arquivo como:
crewai_multi_model.py
Depois execute:
python crewai_multi_model.py \ "Should a small SaaS add AI-generated meeting summaries?"
Uma resposta bem-sucedida conterá informações similares 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>"}
A resposta exata e os valores de uso dependem da entrada, do comportamento do modelo, da versão do CrewAI e do caminho de execução.
Se um erro passível de retry ativar a rota de fallback, o objeto selected_models mostra a rota usada para aquela execução da equipe.
Como projetar o roteamento de modelos em produção?
Uma política de roteamento de produção deve considerar mais do que qualidade do modelo.
Uma função de decisão útil é:
Model Score =Quality+ Reliability+ Context Fit+ Tool Compatibility- Cost- Latency
Você pode implementar isso em vários níveis.
Roteamento baseado em custo
Simple task → economical modelComplex task → premium model
Roteamento baseado em latência
Interactive request → fast modelBackground workflow → higher-quality model
Roteamento baseado em confiabilidade
Primary model ↓transient failure ↓fallback model
Roteamento baseado em tarefa
Research → Model AAnalysis → Model BWriting → Model CCode → Model D
A última abordagem é especialmente natural para CrewAI porque o framework já dá a cada agente um papel distinto.
Como tornar o fallback seguro?
Um sistema de fallback robusto deve impor quatro regras.
Não faça fallback em erros de autenticação
Se a chave de API for inválida:
401
trocar de modelo não vai resolver o problema.
Não faça fallback em solicitações malformadas
Se a solicitação for inválida:
400422
corrija a solicitação em vez disso.
Não faça fallback indefinidamente
Defina um limite rígido:
MAX_FALLBACK_ATTEMPTS = 2
Um sistema de fallback sem limite pode se tornar um loop caro de tentativas.
Torne os modelos de fallback compatíveis com a solicitação
Um modelo de fallback deve suportar os recursos que seu agente exige.
Por exemplo, se o agente primário requer uma ferramenta específica ou comportamento de saída estruturada, o fallback deve suportar o mesmo contrato.
Compatível com OpenAI não significa compatível em recursos.
Quais são os erros mais comuns de CrewAI + CometAPI?
| Sintoma | Causa provável | Correção |
|---|---|---|
| 401 Unauthorized | Chave de API inválida ou ausente | Verifique COMETAPI_KEY; não fazer fallback |
| 400 Bad Request | Parâmetros de solicitação inválidos | Corrigir a solicitação |
| 404 Model Not Found | ID de modelo desatualizado | Verificar o catálogo atual de modelos |
| 408 Timeout | Tempo de solicitação expirado temporário | Tentar novamente dentro de uma política limitada |
| 429 Rate Limited | Solicitações demais | Reduzir e tentar novamente |
| 500–504 | Falha temporária de servidor/gateway | Usar fallback limitado |
| Agente tenta novamente repetidamente | Retries ocultos do SDK | Controlar max_retries |
| Tarefas concluídas rodam novamente | Retry da equipe completa | Usar recuperação baseada em checkpoint |
| Modelo diferente se comporta diferente | Capacidades do modelo diferem | Testar cada modelo independentemente |
| Erro inesperado no construtor do CrewAI | Incompatibilidade de versão | Fixar e verificar versão do CrewAI |
Como separar erros do CrewAI de erros de modelo?
Essa distinção é importante ao depurar.
Erros de configuração
Missing API keyInvalid model IDInvalid base URLUnsupported parameter
Estes devem falhar rapidamente.
Erros do provedor/API
401403404429500503
Estes requerem tratamento diferente dependendo do status.
Erros do aplicativo
Agent output invalidTool returned malformed dataTask context missingSide effect failed
Estes não são necessariamente resolvidos trocando de modelo.
Um sistema de agentes maduro deve, portanto, ter tratamento separado para:
configuration ↓API transport ↓model execution ↓agent logic ↓tool execution ↓application side effects
Isso é muito mais seguro do que um genérico:
except Exception: use_fallback()
Como proteger efeitos colaterais externos?
Fallback torna-se significativamente mais complicado quando os agentes fazem mais do que gerar texto.
Por exemplo, imagine um agente que:
- cria um registro no banco de dados
- envia um e-mail
- chama uma API externa
- atualiza um CRM
Se o modelo atingir timeout após a ação externa ter sido bem-sucedida, rodar novamente toda a equipe pode duplicar a ação.
Use:
- chaves de idempotência
- checkpoints de tarefas
- limites de transação
- IDs de execução
- estado durável de tarefa
- confirmação explícita de efeitos colaterais
Por exemplo:
job_id = crew_run_123task_id = writer_456
Armazene esses identificadores com operações externas para que um retry possa determinar se a operação já aconteceu.
Como monitorar fluxos de trabalho multimodelo no CrewAI?
No mínimo, registre:
workflow_idagentmodeltaskstart_timeend_timelatencystatusfallback_usedfallback_reasoninput_tokensoutput_tokenstotal_tokens
Não registre:
API keysprivate credentialsfull sensitive promptsprivate user dataunredacted model output
Para cada modelo, monitore:
Confiabilidade
success ratetimeout rate5xx ratefallback rate
Desempenho
p50 latencyp95 latencyp99 latency
Custo
input tokensoutput tokenscost per taskcost per completed workflow
Qualidade
task success ratehuman evaluationstructured-output validitytool-call success
Isso transforma o roteamento de modelos de uma preferência codificada em um sistema de engenharia observável.
Como escolher entre APIs diretas de provedores e CometAPI?
A escolha depende da sua arquitetura.
| Arquitetura | Credenciais | Troca de modelo | Integração com provedor | Roteamento centralizado |
|---|---|---|---|---|
| APIs diretas | Múltiplas | Personalizada | Alta | Não |
| Único provedor | Uma | Limitada | Baixa | Limitada |
| CrewAI + CometAPI | Uma credencial do CometAPI | Baseada em ID de modelo | Menor | Sim |
Se seu aplicativo precisa apenas de um provedor e seus recursos nativos, uma integração direta pode ser perfeitamente razoável.
Se seu aplicativo CrewAI precisa de modelos de vários provedores e você quer uma camada única de acesso, o CometAPI se torna mais atraente.
O ponto importante é que CometAPI não substitui CrewAI.
Em vez disso:
CrewAIAgent orchestration ↓CometAPIModel access ↓Multiple models
Cada camada tem uma responsabilidade diferente.
Como essa arquitetura escala?
Depois que a política de roteamento fica separada das definições de agente, adicionar outro modelo não exige reconstruir todo o aplicativo.
Por exemplo:
PRIMARY_MODELS = { "researcher": "gemini-3.7-flash", "analyst": "claude-opus-5", "writer": "gpt-5.6", "coder": "YOUR_CODE_MODEL",}
A mesma arquitetura pode então suportar:
Research agentAnalysis agentCoding agentReview agentWriting agentFact-checking agent
Cada agente pode ter um modelo diferente enquanto compartilha a mesma camada de acesso do CometAPI.
O próximo passo é tornar o roteamento dinâmico.
Em vez de:
"analyst": "claude-opus-5"
você poderia eventualmente usar:
select_model( task="analysis", budget=budget, latency_target=latency_target,)
O sistema de roteamento pode então escolher entre modelos aprovados com base nos requisitos do aplicativo.
Qual é a melhor arquitetura de produção para CrewAI + CometAPI?
Para um pequeno fluxo de trabalho:
User Input ↓CrewAI ↓CometAPI ↓Models
Para produção:
┌───────────────┐ │ Model Catalog │ └───────┬───────┘ │ ▼User → CrewAI → Routing Policy → CometAPI │ │ │ │ │ ├── Gemini │ │ ├── Claude │ │ └── GPT │ │ │ ▼ │ Cost / Quality / │ Latency / Policy │ ▼ Checkpoints │ ▼ Usage Tracking
Os componentes de produção essenciais são:
- Lista de modelos permitidos (allowlist)
- Roteamento por agente
- Retries limitados
- Checkpointing de tarefas
- Rastreamento de uso
- Controles de custo
- Testes de compatibilidade de modelos
- Observabilidade
- Efeitos colaterais idempotentes
Essa arquitetura é substancialmente mais robusta do que simplesmente adicionar um try/except em torno de crew.kickoff().
Uma chave CometAPI, modelos diferentes, papéis de agente mais claros
A maneira mais útil de pensar sobre CrewAI e CometAPI juntos é como duas camadas complementares.
CrewAI define o que os agentes fazem.
CometAPI define como esses agentes acessam modelos.
Essa separação torna possível atribuir um modelo rápido a um agente de pesquisa de alto volume, um modelo com raciocínio mais forte para um agente de análise e um modelo de uso geral para o redator final sem manter integrações separadas de provedores dentro do fluxo de trabalho.
A implementação mais simples usa uma chave do CometAPI e uma base URL compatível com OpenAI:
https://api.cometapi.com/v1
Para produção, leve a arquitetura um passo adiante: mantenha o roteamento de modelos na configuração, valide a disponibilidade de modelos antes da implantação, use fallback limitado apenas para falhas transitórias, faça checkpoint das tarefas concluídas e registre metadados de modelo e uso para cada execução.
Isso dá um padrão muito mais durável do que simplesmente conectar CrewAI a um único LLM:
CrewAI orquestra os agentes. CometAPI centraliza o acesso aos modelos. IDs de modelo controlam o roteamento. Checkpoints protegem o trabalho concluído. O rastreamento de uso controla o custo.
Perguntas frequentes
O CrewAI pode usar vários modelos de IA na mesma equipe?
Sim. Atribua uma configuração de LLM diferente a cada agente do CrewAI. Cada configuração pode especificar seu próprio modelo usando a mesma chave de API do CometAPI e base URL.
O CrewAI pode se conectar a uma API compatível com OpenAI?
Sim. A configuração de LLM do CrewAI suporta uma base_url personalizada e chave de API para endpoints compatíveis com OpenAI.
Com CometAPI, a base URL é:
https://api.cometapi.com/v1
Preciso de chaves de API separadas para GPT, Claude e Gemini?
Ao acessar esses modelos por meio do CometAPI, o aplicativo pode usar a credencial e o endpoint do CometAPI em vez de implementar credenciais separadas de provedores em cada agente do CrewAI.
Uma única chave de API significa que os modelos têm capacidades idênticas?
Não. A interface da API pode ser unificada enquanto as capacidades dos modelos permanecem diferentes. Janelas de contexto, suporte a ferramentas, parâmetros, comportamento de saída, latência e preços podem variar por modelo.
Devo tentar novamente todo o fluxo de trabalho do CrewAI quando um modelo falhar?
Apenas para fluxos simples e sem estado. Um retry da equipe inteira pode repetir tarefas concluídas e aumentar custos. Para fluxos de trabalho de produção, faça checkpoint das tarefas concluídas e retome a partir da parte que falhou quando for viável.
A funcionalidade atual de checkpoint do CrewAI é projetada para preservar o estado de execução e retomar após falhas.
Toda exceção do CrewAI deve acionar fallback de modelo?
Não. Autenticação, solicitações malformadas, IDs de modelo inválidos e parâmetros não suportados geralmente exigem mudanças de configuração em vez de um modelo diferente.
Fallback é melhor para falhas transitórias e limitadas como timeouts, limites de taxa e respostas 5xx temporárias.
Como rastrear o custo de cada agente do CrewAI?
Registre o nome do agente, ID do modelo, uso de tokens, latência, status de execução e informações de fallback para cada tarefa. Use os dados resultantes para calcular o custo por agente e por fluxo de trabalho.
Posso alterar dinamicamente o modelo atribuído a um agente?
Sim. Mantenha os IDs de modelo em uma configuração de roteamento em vez de embuti-los diretamente nas definições de agentes. Seu aplicativo pode então escolher modelos com base em custo, latência, tipo de tarefa ou disponibilidade.
O CometAPI substitui o CrewAI?
Não. Eles operam em camadas diferentes. CrewAI orquestra agentes e tarefas, enquanto CometAPI fornece uma camada unificada de acesso a modelos.
Onde posso encontrar os modelos atuais do CometAPI?
Use o diretório de modelos do CometAPI para descoberta humana de modelos e a API de modelos para validação programática. A página de início rápido atual do CometAPI lista 500+ modelos em categorias de texto, imagem, vídeo e áudio.
