Construire un pipeline d'IA en conditions réelles sur Azure : de la parole aux analyses GenAI

L'infrastructure en tant que code (IaC) est le seul moyen de construire des systèmes de production. J'utilise Terraform pour définir et gérer les ressources cloud de manière déclarative. Cela évite la dérive de configuration et rend l'environnement entier reproductible.

Créez un fichier nommé main.tf :

# main.tf - Provisioning AI Services for our pipeline
terraform {
  required_providers {
    azurerm = {
      source  = "hashicorp/azurerm"
      version = "~> 3.0"
    }
  }
}

provider "azurerm" {
  features {}
}

# Variables for naming and location
variable "resource_group_name" {
  description = "Name of the resource group."
  type        = string
  default     = "rg-ai-pipeline-dev-euw"
}

variable "location" {
  description = "Azure region for deployment."
  type        = string
  default     = "westeurope"
}

variable "base_name" {
  description = "A unique base name for resources to avoid collisions."
  type        = string
  default     = "tcapipelineeuw"
}

# Create a resource group in our target region
resource "azurerm_resource_group" "main" {
  name     = var.resource_group_name
  location = var.location
}

# Create an Azure AI Services account (for Speech and Translator)
resource "azurerm_cognitive_account" "ai_services" {
  name                = "ais-${var.base_name}"
  location            = azurerm_resource_group.main.location
  resource_group_name = azurerm_resource_group.main.name
  kind                = "CognitiveServices"
  sku_name            = "S0" # Standard tier
}

# Create an Azure OpenAI Service account
resource "azurerm_cognitive_account" "openai" {
  name                = "aoai-${var.base_name}"
  location            = azurerm_resource_group.main.location
  resource_group_name = azurerm_resource_group.main.name
  kind                = "OpenAI"
  sku_name            = "S0"
}

# Deploy a GPT-4o model to the Azure OpenAI account
resource "azurerm_cognitive_deployment" "openai_deployment" {
  name                 = "gpt-4o"
  cognitive_account_id = azurerm_cognitive_account.openai.id
  model {
    format  = "OpenAI"
    name    = "gpt-4o"
    version = "2024-05-13"
  }
  scale {
    type     = "Standard"
    capacity = 10 # Throughput units (10 = 10k tokens/min)
  }
  # Note: Ensure your subscription has quota for this model in the target region.
}

# Outputs for our Python application
output "ai_services_endpoint" {
  value = azurerm_cognitive_account.ai_services.endpoint
}

output "ai_services_region" {
  value = azurerm_cognitive_account.ai_services.location
}

output "openai_endpoint" {
  value = azurerm_cognitive_account.openai.endpoint
}

output "openai_deployment_name" {
  value = azurerm_cognitive_deployment.openai_deployment.name
}

Pour provisionner ces ressources, exécutez les commandes Terraform standard :

# Log into Azure first
az login

# Initialize Terraform
terraform init

# Plan and apply the changes
terraform plan
terraform apply --auto-approve

Après application, Terraform affichera les points de terminaison et les noms dont vous aurez besoin pour les variables d'environnement de votre application.

Définissez les variables d'environnement suivantes dans votre shell, en utilisant les sorties de Terraform et le point de terminaison de votre projet d'IA dans Azure AI Studio.

# Get this from the 'Develop' section of your AI Project in Azure AI Studio
export AZURE_AI_PROJECT_ENDPOINT="https://<your-hub-name>.<region>.inference.ai.azure.com/api/projects/<your-project-name>"

# Use these if you're authenticating with a Service Principal
export AZURE_CLIENT_ID="<your-service-principal-client-id>"
export AZURE_CLIENT_SECRET="<your-service-principal-client-secret>"
export AZURE_TENANT_ID="<your-tenant-id>"

# From Terraform outputs
export AZURE_AI_SERVICES_ENDPOINT="https://ais-tcapipelineeuw.cognitiveservices.azure.com/"
export AZURE_AI_SERVICES_REGION="westeurope"
export AZURE_OPENAI_ENDPOINT="https://aoai-tcapipelineeuw.openai.azure.com/"
export AZURE_OPENAI_DEPLOYMENT_NAME="gpt-4o"

N'oubliez pas d'accorder à l'identité que vous utilisez (votre compte d'utilisateur ou votre principal de service) le rôle 'Utilisateur des services cognitifs' sur les ressources AI Services et OpenAI, ainsi que les autorisations appropriées (par exemple, 'Contributeur') sur le projet d'IA lui-même.

Maintenant, créons les fonctions Python qui serviront d'outils à notre agent. Je les placerai dans un fichier nommé ai_pipeline_tools.py. Toutes les fonctions sont async pour garantir des E/S non bloquantes, ce qui est essentiel pour des services évolutifs.

# ai_pipeline_tools.py

import os
import json
import httpx
from typing import Dict, Any

import azure.cognitiveservices.speech as speechsdk
from openai import AsyncAzureOpenAI
from azure.identity.aio import DefaultAzureCredential

# --- Implémentations des outils ---

async def speech_to_text(audio_file_path: str, credential: DefaultAzureCredential) -> str:
    """Transcrire un fichier audio en texte à l'aide du service Azure Speech."""
    speech_region = os.environ["AZURE_AI_SERVICES_REGION"]

    # Le SDK Speech a besoin d'un jeton d'authentification. Nous en obtiendrons un en utilisant notre credential asynchrone.
    token = await credential.get_token("https://cognitiveservices.azure.com/.default")
    speech_config = speechsdk.SpeechConfig(auth_token=token.token, region=speech_region)

    audio_config = speechsdk.AudioConfig(filename=audio_file_path)
    speech_recognizer = speechsdk.SpeechRecognizer(speech_config=speech_config, audio_config=audio_config)

    print(f"Effectuer la reconnaissance vocale sur {audio_file_path}...")
    result = await speech_recognizer.recognize_once_async()

    if result.reason == speechsdk.ResultReason.RecognizedSpeech:
        print(f"Reconnu : {result.text}")
        return result.text
    elif result.reason == speechsdk.ResultReason.NoMatch:
        print("Aucune parole n'a pu être reconnue.")
    elif result.reason == speechsdk.ResultReason.Canceled:
        cancellation = result.cancellation_details
        print(f"Reconnaissance vocale annulée : {cancellation.reason}")
        if cancellation.reason == speechsdk.CancellationReason.Error:
            print(f"Détails de l'erreur : {cancellation.error_details}")
        raise RuntimeError(f"Échec de la reconnaissance vocale : {cancellation.error_details}")
    return ""

async def genai_sentiment_analysis(text: str, credential: DefaultAzureCredential) -> Dict[str, Any]:
    """Analyse le texte pour le sentiment et extrait les points clés à l'aide d'Azure OpenAI."""
    # Cette fonction d'aide fournit un jeton rafraîchi au client OpenAI.
    async def token_provider():
        token = await credential.get_token("https://cognitiveservices.azure.com/.default")
        return token.token

    client = AsyncAzureOpenAI(
        azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
        api_version="2024-05-01-preview",
        azure_ad_token_provider=token_provider,
    )

    prompt = f"""Analysez la transcription d'appel client suivante. Identifiez le sentiment général, les problèmes spécifiques soulevés et résumez les points clés. Votre réponse DOIT être un objet JSON valide avec trois clés : 'overall_sentiment' (chaîne : 'positive', 'neutral' ou 'negative'), 'issues' (liste de chaînes) et 'key_takeaways' (liste de chaînes).

Transcript: \"\"\"{text}\"\"\"

JSON Output:"""

    print(f"Effectuer une analyse GenAI sur : {text[:60]}...")
    response = await client.chat.completions.create(
        model=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
        response_format={ "type": "json_object" },
        messages=[
            {"role": "system", "content": "Vous êtes un assistant IA expert pour l'analyse du service client."},
            {"role": "user", "content": prompt}
        ]
    )
    analysis_result = json.loads(response.choices[0].message.content)
    print(f"Analyse GenAI : {analysis_result}")
    await client.close()
    return analysis_result

async def translate_text(text: str, credential: DefaultAzureCredential, target_language: str = "fr") -> str:
    """Traduit le texte vers une langue cible à l'aide du service Azure Translator."""
    translator_endpoint = os.environ["AZURE_AI_SERVICES_ENDPOINT"]
    token = await credential.get_token('https://cognitiveservices.azure.com/.default')

    headers = {
        'Authorization': f'Bearer {token.token}',
        'Content-Type': 'application/json',
    }

    api_endpoint = f"{translator_endpoint.rstrip('/')}/translator/text/v3.0/translate"
    params = {'api-version': '3.0', 'to': target_language}
    body = [{'text': text}]

    print(f"Traduction du texte vers '{target_language}' : {text[:60]}...")
    async with httpx.AsyncClient() as client:
        response = await client.post(api_endpoint, params=params, headers=headers, json=body)
        response.raise_for_status()

    translation = response.json()
    translated_text = translation[0]['translations'][0]['text']
    print(f"Traduit : {translated_text}")
    return translated_text

Nos outils étant implémentés, nous pouvons maintenant définir un agent qui sait les utiliser. Nous utiliserons l'AIProjectClient pour enregistrer la définition de l'agent, y compris ses instructions et le schéma des outils qu'il peut appeler. Créez un fichier deploy_agent.py :

# deploy_agent.py

import asyncio
import os
from azure.ai.projects.aio import AIProjectClient
from azure.ai.projects.models import PromptAgentDefinition, FunctionTool
from azure.identity.aio import DefaultAzureCredential

async def main():
    project_endpoint = os.environ["AZURE_AI_PROJECT_ENDPOINT"]
    openai_deployment_name = os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"]
    credential = DefaultAzureCredential()

    async with AIProjectClient(endpoint=project_endpoint, credential=credential) as client:
        agent_name = "call-analysis-pipeline-agent"

        # Définir les outils que l'agent peut utiliser en fonction de leurs signatures de fonction
        speech_tool = FunctionTool(
            name="speech_to_text_transcription",
            description="Transcrire un chemin de fichier audio en texte.",
            parameters={
                "type": "object",
                "properties": {"audio_file_path": {"type": "string"}},
                "required": ["audio_file_path"]
            }
        )

        sentiment_tool = FunctionTool(
            name="genai_sentiment_analysis_tool",
            description="Analyse le texte pour le sentiment, les problèmes et les points clés.",
            parameters={
                "type": "object",
                "properties": {"text": {"type": "string"}},
                "required": ["text"]
            }
        )

        translation_tool = FunctionTool(
            name="translate_text_tool",
            description="Traduit le texte vers une langue cible spécifiée.",
            parameters={
                "type": "object",
                "properties": {
                    "text": {"type": "string"},
                    "target_language": {"type": "string", "default": "fr"}
                },
                "required": ["text"]
            }
        )

        # Définir les instructions de l'agent et lier les outils
        agent_definition = PromptAgentDefinition(
            model=f"azure_openai:/{openai_deployment_name}",
            instructions=(
                "Vous êtes un assistant d'analyse d'appels. Votre travail consiste à prendre un chemin de fichier audio, "
                "à le transcrire, à analyser la transcription pour le sentiment et les problèmes, et enfin "
                "à traduire le résumé de l'analyse dans une langue cible (par défaut le français)."
            ),
            tools=[speech_tool, sentiment_tool, translation_tool]
        )

        print(f"Création ou mise à jour de l'agent '{agent_name}'...")
        agent_version = await client.agents.create_version(
            agent_name=agent_name,
            definition=agent_definition
        )

        print(f"Déploiement de l'agent terminé. Nom : {agent_version.name}, Version : {agent_version.version}")

    await credential.close()

if __name__ == "__main__":
    asyncio.run(main())

L'exécution de ce script enregistre notre agent dans le projet d'IA. Le LLM de l'agent sait désormais quels outils il possède et à quoi ils servent, prêt à les orchestrer en fonction d'une invite.

Enfin, exécutons le pipeline. Le script suivant, run_pipeline.py, simule une requête utilisateur. Pour cette démonstration, nous orchestrerons manuellement les appels aux outils pour montrer la logique étape par étape. Dans une application réelle, vous interagiriez avec le point de terminaison de l'agent déployé, et l'agent lui-même gérerait cette orchestration en interne.

# run_pipeline.py

import asyncio
import os
import json
import numpy as np
import scipy.io.wavfile as wavfile
from azure.identity.aio import DefaultAzureCredential

# Import our tool implementations
from ai_pipeline_tools import speech_to_text, genai_sentiment_analysis, translate_text

def create_dummy_audio_file(filename="sample_call.wav"):
    """Crée un fichier WAV factice. Le contenu est juste une onde sinusoïdale."""
    samplerate = 16000
    duration = 3.0
    frequency = 440.0
    t = np.linspace(0., duration, int(samplerate * duration))
    amplitude = np.iinfo(np.int16).max * 0.3
    data = amplitude * np.sin(2. * np.pi * frequency * t)
    wavfile.write(filename, samplerate, data.astype(np.int16))
    print(f"Fichier audio factice '{filename}' créé pour l'entrée STT.")
    return filename

async def main():
    credential = DefaultAzureCredential()

    # --- Configuration de la simulation ---
    # Pour cette démonstration, nous utiliserons un texte clair et connu comme solution de repli,
    # car la transcription d'une onde sinusoïdale factice ne produira pas de résultats significatifs.
    audio_file = create_dummy_audio_file()
    mock_transcript = "Le client a signalé un problème critique avec la stabilité du service. Les performances se sont dégradées de manière significative au cours de la dernière heure."

    print("\n--- Simulation de l'exécution du pipeline de l'agent ---")

    # Étape 1 : Reconnaissance vocale (Speech-to-Text)
    print("\n[AGENT] Appel de l'outil de reconnaissance vocale...")
    transcribed_text = await speech_to_text(audio_file, credential)
    if not transcribed_text:
        print("Le STT n'a renvoyé aucun résultat sur l'audio factice, utilisation de la transcription simulée.")
        transcribed_text = mock_transcript

    # Étape 2 : Analyse des sentiments GenAI
    print("\n[AGENT] Appel de l'outil d'analyse GenAI...")
    analysis = await genai_sentiment_analysis(transcribed_text, credential)
    summary_for_translation = f"Sentiment: {analysis.get('overall_sentiment')}. Problèmes: {analysis.get('issues')}. Points clés: {analysis.get('key_takeaways')}."

    # Étape 3 : Traduction
    print("\n[AGENT] Appel de l'outil de traduction...")
    translated_summary = await translate_text(summary_for_translation, credential, target_language="fr")

    print("\n--- Sortie finale du pipeline ---")
    print(f"Transcription originale : {transcribed_text}")
    print(f"Analyse (JSON) : {json.dumps(analysis, indent=2)}")
    print(f"Résumé traduit (fr) : {translated_summary}")

    await credential.close()

if __name__ == "__main__":
    asyncio.run(main())

Lorsque vous l'exécutez, vous verrez le pipeline complet en action : l'audio factice est créé, le service vocal est appelé (et échoue probablement, déclenchant le repli), le texte simulé est analysé par le LLM, et le résumé résultant est traduit en français.