Crear un sistema multiagente con CrewAI resulta más interesante cuando diferentes agentes pueden usar distintos modelos.
Un investigador puede beneficiarse de un modelo rápido y económico, un analista puede necesitar un modelo con razonamiento más sólido y un redactor puede requerir un modelo optimizado para generación larga de alta calidad. Tradicionalmente, conectar estos agentes a diferentes proveedores implica gestionar credenciales de API separadas, endpoints, SDKs, sistemas de facturación y configuración específica del proveedor.
Una arquitectura más limpia es dejar que CrewAI gestione los agentes y el flujo de trabajo mientras CometAPI gestiona el acceso a los modelos.
CometAPI proporciona un endpoint compatible con OpenAI en https://api.cometapi.com/v1, para que las aplicaciones puedan enrutar solicitudes a modelos de múltiples proveedores a través de una interfaz de API común. Su documentación de inicio rápido actual también admite el uso del SDK estándar de Python de OpenAI cambiando la clave de API y la URL base.
En este tutorial, crearás un flujo de trabajo de CrewAI con tres agentes:
- Gemini 3.7 Flash para investigación
- Claude Opus 5 para análisis
- GPT-5.6 para la redacción final
- Una única clave de API de CometAPI
- Una única URL base de API
- Configuración de modelo por agente
- Fallback acotado para fallos transitorios
- Checkpointing de CrewAI para recuperación en producción
- Seguimiento de uso de tokens y ejecución
- Validación de modelos en el servidor
El límite arquitectónico importante es simple:
CrewAI maneja la orquestación de agentes. CometAPI maneja el acceso a modelos. Los IDs de modelo definen el enrutamiento.
¿Qué es el enrutamiento de modelos multiagente en CrewAI?
CrewAI es un framework de Python para crear agentes, tareas, “crews” y flujos de trabajo multiagente. Cada agente puede tener su propia configuración de LLM, mientras que el Crew coordina cómo esos agentes ejecutan tareas e intercambian contexto.
La configuración actual de LLM de CrewAI admite ajustes explícitos de model, api_key y base_url, incluyendo endpoints personalizados compatibles con OpenAI.
Esto hace que una arquitectura multimodelo sea sencilla:
CometAPI │ https://api.cometapi.com/v1 │ ┌───────────────────┼───────────────────┐ │ │ │ Researcher Analyst Writer │ │ │ Gemini 3.7 Flash Claude Opus 5 GPT-5.6
Los agentes permanecen separados desde una perspectiva lógica, pero su acceso a modelos se centraliza.
Esto es diferente a decir que todos los modelos son intercambiables. Una API compatible con OpenAI proporciona una interfaz de solicitud común; no garantiza límites de contexto idénticos, soporte de herramientas, controles de razonamiento, comportamiento de salida, latencia o precios.
Esa distinción importa al diseñar enrutamiento para producción.
¿Por qué usar CometAPI con CrewAI?
La principal ventaja no es que CrewAI se convierta de repente en un framework multiproveedor. CrewAI ya admite múltiples proveedores de LLM.
La ventaja es que el acceso a los modelos puede consolidarse detrás de una capa única de API.
Sin una capa de API unificada, un flujo de trabajo de tres agentes podría verse así:
| Agente | Proveedor | Credencial | Integración |
|---|---|---|---|
| Researcher | Google API key | Específica del prov | |
| Analyst | Anthropic | Anthropic API key | Específica del prov |
| Writer | OpenAI | OpenAI API key | Específica del prov |
Con CometAPI:
| Agente | Modelo | Credencial | Endpoint |
|---|---|---|---|
| Researcher | Gemini 3.7 Flash | CometAPI key | CometAPI |
| Analyst | Claude Opus 5 | CometAPI key | CometAPI |
| Writer | GPT-5.6 | CometAPI key | CometAPI |
La documentación de inicio rápido actual de CometAPI describe su endpoint como un reemplazo directo de la URL base de la API de OpenAI y lista modelos de múltiples proveedores a través del mismo servicio.
Esto le da a la aplicación una separación útil:
CrewAI
- Define roles de agente
- Define tareas
- Pasa contexto
- Controla la ejecución
- Gestiona iteraciones de agentes
- Maneja la orquestación a nivel de crew
CometAPI
- Proporciona una capa común de acceso a modelos
- Centraliza la autenticación de API
- Proporciona enrutamiento de modelos a través de IDs de modelo
- Da a la aplicación un único endpoint de API
- Ofrece visibilidad centralizada de uso y facturación
¿Qué construirá este flujo de trabajo de CrewAI?
El ejemplo crea tres agentes secuenciales.
| Agente de CrewAI | Modelo principal | Fallback | Rol |
|---|---|---|---|
| Market Researcher | gemini-3.7-flash | gpt-5.6 | Recopilar hechos e investigación |
| Product Analyst | claude-opus-5 | gpt-5.6 | Sintetizar evidencias y trade-offs |
| Technical Writer | gpt-5.6 | gemini-3.7-flash | Producir el memo de decisión final |
Esta es una política de enrutamiento de ejemplo, no un ranking de benchmarks.
El modelo adecuado para tu agente depende de:
- complejidad de la tarea
- longitud de contexto requerida
- uso de herramientas
- requisitos de salida estructurada
- latencia
- fiabilidad
- costo de tokens
- calidad de salida
- resultados de evaluación específicos de la aplicación
Una regla útil es:
Elige un modelo para el trabajo que realiza el agente, no simplemente por el proveedor del que proviene.
¿Qué modelo debería usar cada agente de CrewAI?
Para este ejemplo, la asignación de modelos sigue una estrategia simple de costo versus capacidad.
Researcher: Gemini 3.7 Flash
La investigación a menudo implica procesar cantidades relativamente grandes de información y producir un resultado intermedio compacto.
Un modelo rápido puede ser útil para tareas de investigación de alto volumen.
"researcher": "gemini-3.7-flash"
Analyst: Claude Opus 5
El analista tiene un rol más estrecho pero más intensivo en razonamiento. Recibe el resultado de investigación y lo convierte en una recomendación.
"analyst": "claude-opus-5"
Writer: GPT-5.6
El agente final convierte la investigación y el análisis en un memo de decisión orientado a desarrolladores.
"writer": "gpt-5.6"
Lo importante no son exactamente estas tres asignaciones. Tu aplicación debe evaluar modelos candidatos frente a tareas representativas antes de fijar la política de enrutamiento.
¿Qué necesitas antes de comenzar?
Necesitas:
- Python 3.10+
- CrewAI
- Compatibilidad con el SDK de Python de OpenAI
python-dotenv- Una clave de API de CometAPI
- Los IDs de modelo que pretendes usar
La integración actual de Python de CometAPI admite la API compatible con OpenAI, y el paquete oficial de Python de CometAPI documenta COMETAPI_KEY y COMETAPI_BASE_URL como opciones de configuración basadas en entorno.
El endpoint estándar es:
https://api.cometapi.com/v1
Antes de desplegar, verifica que los IDs de modelo seleccionados estén disponibles y soporten el endpoint y los parámetros requeridos por tu carga de trabajo de CrewAI. Los catálogos de modelos y los precios pueden cambiar.
¿Cómo instalas CrewAI y las dependencias?
Crea un nuevo entorno de Python:
python -m venv .venv
Actívalo:
source .venv/bin/activate
En Windows:
.venv\Scripts\Activate.ps1
Luego instala las dependencias:
pip install "crewai[openai]" openai python-dotenv
Usar openai explícitamente es intencional porque la implementación de fallback debajo importa directamente las clases de excepción del SDK de OpenAI.
Para producción, fija las versiones que pruebas en lugar de depender indefinidamente de las versiones más recientes flotantes.
Por ejemplo:
crewai==YOUR_TESTED_VERSIONopenai==YOUR_TESTED_VERSIONpython-dotenv==YOUR_TESTED_VERSION
La capa LLM de CrewAI evoluciona activamente, así que el constructor exacto y la configuración del proveedor se deben verificar con la versión de CrewAI usada por tu aplicación. La documentación actual de CrewAI admite configurar un LLM con un base_url personalizado y una clave de API.
¿Cómo configuras la clave de API de CometAPI?
Crea un archivo .env:
COMETAPI_KEY=your_cometapi_keyCOMETAPI_BASE_URL=https://api.cometapi.com/v1
Carga estos valores 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",)
Nunca confirmes .env en Git.
Agrégalo a .gitignore:
.env.venv/__pycache__/
La clave de API debe permanecer como credencial del lado del servidor. La guía de inicio rápido actual de CometAPI igualmente recomienda almacenar la clave en variables de entorno en lugar de código fuente.
¿Cómo conectas CrewAI a CometAPI?
El objeto LLM de CrewAI puede recibir un nombre de modelo, una clave de API y una URL base personalizada.
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, )
Esto es preferible a incrustar la misma configuración por separado en cada agente.
Cada agente ahora solo necesita un 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 qué establecer max_retries=0?
La razón es el control del fallback.
Si el cliente LLM subyacente reintenta automáticamente y tu aplicación también implementa fallback, un fallo puede convertirse en varias solicitudes ocultas antes de que la lógica de fallback se ejecute.
Para un tutorial con enrutamiento explícito, es más limpio dejar que la aplicación decida cuándo reintentar o cambiar de modelo.
¿Cómo defines la política de enrutamiento de modelos?
Mantén el enrutamiento fuera de tus 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",}
Esto crea un límite de configuración claro.
Más adelante puedes mover el mismo mapeo a:
- configuración de entorno
- YAML
- JSON
- una base de datos
- feature flags
- un servicio interno de enrutamiento de modelos
sin reescribir los prompts de los agentes.
¿Cómo construyes los tres agentes de CrewAI?
Crea un 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
La asignación de modelos ahora es completamente independiente de la definición del rol del agente.
Eso es lo que hace práctico el enrutamiento de modelos.
¿Cómo conectas los agentes con tareas secuenciales?
Crea tres tareas:
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 cadena de dependencias es:
Topic ↓Research ↓Analysis ↓Final memo
El analista recibe la salida de la tarea de investigación, mientras que el redactor recibe tanto el contexto de investigación como el de análisis.
¿Cómo construyes el Crew?
Combina los agentes y las tareas:
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, )
Ahora el enrutamiento de modelos es completamente impulsado por configuración.
Cambiar:
"researcher": "gemini-3.7-flash"
a otro modelo soportado no requiere cambiar el prompt de investigación o la definición de la tarea.
¿Cómo debería funcionar el fallback de modelos en CrewAI?
Aquí es donde una implementación orientada a producción requiere más cuidado.
Un error común es:
Any error ↓Switch model
Eso es demasiado agresivo.
Por ejemplo, estos errores generalmente no deberían activar fallback de modelo:
400 Bad Request401 Unauthorized403 Forbidden404 Not Found422 Validation Error
Cambiar de modelo no arreglará una clave de API inválida o una solicitud malformada.
El fallback es más apropiado para fallos temporales como:
408 Request Timeout429 Rate Limit500 Internal Server Error502 Bad Gateway503 Service Unavailable504 Gateway TimeoutConnection errorTimeout
Por lo tanto, la política de fallback debería ser:
Reintenta o cambia de modelo solo para fallos transitorios acotados y únicamente cuando el modelo de fallback soporte el mismo contrato de solicitud.
¿Cómo detectas errores reintentos?
Puedes usar las clases de error del SDK de 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
Esto excluye deliberadamente errores de configuración de la serie 400, excepto 408 y 429.
¿Deberías reintentar todo el Crew o solo el agente que falló?
Hay dos estrategias de fallback diferentes.
Fallback a nivel de crew
La implementación más simple es:
Start crew ↓failure ↓change routing ↓run crew again
Esto es fácil de entender, pero puede repetir tareas completadas.
Por ejemplo:
Research → completedAnalysis → completedWriter → failed
Un reintento completo de kickoff() puede ejecutar:
Research → againAnalysis → againWriter → fallback
Eso incrementa:
- uso de tokens
- latencia
- costo de API
- efectos secundarios potenciales
Recuperación a nivel de tarea
Un flujo de trabajo de producción debería, en cambio, hacer checkpoint del trabajo completado:
Research ↓checkpoint ↓Analysis ↓checkpoint ↓Writer fails ↓retry writer with fallback
CrewAI actualmente proporciona checkpointing que guarda el estado de ejecución y permite que una ejecución se reanude después de un fallo. El comportamiento de checkpoint documentado omite tareas completadas y continúa el trabajo aguas abajo desde el estado guardado.
Esta es la mejor arquitectura para flujos de trabajo costosos o que producen efectos secundarios.
¿Cómo agregas checkpointing en CrewAI?
Para flujos de trabajo de producción, habilita checkpointing en el crew:
crew = Crew( agents=[researcher, analyst, writer], tasks=[ research_task, analysis_task, writing_task, ], process=Process.sequential, checkpoint=True, verbose=True,)
El sistema de checkpointing de CrewAI puede persistir el estado de ejecución después de completar una tarea y restaurar el crew desde un checkpoint.
Por ejemplo, una ejecución restaurada puede usar:
from crewai import CheckpointConfigresult = crew.kickoff( from_checkpoint=CheckpointConfig( restore_from="./.checkpoints/checkpoint.json", ))
La configuración exacta de checkpoint debe seguir la versión de CrewAI utilizada por tu proyecto.
El punto arquitectónico importante es:
Primero checkpoint, luego fallback.
Esto evita que un fallo temporal de modelo fuerce la reejecución de trabajo completado costoso.
¿Cómo implementas un fallback simple y acotado?
Para un tutorial, aún puedes demostrar un fallback simple a nivel de 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
Observa la distinción importante:
Esto no afirma que se haya identificado el agente fallido.
Es una estrategia acotada de fallback a nivel de crew.
Para flujos pequeños sin estado, esto puede ser aceptable. Para flujos de producción con investigación costosa, herramientas o efectos secundarios, usa recuperación basada en checkpoints.
¿Cómo haces el seguimiento del uso de tokens en CrewAI?
El seguimiento de uso debe formar parte de la capa de enrutamiento, no ser una ocurrencia tardía.
Al final de la ejecución, inspecciona el resultado de 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)
Los campos de uso exactos disponibles pueden depender de la versión de CrewAI y de la ruta de ejecución, por lo que trata el objeto de resultado devuelto como la fuente de verdad para la versión que despliegues.
Un registro de uso de producción idealmente debería contener:
job_idagentmodelinput_tokensoutput_tokenstotal_tokenslatency_msfallback_usedfallback_reasonstatuscreated_at
Esto te permite responder preguntas como:
¿Qué agente consume la mayor parte del presupuesto?
¿Con qué frecuencia cae en fallback el analista?
¿Qué modelo tiene mayor latencia?
¿Cuánto cuesta cada flujo de trabajo?
¿Cómo controlas el costo a nivel de agente?
El enrutamiento multimodelo es más útil cuando refleja diferencias reales de carga de trabajo.
Por ejemplo:
Researcher→ high volume→ lower-cost modelAnalyst→ low volume→ stronger reasoning modelWriter→ medium volume→ general-purpose production model
También puedes constreñir el costo a través de la configuración del agente.
Por ejemplo:
max_iter=3
acota el bucle de iteración del agente. No debe interpretarse como un límite rígido de exactamente tres llamadas a la API o tres presupuestos de tokens.
Controles adicionales incluyen:
- limitar el contexto de la tarea
- resumir salidas intermedias
- cachear investigación repetible
- limitar tamaño máximo de entrada
- limitar tokens máximos de salida donde se soporte
- restringir llamadas a herramientas
- establecer presupuestos por usuario
- establecer presupuestos por flujo de trabajo
- rastrear la frecuencia de fallback
¿Cómo validas los modelos antes del despliegue?
No codifiques IDs de modelo para siempre.
Un modelo puede volverse:
- no disponible
- renombrado
- deprecado
- restringido
- cambiado en capacidad
- cambiado en precio
- incompatible con un parámetro que usa tu aplicación
CometAPI proporciona un endpoint de catálogo de modelos que se puede consultar programáticamente, mientras que su directorio público de modelos puede usarse para descubrimiento humano.
Una comprobación de despliegue puede verse así:
curl -s \ https://api.cometapi.com/api/models \ -H "Authorization: Bearer $COMETAPI_KEY"
Luego valida que tus IDs de modelo configurados existan antes del despliegue.
Por ejemplo, tu proceso de CI puede verificar:
gemini-3.7-flash → availableclaude-opus-5 → availablegpt-5.6 → available
No conviertas las comprobaciones de disponibilidad en sustituto de las pruebas de la aplicación. Que un modelo esté presente en un catálogo no significa que cada parámetro, herramienta o formato de salida usado por tu agente de CrewAI esté soportado.
¿Cómo se ve el ejemplo completo de CrewAI?
Aquí hay una implementación 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()
La mejora importante sobre la versión original es que el código ya no implica falsamente que una excepción identifica el agente exacto que falló.
Es explícitamente una implementación de fallback acotada a nivel de crew.
Para producción, combina la misma política de enrutamiento con checkpointing de CrewAI.
¿Cómo ejecutas el flujo de trabajo de CrewAI?
Guarda el archivo como:
crewai_multi_model.py
Luego ejecuta:
python crewai_multi_model.py \ "Should a small SaaS add AI-generated meeting summaries?"
Una respuesta exitosa contendrá información similar 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 respuesta exacta y los valores de uso dependen de la entrada, el comportamiento del modelo, la versión de CrewAI y la ruta de ejecución.
Si un error reintetable activa la ruta de fallback, el objeto selected_models muestra la ruta utilizada para esa ejecución del crew.
¿Cómo deberías diseñar el enrutamiento de modelos para producción?
Una política de enrutamiento de producción debe considerar más que la calidad del modelo.
Una función de decisión útil es:
Model Score =Quality+ Reliability+ Context Fit+ Tool Compatibility- Cost- Latency
Puedes implementarlo en varios niveles.
Enrutamiento basado en costo
Simple task → economical modelComplex task → premium model
Enrutamiento basado en latencia
Interactive request → fast modelBackground workflow → higher-quality model
Enrutamiento basado en fiabilidad
Primary model ↓transient failure ↓fallback model
Enrutamiento basado en tarea
Research → Model AAnalysis → Model BWriting → Model CCode → Model D
El último enfoque es especialmente natural para CrewAI porque el framework ya da a cada agente un rol distinto.
¿Cómo haces que el fallback sea seguro?
Un sistema de fallback robusto debe imponer cuatro reglas.
No hacer fallback en errores de autenticación
Si la clave de API es inválida:
401
cambiar de modelo no solucionará el problema.
No hacer fallback en solicitudes malformadas
Si la solicitud es inválida:
400422
corrige la solicitud en su lugar.
No hacer fallback indefinidamente
Establece un límite rígido:
MAX_FALLBACK_ATTEMPTS = 2
Un sistema de fallback sin límite puede convertirse en un costoso bucle de reintentos.
Asegura la compatibilidad de solicitudes de los modelos de fallback
Un modelo de fallback debe soportar las funciones que requiere tu agente.
Por ejemplo, si el agente principal requiere una herramienta específica o un comportamiento de salida estructurada, el fallback debe soportar ese mismo contrato.
Compatible con OpenAI no significa compatible en funcionalidades.
¿Cuáles son los errores más comunes de CrewAI + CometAPI?
| Síntoma | Causa probable | Solución |
|---|---|---|
| 401 Unauthorized | Clave de API inválida o ausente | Verifica COMETAPI_KEY; no hacer fallback |
| 400 Bad Request | Parámetros de solicitud inválidos | Corrige la solicitud |
| 404 Model Not Found | ID de modelo obsoleto | Verifica el catálogo de modelos actual |
| 408 Timeout | Timeout temporal de solicitud | Reintenta con política acotada |
| 429 Rate Limited | Demasiadas solicitudes | Retrocede y reintenta |
| 500–504 | Falla temporal servidor/gateway | Usa fallback acotado |
| Agent repeatedly retries | Reintentos ocultos del SDK | Controla max_retries |
| Completed tasks run again | Reintento completo del crew | Usa recuperación basada en checkpoints |
| Different model behaves differently | Capacidades del modelo difieren | Prueba cada modelo independientemente |
| Unexpected CrewAI constructor error | Desajuste de versión | Fija y verifica la versión de CrewAI |
¿Cómo separas errores de CrewAI de errores de modelo?
Esta distinción es importante al depurar.
Errores de configuración
Missing API keyInvalid model IDInvalid base URLUnsupported parameter
Estos deben fallar rápidamente.
Errores del proveedor/API
401403404429500503
Requieren manejo diferente según el estado.
Errores de la aplicación
Agent output invalidTool returned malformed dataTask context missingSide effect failed
Estos no necesariamente se solucionan cambiando de modelo.
Un sistema de agentes maduro debería, por lo tanto, tener manejo separado para:
configuration ↓API transport ↓model execution ↓agent logic ↓tool execution ↓application side effects
Esto es mucho más seguro que un genérico:
except Exception: use_fallback()
¿Cómo proteges efectos secundarios externos?
El fallback se complica significativamente cuando los agentes hacen más que generar texto.
Por ejemplo, imagina un agente que:
- crea un registro en base de datos
- envía un correo electrónico
- llama a una API externa
- actualiza un CRM
Si el modelo hace timeout después de que la acción externa se complete, volver a ejecutar todo el crew puede duplicar la acción.
Usa:
- claves de idempotencia
- checkpoints de tareas
- límites de transacción
- IDs de ejecución
- estado de tareas durables
- confirmación explícita de efectos secundarios
Por ejemplo:
job_id = crew_run_123task_id = writer_456
Almacena estos identificadores con operaciones externas para que un reintento pueda determinar si la operación ya ocurrió.
¿Cómo monitorizas flujos de trabajo multi-modelo en CrewAI?
Como mínimo, registra:
workflow_idagentmodeltaskstart_timeend_timelatencystatusfallback_usedfallback_reasoninput_tokensoutput_tokenstotal_tokens
No registres:
API keysprivate credentialsfull sensitive promptsprivate user dataunredacted model output
Para cada modelo, monitoriza:
Fiabilidad
success ratetimeout rate5xx ratefallback rate
Rendimiento
p50 latencyp95 latencyp99 latency
Costo
input tokensoutput tokenscost per taskcost per completed workflow
Calidad
task success ratehuman evaluationstructured-output validitytool-call success
Esto convierte el enrutamiento de modelos de una preferencia codificada en un sistema de ingeniería observable.
¿Cómo eliges entre APIs directas de proveedores y CometAPI?
La elección depende de tu arquitectura.
| Arquitectura | Credenciales | Cambio de modelo | Integración con proveedor | Enrutamiento centralizado |
|---|---|---|---|---|
| APIs directas | Múltiples | Personalizado | Alta | No |
| Proveedor único | Una | Limitado | Baja | Limitado |
| CrewAI + CometAPI | Una credencial CometAPI | Basado en ID | Menor | Sí |
Si tu aplicación solo necesita un proveedor y sus capacidades nativas, una integración directa puede ser perfectamente razonable.
Si tu aplicación de CrewAI necesita modelos de varios proveedores y quieres una capa única de acceso, CometAPI resulta más atractiva.
Lo importante es que CometAPI no reemplaza a CrewAI.
En su lugar:
CrewAIAgent orchestration ↓CometAPIModel access ↓Multiple models
Cada capa tiene una responsabilidad distinta.
¿Cómo escala esta arquitectura?
Una vez que la política de enrutamiento se separa de las definiciones de agentes, agregar otro modelo no requiere reconstruir toda la aplicación.
Por ejemplo:
PRIMARY_MODELS = { "researcher": "gemini-3.7-flash", "analyst": "claude-opus-5", "writer": "gpt-5.6", "coder": "YOUR_CODE_MODEL",}
La misma arquitectura puede entonces soportar:
Research agentAnalysis agentCoding agentReview agentWriting agentFact-checking agent
Cada agente puede tener un modelo distinto mientras comparte la misma capa de acceso de CometAPI.
El siguiente paso es hacer que el enrutamiento sea dinámico.
En lugar de:
"analyst": "claude-opus-5"
eventualmente podrías usar:
select_model( task="analysis", budget=budget, latency_target=latency_target,)
El sistema de enrutamiento puede entonces elegir entre modelos aprobados en función de los requisitos de la aplicación.
¿Cuál es la mejor arquitectura de producción para CrewAI + CometAPI?
Para un flujo pequeño:
User Input ↓CrewAI ↓CometAPI ↓Models
Para producción:
┌───────────────┐ │ Model Catalog │ └───────┬───────┘ │ ▼User → CrewAI → Routing Policy → CometAPI │ │ │ │ │ ├── Gemini │ │ ├── Claude │ │ └── GPT │ │ │ ▼ │ Cost / Quality / │ Latency / Policy │ ▼ Checkpoints │ ▼ Usage Tracking
Los componentes clave de producción son:
- Lista permitida de modelos
- Enrutamiento por agente
- Reintentos acotados
- Checkpointing de tareas
- Seguimiento de uso
- Controles de costo
- Pruebas de compatibilidad de modelos
- Observabilidad
- Efectos secundarios idempotentes
Esa arquitectura es sustancialmente más robusta que simplemente añadir un try/except alrededor de crew.kickoff().
Una única clave de CometAPI, diferentes modelos, roles de agente más claros
La forma más útil de pensar en CrewAI y CometAPI juntos es como dos capas complementarias.
CrewAI define lo que hacen los agentes.
CometAPI define cómo esos agentes acceden a los modelos.
Esa separación permite asignar un modelo rápido a un agente de investigación de alto volumen, un modelo de razonamiento más fuerte a un agente de análisis y un modelo de propósito general al redactor final sin mantener integraciones de proveedores separadas dentro del flujo de trabajo.
La implementación más simple usa una clave de CometAPI y una URL base compatible con OpenAI:
https://api.cometapi.com/v1
Para producción, lleva la arquitectura un paso más allá: mantén el enrutamiento de modelos en configuración, valida la disponibilidad de modelos antes del despliegue, usa fallback acotado solo para fallos transitorios, haz checkpoint de las tareas completadas y registra el modelo y los metadatos de uso para cada ejecución.
Eso te da un patrón mucho más duradero que simplemente conectar CrewAI a un único LLM:
CrewAI orquesta los agentes. CometAPI centraliza el acceso a modelos. Los IDs de modelo controlan el enrutamiento. Los checkpoints protegen el trabajo completado. El seguimiento de uso controla el costo.
Preguntas frecuentes
¿Puede CrewAI usar múltiples modelos de IA en el mismo crew?
Sí. Asigna una configuración de LLM diferente a cada agente de CrewAI. Cada configuración puede especificar su propio modelo mientras usa la misma clave de API y URL base de CometAPI.
¿Puede CrewAI conectarse a una API compatible con OpenAI?
Sí. La configuración de LLM de CrewAI admite un base_url personalizado y una clave de API para endpoints compatibles con OpenAI.
Con CometAPI, la URL base es:
https://api.cometapi.com/v1
¿Necesito claves de API separadas para GPT, Claude y Gemini?
Al acceder a estos modelos a través de CometAPI, la aplicación puede usar la credencial y el endpoint de CometAPI en lugar de implementar credenciales de proveedores separados en cada agente de CrewAI.
¿Significa una sola clave de API que los modelos tienen capacidades idénticas?
No. La interfaz de la API puede estar unificada mientras que las capacidades de los modelos siguen siendo diferentes. Las ventanas de contexto, el soporte de herramientas, los parámetros, el comportamiento de salida, la latencia y los precios pueden variar por modelo.
¿Debería reintentar todo el flujo de trabajo de CrewAI cuando falla un modelo?
Solo para flujos simples y sin estado. Un reintento de todo el crew puede repetir tareas completadas e incrementar el costo. Para flujos de producción, haz checkpoint de las tareas completadas y reanuda desde la parte fallida cuando sea práctico.
La funcionalidad actual de checkpointing de CrewAI está diseñada para preservar el estado de ejecución y reanudar después de fallos.
¿Debería cada excepción de CrewAI activar fallback de modelo?
No. La autenticación, las solicitudes malformadas, los IDs de modelo inválidos y los parámetros no soportados generalmente requieren cambios de configuración en lugar de un modelo diferente.
El fallback es mejor reservarlo para fallos transitorios acotados como timeouts, límites de tasa y respuestas 5xx temporales.
¿Cómo hago seguimiento del costo de cada agente de CrewAI?
Registra el nombre del agente, el ID de modelo, el uso de tokens, la latencia, el estado de ejecución y la información de fallback para cada tarea. Usa los datos resultantes para calcular el costo por agente y por flujo de trabajo.
¿Puedo cambiar dinámicamente el modelo asignado a un agente?
Sí. Mantén los IDs de modelo en una configuración de enrutamiento en lugar de incrustarlos directamente en las definiciones de los agentes. Tu aplicación puede entonces elegir modelos basados en costo, latencia, tipo de tarea o disponibilidad.
¿CometAPI reemplaza a CrewAI?
No. Operan en diferentes capas. CrewAI orquesta agentes y tareas, mientras que CometAPI proporciona una capa unificada de acceso a modelos.
¿Dónde puedo encontrar los modelos actuales de CometAPI?
Usa el Directorio de modelos de CometAPI para descubrimiento humano y la API de modelos para validación programática. La página de inicio rápido actual de CometAPI lista 500+ modelos en categorías de texto, imagen, video y audio.