Construyendo un pipeline de IA en el mundo real en Azure: de la voz a insights de GenAI

La infraestructura como código (IaC) es la única forma de construir sistemas de producción. Utilizo Terraform para definir y gestionar los recursos de la nube de forma declarativa. Esto evita la desviación de la configuración y hace que todo el entorno sea reproducible.

Cree un archivo llamado 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
}

Para aprovisionar estos recursos, ejecute los comandos estándar de Terraform:

# Log into Azure first
az login

# Initialize Terraform
terraform init

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

Después de la aplicación, Terraform generará los endpoints y nombres que necesitará para las variables de entorno de su aplicación.

Establezca las siguientes variables de entorno en su shell, utilizando las salidas de Terraform y el endpoint de su Proyecto de IA en 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"

Recuerde conceder a la identidad que esté utilizando (su cuenta de usuario o Entidad de Servicio) el rol de 'Usuario de Servicios Cognitivos' en los recursos de AI Services y OpenAI, y los permisos adecuados (por ejemplo, 'Colaborador') en el propio Proyecto de IA.

Ahora, creemos las funciones de Python que servirán como herramientas de nuestro agente. Las pondré en un archivo llamado ai_pipeline_tools.py. Todas las funciones son async para asegurar E/S no bloqueante, lo cual es crítico para servicios escalables.

# 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

# --- Tool Implementations ---

async def speech_to_text(audio_file_path: str, credential: DefaultAzureCredential) -> str:
    """Transcribes an audio file to text using Azure Speech Service."""
    speech_region = os.environ["AZURE_AI_SERVICES_REGION"]

    # The Speech SDK needs an auth token. We'll get one using our async credential.
    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"Performing speech-to-text on {audio_file_path}...")
    result = await speech_recognizer.recognize_once_async()

    if result.reason == speechsdk.ResultReason.RecognizedSpeech:
        print(f"Recognized: {result.text}")
        return result.text
    elif result.reason == speechsdk.ResultReason.NoMatch:
        print("No speech could be recognized.")
    elif result.reason == speechsdk.ResultReason.Canceled:
        cancellation = result.cancellation_details
        print(f"Speech Recognition canceled: {cancellation.reason}")
        if cancellation.reason == speechsdk.CancellationReason.Error:
            print(f"Error details: {cancellation.error_details}")
        raise RuntimeError(f"Speech recognition failed: {cancellation.error_details}")
    return ""

async def genai_sentiment_analysis(text: str, credential: DefaultAzureCredential) -> Dict[str, Any]:
    """Analyzes text for sentiment and extracts key takeaways using Azure OpenAI."""
    # This helper function provides a refreshed token to the OpenAI client.
    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"""Analyze the following customer call transcript. Identify the overall sentiment, any specific issues raised, and summarize the key takeaways. Your response MUST be a valid JSON object with three keys: 'overall_sentiment' (string: 'positive', 'neutral', or 'negative'), 'issues' (list of strings), and 'key_takeaways' (list of strings).

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

JSON Output:"""

    print(f"Performing GenAI analysis on: {text[:60]}...")
    response = await client.chat.completions.create(
        model=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
        response_format={ "type": "json_object" },
        messages=[
            {"role": "system", "content": "You are an expert AI assistant for customer service analysis."},
            {"role": "user", "content": prompt}
        ]
    )
    analysis_result = json.loads(response.choices[0].message.content)
    print(f"GenAI Analysis: {analysis_result}")
    await client.close()
    return analysis_result

async def translate_text(text: str, credential: DefaultAzureCredential, target_language: str = "fr") -> str:
    """Translates text to a target language using Azure Translator Service."""
    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"Translating text to '{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"Translated: {translated_text}")
    return translated_text

Con nuestras herramientas implementadas, ahora podemos definir un agente que sepa cómo usarlas. Utilizaremos AIProjectClient para registrar la definición del agente, incluyendo sus instrucciones y el esquema de las herramientas que puede invocar. Cree un archivo 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"

        # Define the tools the agent can use based on their function signatures
        speech_tool = FunctionTool(
            name="speech_to_text_transcription",
            description="Transcribes an audio file path into text.",
            parameters={
                "type": "object",
                "properties": {"audio_file_path": {"type": "string"}},
                "required": ["audio_file_path"]
            }
        )

        sentiment_tool = FunctionTool(
            name="genai_sentiment_analysis_tool",
            description="Analyzes text for sentiment, issues, and key takeaways.",
            parameters={
                "type": "object",
                "properties": {"text": {"type": "string"}},
                "required": ["text"]
            }
        )

        translation_tool = FunctionTool(
            name="translate_text_tool",
            description="Translates text to a specified target language.",
            parameters={
                "type": "object",
                "properties": {
                    "text": {"type": "string"},
                    "target_language": {"type": "string", "default": "fr"}
                },
                "required": ["text"]
            }
        )

        # Define the agent's instructions and link the tools
        agent_definition = PromptAgentDefinition(
            model=f"azure_openai:/{openai_deployment_name}",
            instructions=(
                "You are a call analysis assistant. Your job is to take an audio file path, "
                "transcribe it, analyze the transcript for sentiment and issues, and finally "
                "translate the analysis summary into a target language (defaulting to French)."
            ),
            tools=[speech_tool, sentiment_tool, translation_tool]
        )

        print(f"Creating or updating agent '{agent_name}'...")
        agent_version = await client.agents.create_version(
            agent_name=agent_name,
            definition=agent_definition
        )

        print(f"Agent deployment complete. Name: {agent_version.name}, Version: {agent_version.version}")

    await credential.close()

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

Ejecutar este script registra nuestro agente en el Proyecto de IA. El LLM del agente ahora sabe qué herramientas tiene y para qué sirven, listo para orquestarlas basándose en un prompt.

Finalmente, ejecutemos el pipeline. El siguiente script, run_pipeline.py, simula una solicitud de usuario. Para esta demostración, orquestaremos manualmente las llamadas a las herramientas para mostrar la lógica paso a paso. En una aplicación real, interactuaría con el endpoint del agente desplegado, y el propio agente gestionaría esta orquestación internamente.

# 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"):
    """Creates a dummy WAV file. The content is just a sine wave."""
    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"Dummy audio file '{filename}' created for STT input.")
    return filename

async def main():
    credential = DefaultAzureCredential()

    # --- Simulation Setup ---
    # For this demo, we'll use a clear, known text as a fallback,
    # since transcribing a dummy sine wave won't produce meaningful results.
    audio_file = create_dummy_audio_file()
    mock_transcript = "The customer reported a critical issue with the service stability. Performance has degraded significantly over the past hour."

    print("\n--- Simulating Agent Pipeline Execution ---")

    # Step 1: Speech-to-Text
    print("\n[AGENT] Calling Speech-to-Text tool...")
    transcribed_text = await speech_to_text(audio_file, credential)
    if not transcribed_text:
        print("STT returned no result on dummy audio, using mock transcript.")
        transcribed_text = mock_transcript

    # Step 2: GenAI Sentiment Analysis
    print("\n[AGENT] Calling GenAI Analysis tool...")
    analysis = await genai_sentiment_analysis(transcribed_text, credential)
    summary_for_translation = f"Sentiment: {analysis.get('overall_sentiment')}. Issues: {analysis.get('issues')}. Takeaways: {analysis.get('key_takeaways')}."

    # Step 3: Translation
    print("\n[AGENT] Calling Translation tool...")
    translated_summary = await translate_text(summary_for_translation, credential, target_language="fr")

    print("\n--- Final Pipeline Output ---")
    print(f"Original Transcript: {transcribed_text}")
    print(f"Analysis (JSON): {json.dumps(analysis, indent=2)}")
    print(f"Translated Summary (fr): {translated_summary}")

    await credential.close()

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

Cuando ejecute esto, verá el pipeline completo en acción: se crea el audio ficticio, se llama al servicio de voz (y es probable que falle, activando la alternativa), el texto simulado es analizado por el LLM y el resumen resultante se traduce al francés.