AI Endpoints - Intégration avec Apache Airflow

Découvrez comment orchestrer des charges de travail IA sur les AI Endpoints OVHcloud via Apache Airflow

Info

AI Endpoints est couvert par les Conditions particulières OVHcloud Public Cloud (voir l’Annexe 10 – « Conditions spécifiques – AI Endpoints »).

Nouvelle intégration disponible : nous sommes ravis d'annoncer une nouvelle intégration pour AI Endpoints avec Apache Airflow. Cette intégration vous permet d'orchestrer facilement des charges de travail IA sur l'infrastructure OVHcloud directement depuis vos DAG Airflow, et poursuit notre engagement à intégrer AI Endpoints dans un maximum d'outils open source afin d'en simplifier l'usage.

Objectif

OVHcloud AI Endpoints permet aux développeurs d'ajouter facilement des fonctionnalités d'IA à leurs développements au quotidien.

Dans ce guide, nous allons vous montrer comment utiliser Apache Airflow pour intégrer OVHcloud AI Endpoints dans vos pipelines d'orchestration de workflows.

Grâce aux puissantes fonctionnalités de gestion de workflows d'Apache Airflow et à l'infrastructure IA évolutive d'OVHcloud, vous pouvez créer, planifier et surveiller de manière programmatique des workflows pilotés par l'IA en toute simplicité.

Définition

Apache Airflow : une plateforme open source permettant de créer, planifier et surveiller des workflows de manière programmatique. Airflow permet de définir des workflows complexes sous forme de graphes acycliques dirigés (DAG) en Python, ce qui en fait un outil idéal pour l'orchestration de pipelines de données, de charges de travail IA et de tâches automatisées.
AI Endpoints : une plateforme serverless d'OVHcloud offrant un accès simplifié à une variété de modèles d'IA reconnus mondialement, dont Mistral, LLaMA, et bien d'autres. Cette plateforme est conçue pour être simple, sécurisée et intuitive, avec la confidentialité des données comme priorité absolue.

Pourquoi cette intégration est-elle importante ?

Cette nouvelle intégration vous offre plusieurs avantages :

Orchestration de workflows : planifiez et surveillez les tâches IA au sein de vos pipelines de données.
Scalabilité : tirez parti de l'architecture distribuée d'Airflow pour un traitement IA en parallèle.
Fiabilité : mécanismes de nouvelle tentative intégrés et gestion des erreurs pour les workflows en production.
Flexibilité : combinez les tâches IA avec d'autres opérations sur les données au sein de workflows unifiés.
Observabilité : surveillez l'exécution des tâches IA grâce à l'interface et aux logs riches d'Airflow.
Modèles : tous nos modèles sont disponibles via le fournisseur Airflow.

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

Un compte OVHcloud avec un accès à AI Endpoints.
Python 3.8 ou une version ultérieure installé.
Apache Airflow 2.3.0 ou une version ultérieure installé.
Une clé API générée depuis l', dans la section Public Cloud > AI Endpoints > Clés API.

En pratique

Installation

Installez le fournisseur AI Endpoints d'OVHcloud pour Apache Airflow avec pip :

pip install apache-airflow-provider-ovhcloud-ai

Vous êtes maintenant prêt à commencer.

Configuration de base

Configuration de la connexion Airflow

La méthode recommandée pour configurer votre clé API consiste à utiliser les connexions Airflow. Vous pouvez créer une connexion via l'interface Airflow ou en ligne de commande.

Via l'interface Airflow :

Rendez-vous dans Admin > Connections.
Cliquez sur le bouton + pour ajouter une nouvelle connexion.
Renseignez les informations suivantes :

Attribut	Valeur
Connection Id	`ovh_ai_endpoints_default`
Connection Type	`generic`
Password	Votre clé API AI Endpoints OVHcloud

Via la ligne de commande Airflow :

airflow connections add ovh_ai_endpoints_default \
    --conn-type generic \
    --conn-password your-api-key-here

Utilisation de base

Voici un exemple simple d'utilisation pour générer du texte avec des modèles de langage (LLM) :

from airflow import DAG
from apache_airflow_provider_ovhcloud_ai.operators.ai_endpoints import (
    OVHCloudAIEndpointsChatCompletionsOperator
)
from datetime import datetime

with DAG(
    dag_id='llm_text_generation',
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
) as dag:

    generate_text = OVHCloudAIEndpointsChatCompletionsOperator(
        task_id='generate_response',
        model='ovhcloud/gpt-oss-120b',
        messages=[
            {"role": "system", "content": "You are a helpful AI assistant."},
            {"role": "user", "content": "Explain machine learning in simple terms."}
        ],
        temperature=0.7,
        max_tokens=200,
    )

Fonctionnalités avancées

Contenu dynamique avec le templating Jinja

Utilisez le templating Jinja d'Airflow pour du contenu dynamique dans vos tâches IA :

from airflow import DAG
from apache_airflow_provider_ovhcloud_ai.operators.ai_endpoints import (
    OVHCloudAIEndpointsChatCompletionsOperator
)
from datetime import datetime

with DAG(
    dag_id='dynamic_llm_generation',
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
) as dag:

    analyze_sentiment = OVHCloudAIEndpointsChatCompletionsOperator(
        task_id='analyze_sentiment',
        model='ovhcloud/gpt-oss-120b',
        messages=[
            {
                "role": "system", 
                "content": "You are a sentiment analysis expert. Respond only with: positive, negative, or neutral."
            },
            {
                "role": "user", 
                "content": "Analyze the sentiment: {{ dag_run.conf['text'] }}"
            }
        ],
        temperature=0.3,
        max_tokens=10,
    )

Déclenchez ce DAG avec une configuration :

airflow dags trigger dynamic_llm_generation \
    --conf '{"text": "I love this product! It works great!"}'

Embeddings

Créez des embeddings vectoriels pour la recherche sémantique et la mise en correspondance par similarité :

from airflow import DAG
from apache_airflow_provider_ovhcloud_ai.operators.ai_endpoints import (
    OVHCloudAIEndpointsEmbeddingOperator
)
from datetime import datetime

with DAG(
    dag_id='create_embeddings',
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
) as dag:

    embed_text = OVHCloudAIEndpointsEmbeddingOperator(
        task_id='create_embedding',
        model='ovhcloud/BGE-M3',
        input="Apache Airflow is a platform to programmatically author, schedule and monitor workflows."
    )

Embeddings par lot

Traitez plusieurs textes en une seule opération :

from airflow import DAG
from apache_airflow_provider_ovhcloud_ai.operators.ai_endpoints import (
    OVHCloudAIEndpointsEmbeddingOperator
)
from datetime import datetime

with DAG(
    dag_id='batch_embeddings',
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
) as dag:

    embed_documents = OVHCloudAIEndpointsEmbeddingOperator(
        task_id='embed_documents',
        model='ovhcloud/BGE-M3',
        input=[
            "First document to embed",
            "Second document to embed",
            "Third document to embed"
        ]
    )

Sortie de tâche et XCom

Accédez aux sorties des opérateurs dans les tâches en aval grâce à la fonctionnalité XCom d'Airflow :

from airflow import DAG
from apache_airflow_provider_ovhcloud_ai.operators.ai_endpoints import (
    OVHCloudAIEndpointsChatCompletionsOperator
)
from airflow.operators.python import PythonOperator
from datetime import datetime

def process_llm_response(**context):
    # Pull the response from XCom
    ti = context['ti']
    llm_response = ti.xcom_pull(task_ids='generate_text')
    print(f"LLM said: {llm_response}")
    
    # Process the response
    return {"processed": True}

with DAG(
    dag_id='xcom_example',
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
) as dag:

    generate = OVHCloudAIEndpointsChatCompletionsOperator(
        task_id='generate_text',
        model='ovhcloud/gpt-oss-120b',
        messages=[
            {"role": "user", "content": "Say hello!"}
        ],
    )

    process = PythonOperator(
        task_id='process_response',
        python_callable=process_llm_response,
    )

    generate >> process

Gestion des erreurs et nouvelles tentatives

Configurez les nouvelles tentatives et la gestion des erreurs pour les workflows en production :

from airflow import DAG
from apache_airflow_provider_ovhcloud_ai.operators.ai_endpoints import (
    OVHCloudAIEndpointsChatCompletionsOperator
)
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    dag_id='production_llm_pipeline',
    default_args=default_args,
    start_date=datetime(2024, 1, 1),
    schedule='@hourly',
    catchup=False,
) as dag:

    generate_text = OVHCloudAIEndpointsChatCompletionsOperator(
        task_id='generate_text',
        model='ovhcloud/gpt-oss-120b',
        messages=[
            {"role": "user", "content": "Generate content"}
        ],
        execution_timeout=timedelta(minutes=10),
    )

Traitement en parallèle

Exécutez plusieurs tâches IA en parallèle pour maximiser le débit :

from airflow import DAG
from apache_airflow_provider_ovhcloud_ai.operators.ai_endpoints import (
    OVHCloudAIEndpointsChatCompletionsOperator,
    OVHCloudAIEndpointsEmbeddingOperator
)
from datetime import datetime

with DAG(
    dag_id='parallel_ai_tasks',
    start_date=datetime(2024, 1, 1),
    schedule=None,
    catchup=False,
) as dag:

    tasks = []
    
    # Generate multiple responses in parallel
    for i in range(5):
        task = OVHCloudAIEndpointsChatCompletionsOperator(
            task_id=f'generate_response_{i}',
            model='ovhcloud/gpt-oss-120b',
            messages=[
                {"role": "user", "content": f"Generate idea number {i}"}
            ],
        )
        tasks.append(task)
    
    # All tasks run in parallel (no dependencies)

Aller plus loin

Vous trouverez plus d'informations sur Apache Airflow dans leur documentation officielle. Vous pouvez également parcourir le catalogue AI Endpoints pour découvrir les modèles disponibles via le fournisseur Airflow.

Pour des informations détaillées sur le fournisseur, y compris des opérateurs supplémentaires et des fonctionnalités avancées, consultez la documentation du fournisseur Apache Airflow OVHcloud.

Parcourez la documentation AI Endpoints complète pour mieux comprendre les concepts principaux et démarrer.

Pour une formation ou une assistance technique sur la mise en œuvre de nos solutions, contactez votre commercial ou consultez la page Professional Services pour obtenir un devis et faire analyser votre projet par nos experts.

Votre avis nous intéresse !

N’hésitez pas à nous faire part de vos questions, retours et suggestions concernant AI Endpoints et ses fonctionnalités :

Dans le canal #ai-endpoints du serveur Discord OVHcloud, où vous pouvez échanger avec la communauté et les équipes OVHcloud.
Sur le dépôt GitHub pour signaler des bugs et contribuer.

Cette page vous a-t-elle aidé ?