Costruire una pipeline AI reale su Azure: dal parlato alle analisi GenAI

L'Infrastructure as Code (IaC) è l'unico modo per costruire sistemi di produzione. Utilizzo Terraform per definire e gestire le risorse cloud in modo dichiarativo. Questo previene il 'configuration drift' e rende l'intero ambiente riproducibile.

Create un file chiamato 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
}

Per effettuare il provisioning di queste risorse, eseguite i comandi Terraform standard:

# Log into Azure first
az login

# Initialize Terraform
terraform init

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

Dopo l'applicazione, Terraform genererà gli endpoint e i nomi di cui avrete bisogno per le variabili d'ambiente della vostra applicazione.

Impostate le seguenti variabili d'ambiente nella vostra shell, utilizzando gli output di Terraform e l'endpoint del vostro AI Project in 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"

Ricordate di concedere all'identità che state utilizzando (il vostro account utente o Service Principal) il ruolo di 'Utente di Servizi Cognitivi' sulle risorse AI Services e OpenAI, e le autorizzazioni appropriate (es. 'Collaboratore') sull'AI Project stesso.

Ora, creiamo le funzioni Python che serviranno come strumenti del nostro agente. Le collocherò in un file chiamato ai_pipeline_tools.py. Tutte le funzioni sono async per garantire I/O non bloccante, il che è fondamentale per servizi scalabili.

# 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 i nostri strumenti implementati, possiamo ora definire un agente che sa come usarli. Utilizzeremo l'AIProjectClient per registrare la definizione dell'agente, comprese le sue istruzioni e lo schema degli strumenti che può richiamare. Create un file 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())

L'esecuzione di questo script registra il nostro agente nell'AI Project. L'LLM dell'agente ora sa quali strumenti ha e a cosa servono, pronto a orchestrarli in base a un prompt.

Infine, eseguiamo la pipeline. Il seguente script, run_pipeline.py, simula una richiesta utente. Per questa dimostrazione, orchesteremo manualmente le chiamate agli strumenti per mostrare la logica passo-passo. In un'applicazione reale, interagireste con l'endpoint dell'agente distribuito e l'agente stesso gestirebbe internamente questa orchestrazione.

# 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())

Quando lo eseguirete, vedrete l'intera pipeline in azione: l'audio fittizio viene creato, il servizio vocale viene chiamato (e probabilmente fallisce, attivando il fallback), il testo fittizio viene analizzato dall'LLM e il riassunto risultante viene tradotto in francese.