ktg-plugin-marketplace/plugins/ms-ai-architect/skills/ms-ai-engineering/references/multi-modal/speech-to-ai-pipelines.md

Speech-to-AI Integration Pipelines

Last updated: 2026-02 Status: GA Category: Multi-Modal AI

---

Introduksjon

Speech-to-AI-integrasjon handlar om å kople talegjenkjenning med downstream AI-tenester for å skape ende-til-ende-pipelines som konverterer tale til handlingsorienterte innsikter. Azure Speech Service dannar grunnlaget, med støtte for sanntids talegjenkjenning, batchtranskribering, språkdeteksjon, talardiarisering og tilpassa talemodular.

For norsk offentleg sektor er tale-til-AI-pipelines avgjerande for tilgjengelegheit (automatisk teksting), møtetranskribering (kommunestyre, utval), klageblandling per telefon, og innbyggardialog via talebaserte brukargrensesnitt. Azure Speech Service støttar norsk bokmål (nb-NO) og kan kombinerast med Azure OpenAI for intelligente samtaleagenter.

Denne referansefila dekkjer arkitekturmønster for sanntidsstreaming, batchprosessering, språkdeteksjon og feilhandtering — med fokus på robuste produksjonsklare implementeringar.

---

Speech Recognition og Language Detection

Azure Speech Service oversikt

Funksjon	Tilgjengelegheit	Brukstilfelle
Real-time STT	GA	Samtaleagenter, live teksting
Batch transcription	GA	Arkivprosessering, møtereferat
Fast transcription	GA	Rask transkribering av filer
Language identification	GA	Fleirspråklege samtalar
Speaker diarization	GA	Møtetranskribering
Custom speech	GA	Bransjespecifikk terminologi
Real-time TTS	GA	Talebaserte assistentar
Text streaming	GA	Sanntids TTS frå GPT-output
gpt-4o-realtime	Preview	Direkte tale-til-tale

Språkgjenkjenning

Azure Speech Service tilbyr to typar språkdeteksjon:

At-Start Language Identification:

• Detekterer språk i starten av audiostraumen
• Støtta i C#, C++, Python, Java, JavaScript, Objective-C
• Best for scenario med eitt hovudspråk per sesjon

Continuous Language Identification:

• Detekterer språkbytter undervegs i samtalen
• Støtta i C#, C++, Java, JavaScript, Python
• Kritisk for fleirspråklege norske kommunar

import azure.cognitiveservices.speech as speechsdk

speech_config = speechsdk.SpeechConfig(
    subscription=os.environ["SPEECH_KEY"],
    region=os.environ["SPEECH_REGION"]
)

# Konfigurer automatisk språkdeteksjon
auto_detect_config = speechsdk.languageconfig.AutoDetectSourceLanguageConfig(
    languages=["nb-NO", "nn-NO", "en-US", "se-NO"]  # Bokmål, nynorsk, engelsk, nordsamisk
)

speech_recognizer = speechsdk.SpeechRecognizer(
    speech_config=speech_config,
    auto_detect_source_language_config=auto_detect_config,
    audio_config=audio_config
)

def recognized_handler(evt):
    result = evt.result
    auto_detect_result = speechsdk.AutoDetectSourceLanguageResult(result)
    detected_language = auto_detect_result.language
    print(f"[{detected_language}] {result.text}")

speech_recognizer.recognized.connect(recognized_handler)
speech_recognizer.start_continuous_recognition()

Tilpassa talemodular (Custom Speech)

For norsk offentleg sektor med spesialisert terminologi:

Tilpasning	Brukstilfelle	Eksempel
Phrase list	Forbetra gjenkjenning av spesifikke ord	Stadnamn, fagtermar
Custom model	Trenar ny modell med eigne data	Vegsektoren, helsesektor
Display format	Tilpass visning av gjenkjend tekst	Tal-til-siffer, dato-format

# Phrase list for forbetra norsk gjenkjenning
phrase_list = speechsdk.PhraseListGrammar.from_recognizer(speech_recognizer)
phrase_list.addPhrase("Statens vegvesen")
phrase_list.addPhrase("E6 Megården-Mørsvikbotn")
phrase_list.addPhrase("Utredningsinstruksen")
phrase_list.addPhrase("Forvaltningsloven")
phrase_list.addPhrase("personvernforordningen")

---

Audio Preprocessing og Quality Assessment

Audioformat og kvalitetskrav

Parameter	Tilrådde verdi	Minimum	Kommentar
Samplingsrate	16 kHz	8 kHz	16 kHz gir best nøyaktigheit
Bit depth	16-bit	8-bit	Mono PCM tilrådde
Kanalar	Mono	Mono	Stereo splittar automatisk
Format	WAV (PCM)	WAV, MP3, OGG	WAV gir minst komprimeringstap
SNR	>20 dB	>10 dB	Signal-to-noise ratio

Audio preprocessing pipeline

import numpy as np
from scipy.io import wavfile
from scipy.signal import butter, lfilter

class AudioPreprocessor:
    """Preprosessering av audio for optimal talegjenkjenning."""

    def __init__(self, target_sample_rate=16000):
        self.target_sample_rate = target_sample_rate

    def assess_quality(self, audio_data: np.ndarray, sample_rate: int) -> dict:
        """Vurder audiokvalitet før talegjenkjenning."""
        # Berekn SNR
        signal_power = np.mean(audio_data ** 2)
        noise_floor = np.percentile(np.abs(audio_data), 5) ** 2
        snr = 10 * np.log10(signal_power / max(noise_floor, 1e-10))

        # Berekn varigheit
        duration = len(audio_data) / sample_rate

        # Detekter stille
        silence_threshold = np.percentile(np.abs(audio_data), 10)
        silence_ratio = np.sum(np.abs(audio_data) < silence_threshold) / len(audio_data)

        # Berekn peak level
        peak_level = 20 * np.log10(np.max(np.abs(audio_data)) / 32768)

        return {
            "snr_db": round(snr, 1),
            "duration_seconds": round(duration, 1),
            "silence_ratio": round(silence_ratio, 3),
            "peak_level_db": round(peak_level, 1),
            "sample_rate": sample_rate,
            "quality_score": self._calculate_quality_score(snr, silence_ratio, peak_level),
            "recommendation": self._get_recommendation(snr, silence_ratio)
        }

    def _calculate_quality_score(self, snr, silence_ratio, peak_level):
        score = 0
        if snr > 20: score += 40
        elif snr > 15: score += 30
        elif snr > 10: score += 20
        else: score += 10

        if silence_ratio < 0.3: score += 30
        elif silence_ratio < 0.5: score += 20
        else: score += 10

        if -6 < peak_level < -1: score += 30
        elif -12 < peak_level < 0: score += 20
        else: score += 10

        return min(score, 100)

    def _get_recommendation(self, snr, silence_ratio):
        if snr < 10:
            return "Låg SNR — vurder støyreduksjon eller betre mikrofon"
        if silence_ratio > 0.7:
            return "Mykje stille — sjekk at audio faktisk inneheld tale"
        return "Kvaliteten er akseptabel for talegjenkjenning"

    def apply_noise_reduction(self, audio_data: np.ndarray, sample_rate: int) -> np.ndarray:
        """Enkel bandpassfiltrering for å redusere støy."""
        low_freq = 80   # Hz — under menneskeleg tale
        high_freq = 8000  # Hz — over dei fleste talefrekvensane

        nyquist = sample_rate / 2
        low = low_freq / nyquist
        high = high_freq / nyquist

        b, a = butter(4, [low, high], btype='band')
        return lfilter(b, a, audio_data).astype(np.int16)

Quality gates for produksjon

Gate	Terskel	Handling
SNR-sjekk	< 10 dB	Avvis, be om ny innspeling
Varighetssjekk	< 1 sekund	Avvis, for kort
Stillesjekk	> 80% stille	Åtvar, be om verifikasjon
Peak clipping	> -0.5 dB	Åtvar, mogleg klipping
Format-validering	Støtta format	Konverter automatisk

---

Low-Latency Streaming Architectures

Sanntids talegjenkjenning

Mikrofon → Audio stream → Azure Speech SDK
                              ↓
                     ┌── Recognizing event (interim)
                     │   "Eg trur at veg..."
                     ├── Recognized event (final)
                     │   "Eg trur at vegsektoren bør investere meir."
                     │   Offset: 1800000 ticks
                     │   Duration: 30500000 ticks
                     └── SessionStopped event

Speech SDK streaming-arkitektur

import azure.cognitiveservices.speech as speechsdk
import asyncio
import json

class StreamingSpeechPipeline:
    """Sanntids tale-til-AI pipeline med låg latency."""

    def __init__(self, speech_key, speech_region, openai_client):
        self.speech_config = speechsdk.SpeechConfig(
            subscription=speech_key,
            region=speech_region
        )
        self.speech_config.speech_recognition_language = "nb-NO"
        self.speech_config.request_word_level_timestamps()
        self.speech_config.set_property(
            speechsdk.PropertyId.SpeechServiceResponse_StablePartialResultThreshold,
            "3"  # Reduser flimring i delresultat
        )

        self.openai_client = openai_client
        self.transcript_buffer = []

    async def start_streaming(self, audio_config=None):
        """Start sanntids talegjenkjenning med AI-prosessering."""
        if audio_config is None:
            audio_config = speechsdk.audio.AudioConfig(use_default_microphone=True)

        recognizer = speechsdk.SpeechRecognizer(
            speech_config=self.speech_config,
            audio_config=audio_config
        )

        # Interim-resultat for live visning
        recognizer.recognizing.connect(self._on_recognizing)

        # Endelege resultat for AI-prosessering
        recognizer.recognized.connect(self._on_recognized)

        # Start gjenkjenning
        recognizer.start_continuous_recognition()
        return recognizer

    def _on_recognizing(self, evt):
        """Handter interim-resultat (vis til brukar)."""
        print(f"\r  [interim] {evt.result.text}", end="", flush=True)

    def _on_recognized(self, evt):
        """Handter endeleg resultat (send til AI)."""
        if evt.result.reason == speechsdk.ResultReason.RecognizedSpeech:
            text = evt.result.text
            offset = evt.result.offset
            duration = evt.result.duration

            self.transcript_buffer.append({
                "text": text,
                "offset_ticks": offset,
                "duration_ticks": duration,
                "timestamp": offset / 10_000_000  # Konverter til sekund
            })

            print(f"\n[{offset/10_000_000:.1f}s] {text}")

Speech-to-Speech med Azure OpenAI

For samtaleagenter med direkte tale-til-tale:

Brukar tale → Speech SDK (STT) → Tekst
                                    ↓
                            Azure OpenAI (GPT-4o)
                                    ↓
                                  Svar-tekst
                                    ↓
                      Speech SDK (TTS) med text streaming
                                    ↓
                              Syntetisert tale → Brukar

# Text streaming for låg-latency TTS
speech_config.set_speech_synthesis_output_format(
    speechsdk.SpeechSynthesisOutputFormat.Raw24Khz16BitMonoPcm
)

tts_endpoint = (
    f"wss://{region}.tts.speech.microsoft.com"
    f"/cognitiveservices/websocket/v2"
)

# Stream GPT-output direkte til TTS
async def stream_response_to_speech(gpt_stream, synthesizer):
    """Stream GPT-4o tokens direkte til TTS for minimal latency."""
    request = speechsdk.SpeechSynthesisRequest(
        input_type=speechsdk.SpeechSynthesisRequestInputType.TextStream
    )

    # Start TTS-syntese
    result_future = synthesizer.speak_async(request)

    # Stream tekst-chunks frå GPT
    async for chunk in gpt_stream:
        if chunk.choices[0].delta.content:
            request.input_stream.write(chunk.choices[0].delta.content)

    # Steng straumen
    request.input_stream.close()
    result = await result_future

GPT-4o Realtime API for direkte tale-til-tale

Brukar tale → WebSocket → GPT-4o Realtime API → Syntetisert tale
                (bidireksjonell audiostraum)

Eigenskap	Verdi
Latency	< 500ms
Protokoll	WebSocket
Audio input	PCM 24kHz 16-bit mono
Audio output	PCM 24kHz 16-bit mono
Innhaldsmoderering	Ja, innebygd
Norsk støtte	Ja (nb-NO)

---

Error Handling og Confidence Scoring

Confidence scoring i talegjenkjenning

Azure Speech Service rapporterer confidence på fleire nivå:

Nivå	Tilgjengeleg	Brukstilfelle
Ytring-nivå	Recognized event	Filtrering av låg-kvalitets resultat
Ord-nivå	Med word timestamps aktivert	Identifisere usikre ord

speech_config.request_word_level_timestamps()
speech_config.output_format = speechsdk.OutputFormat.Detailed

def handle_detailed_result(evt):
    result = evt.result
    detailed = json.loads(result.json)

    # N-best alternatives
    for nbest in detailed.get("NBest", []):
        confidence = nbest.get("Confidence", 0)
        text = nbest.get("Display", "")

        if confidence < 0.6:
            print(f"  [LAV CONFIDENCE {confidence:.2f}] {text}")
        else:
            print(f"  [OK {confidence:.2f}] {text}")

        # Ord-nivå confidence
        for word in nbest.get("Words", []):
            word_confidence = word.get("Confidence", 0)
            if word_confidence < 0.5:
                print(f"    Usikkert ord: '{word['Word']}' ({word_confidence:.2f})")

Robuste feilhandteringsmønster

class ResilientSpeechPipeline:
    """Robust tale-pipeline med retry og fallback."""

    MAX_RETRIES = 3
    RETRY_DELAY = 1.0  # sekund

    def __init__(self, primary_config, fallback_config=None):
        self.primary_config = primary_config
        self.fallback_config = fallback_config
        self.error_counts = {"no_match": 0, "canceled": 0, "timeout": 0}

    async def recognize_with_retry(self, audio_data):
        """Gjenkjenn tale med retry-logikk."""
        for attempt in range(self.MAX_RETRIES):
            try:
                result = await self._attempt_recognition(audio_data, self.primary_config)

                if result.reason == speechsdk.ResultReason.RecognizedSpeech:
                    self.error_counts = {"no_match": 0, "canceled": 0, "timeout": 0}
                    return {"status": "success", "text": result.text, "confidence": self._get_confidence(result)}

                elif result.reason == speechsdk.ResultReason.NoMatch:
                    self.error_counts["no_match"] += 1
                    details = result.no_match_details
                    return {
                        "status": "no_match",
                        "reason": str(details.reason),
                        "recommendation": self._no_match_recommendation(details)
                    }

                elif result.reason == speechsdk.ResultReason.Canceled:
                    cancellation = result.cancellation_details
                    if cancellation.reason == speechsdk.CancellationReason.Error:
                        if "timeout" in str(cancellation.error_details).lower():
                            self.error_counts["timeout"] += 1
                            await asyncio.sleep(self.RETRY_DELAY * (attempt + 1))
                            continue
                        raise SpeechServiceError(cancellation.error_details)

            except Exception as e:
                if attempt == self.MAX_RETRIES - 1:
                    if self.fallback_config:
                        return await self._attempt_recognition(audio_data, self.fallback_config)
                    raise
                await asyncio.sleep(self.RETRY_DELAY * (attempt + 1))

    def _no_match_recommendation(self, details):
        if details.reason == speechsdk.NoMatchReason.InitialSilenceTimeout:
            return "Ingen tale detektert — sjekk mikrofon eller audiokjelde"
        elif details.reason == speechsdk.NoMatchReason.InitialBabbleTimeout:
            return "For mykje bakgrunnsstøy — forbetra audiokvalitet"
        return "Ukjend årsak — prøv på nytt"

Feilkategoriar og handtering

Feiltype	Årsak	Handling	Retry?
NoMatch - InitialSilenceTimeout	Ingen tale i starten	Sjekk mikrofon, auk timeout	Nei
NoMatch - InitialBabbleTimeout	For mykje støy	Forbetra audiokvalitet	Nei
Canceled - AuthenticationError	Ugyldig nøkkel/token	Forny token	Ja (etter fornyelse)
Canceled - ConnectionError	Nettverksproblem	Retry med exponential backoff	Ja
Canceled - ServiceTimeout	Tenesta overbelasta	Retry med delay	Ja
Canceled - RuntimeError	Intern feil	Retry	Ja

Monitoring og observability

from azure.monitor.opentelemetry import configure_azure_monitor
from opentelemetry import metrics

# Konfigurer Azure Monitor
configure_azure_monitor(connection_string=os.environ["APPLICATIONINSIGHTS_CONNECTION_STRING"])

meter = metrics.get_meter("speech-pipeline")
recognition_counter = meter.create_counter("speech.recognition.count")
confidence_histogram = meter.create_histogram("speech.recognition.confidence")
latency_histogram = meter.create_histogram("speech.recognition.latency_ms")

def track_recognition(result, latency_ms):
    recognition_counter.add(1, {"status": result["status"], "language": "nb-NO"})
    if result.get("confidence"):
        confidence_histogram.record(result["confidence"])
    latency_histogram.record(latency_ms)

---

Pipeline-arkitekturar for norsk offentleg sektor

Møtetranskribering med diarisering

Møtelyd → Audio preprocessing → Quality gate
                                     ↓
                          Speaker diarization
                          "Talar 1: ..." | "Talar 2: ..."
                                     ↓
                          Continuous recognition
                          (nb-NO med phrase list)
                                     ↓
                          Strukturert transkripsjon
                          [tidskode, talar, tekst]
                                     ↓
                     GPT-4o: Oppsummering + handlingspunkt
                                     ↓
                     Dokument: Møtereferat + opptak

Innbyggardialog via telefon

Innringar → Azure Communication Services
                      ↓
              Real-time transcription
              + Language detection
                      ↓
              GPT-4o: Klassifisering
              + Intentdeteksjon
                      ↓
              Routing til rett etat/saksbehandlar
              ELLER
              Automatisk svar via TTS

---

For Cosmo

• Azure Speech Service støttar norsk bokmål (nb-NO) fullt ut for STT og TTS. Nynorsk (nn-NO) og nordsamisk (se-NO) har meir avgrensa støtte. Bruk language identification for fleirspråklege scenario.
• Word-level timestamps og confidence scoring er avgjerande for produksjonsbruk — aktiver request_word_level_timestamps() og filtrer resultat med confidence under 0.6 for kvalitetssikring.
• Text streaming for TTS (websocket v2) reduserer opplevd latency dramatisk når du kombinerer GPT-4o med Speech Service. Stream GPT-tokens direkte til TTS i staden for å vente på fullstendig svar.
• Custom Speech med phrase lists er eit låg-innsats, høg-verdi tiltak for norske offentlege scenario. Legg til stadnamn, fagtermar og organisasjonsnamn for vesentleg forbetra gjenkjenning.
• Implementer quality gates før talegjenkjenning — sjekk SNR, varigheit og støynivå. Det reduserer feilrate og unødvendige API-kall mot Speech Service.