# 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 ```python 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 | ```python # 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 ```python 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 ```python 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 ``` ```python # 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 | ```python 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 ```python 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 ```python 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.