Building a Real-World AI Pipeline on Azure: From Speech to GenAI Insights

Infrastructure as Code (IaC) is the only way to build production systems. I use Terraform to define and manage cloud resources declaratively. This prevents configuration drift and makes the entire environment reproducible.

Create a file named 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
}

To provision these resources, run the standard Terraform commands:

# Log into Azure first
az login

# Initialize Terraform
terraform init

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

After applying, Terraform will output the endpoints and names you'll need for your application's environment variables.

Set the following environment variables in your shell, using the outputs from Terraform and the endpoint from your 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"

Remember to grant the identity you're using (your user account or Service Principal) the 'Cognitive Services User' role on the AI Services and OpenAI resources, and appropriate permissions (e.g., 'Contributor') on the AI Project itself.

Now, let's create the Python functions that will serve as our agent's tools. I'll put these in a file named ai_pipeline_tools.py. All functions are async to ensure non-blocking I/O, which is critical for scalable 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

With our tools implemented, we can now define an agent that knows how to use them. We'll use the AIProjectClient to register the agent's definition, including its instructions and the schema of the tools it can call. Create a 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())

Running this script registers our agent in the AI Project. The agent's LLM now knows what tools it has and what they are for, ready to orchestrate them based on a prompt.

Finally, let's execute the pipeline. The following script, run_pipeline.py, simulates a user request. For this demonstration, we will manually orchestrate the tool calls to show the step-by-step logic. In a real-world application, you would interact with the deployed agent endpoint, and the agent itself would manage this orchestration internally.

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

When you run this, you'll see the full pipeline in action: the dummy audio is created, the speech service is called (and likely fails, triggering the fallback), the mock text is analyzed by the LLM, and the resulting summary is translated into French.