Een realistische AI-pijplijn bouwen op Azure: van spraak naar GenAI-inzichten

Infrastructure as Code (IaC) is de enige manier om productiesystemen te bouwen. Ik gebruik Terraform om cloudresources declaratief te definiëren en te beheren. Dit voorkomt configuratieafwijkingen en maakt de hele omgeving reproduceerbaar.

Maak een bestand met de naam 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
}

Om deze resources in te richten, voert u de standaard Terraform-commando's uit:

# Log into Azure first
az login

# Initialize Terraform
terraform init

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

Na de toepassing zal Terraform de eindpunten en namen uitvoeren die je nodig hebt voor de omgevingsvariabelen van je applicatie.

Stel de volgende omgevingsvariabelen in je shell in, gebruikmakend van de outputs van Terraform en het eindpunt van je 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"

Vergeet niet de identiteit die je gebruikt (je gebruikersaccount of Service Principal) de rol 'Cognitive Services User' toe te kennen op de AI Services- en OpenAI-resources, en geschikte machtigingen (bijv. 'Contributor') op het AI Project zelf.

Nu creëren we de Python-functies die als tools voor onze agent zullen dienen. Ik plaats deze in een bestand met de naam ai_pipeline_tools.py. Alle functies zijn async om niet-blokkerende I/O te garanderen, wat cruciaal is voor schaalbare services.

# 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

Met onze geïmplementeerde tools kunnen we nu een agent definiëren die weet hoe ze te gebruiken. We zullen de AIProjectClient gebruiken om de definitie van de agent te registreren, inclusief zijn instructies en het schema van de tools die het kan aanroepen. Maak een bestand 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())

Door dit script uit te voeren, wordt onze agent geregistreerd in het AI Project. De LLM van de agent weet nu welke tools het heeft en waarvoor ze dienen, klaar om ze te orkestreren op basis van een prompt.

Laten we tot slot de pijplijn uitvoeren. Het volgende script, run_pipeline.py, simuleert een gebruikersverzoek. Voor deze demonstratie zullen we de tool-oproepen handmatig orkestreren om de stapsgewijze logica te tonen. In een real-world applicatie zou je interacteren met het geïmplementeerde agent-eindpunt, en de agent zelf zou deze orkestratie intern beheren.

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

Wanneer je dit uitvoert, zie je de volledige pijplijn in actie: de dummy-audio wordt gemaakt, de speech-service wordt aangeroepen (en faalt waarschijnlijk, wat de fallback activeert), de mock-tekst wordt geanalyseerd door de LLM, en de resulterende samenvatting wordt vertaald naar het Frans.