Aufbau einer realen KI-Pipeline in Azure: Von Sprache zu GenAI-Erkenntnissen

Infrastructure as Code (IaC) ist der einzige Weg, Produktionssysteme zu erstellen. Ich verwende Terraform, um Cloud-Ressourcen deklarativ zu definieren und zu verwalten. Dies verhindert Konfigurationsdrifts und macht die gesamte Umgebung reproduzierbar.

Erstellen Sie eine Datei namens 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
}

Um diese Ressourcen bereitzustellen, führen Sie die Standard-Terraform-Befehle aus:

# Log into Azure first
az login

# Initialize Terraform
terraform init

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

Nach der Anwendung gibt Terraform die Endpunkte und Namen aus, die Sie für die Umgebungsvariablen Ihrer Anwendung benötigen.

Legen Sie die folgenden Umgebungsvariablen in Ihrer Shell fest, indem Sie die Ausgaben von Terraform und den Endpunkt Ihres KI-Projekts im Azure AI Studio verwenden.

# 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"

Denken Sie daran, der von Ihnen verwendeten Identität (Ihrem Benutzerkonto oder Dienstprinzipal) die Rolle 'Cognitive Services User' für die AI Services- und OpenAI-Ressourcen sowie entsprechende Berechtigungen (z. B. 'Contributor') für das AI Project selbst zu erteilen.

Nun erstellen wir die Python-Funktionen, die als Tools unseres Agenten dienen werden. Ich werde diese in einer Datei namens ai_pipeline_tools.py ablegen. Alle Funktionen sind async, um nicht-blockierende E/A zu gewährleisten, was für skalierbare Dienste entscheidend ist.

# 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

Nachdem unsere Tools implementiert sind, können wir nun einen Agenten definieren, der weiß, wie man sie verwendet. Wir werden den AIProjectClient verwenden, um die Definition des Agenten zu registrieren, einschließlich seiner Anweisungen und des Schemas der Tools, die er aufrufen kann. Erstellen Sie eine Datei 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())

Das Ausführen dieses Skripts registriert unseren Agenten im KI-Projekt. Das LLM des Agenten weiß nun, welche Tools es hat und wofür sie sind, bereit, sie basierend auf einer Eingabeaufforderung zu orchestrieren.

Schließlich führen wir die Pipeline aus. Das folgende Skript, run_pipeline.py, simuliert eine Benutzeranfrage. Für diese Demonstration werden wir die Tool-Aufrufe manuell orchestrieren, um die Schritt-für-Schritt-Logik zu zeigen. In einer realen Anwendung würden Sie mit dem bereitgestellten Agenten-Endpunkt interagieren, und der Agent selbst würde diese Orchestrierung intern verwalten.

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

Wenn Sie dies ausführen, sehen Sie die gesamte Pipeline in Aktion: Die Dummy-Audiodatei wird erstellt, der Spracherkennungsdienst wird aufgerufen (und schlägt wahrscheinlich fehl, wodurch der Fallback ausgelöst wird), der Mock-Text wird vom LLM analysiert und die resultierende Zusammenfassung wird ins Französische übersetzt.