Sesja 7: Technologie przetwarzania mowy i asystenci głosowi
Azure Speech Services i voice AI
🎯 Cele sesji
- Implementacja Azure AI Speech Services
- Budowa systemów Speech-to-Text i Text-to-Speech
- Tworzenie asystentów głosowych
- Integracja z aplikacjami biznesowymi
🎙️ Podstawy przetwarzania mowy
Pipeline przetwarzania mowy
WEJŚCIE AUDIO → PREPROCESSING → EKSTRAKCJA CECH → MODEL → WYJŚCIE TEKSTOWE
Kluczowe komponenty:
- Preprocessing audio - redukcja szumu, normalizacja
- Ekstrakcja cech - MFCC, spektrogramy
- Modelowanie akustyczne - sieci neuronowe
- Modelowanie językowe - rozumienie kontekstu
- Dekoder - generowanie tekstu końcowego
Azure AI Speech Services
import azure.cognitiveservices.speech as speechsdk
from azure.cognitiveservices.speech import AudioDataStream, SpeechConfig
from azure.cognitiveservices.speech.audio import AudioConfig
class AzureSpeechProcessor:
def __init__(self, subscription_key, region):
self.speech_config = speechsdk.SpeechConfig(
subscription=subscription_key,
region=region
)
self.speech_config.speech_recognition_language = "pl-PL"
def speech_to_text_continuous(self):
"""Ciągłe rozpoznawanie mowy"""
# Konfiguracja audio
audio_config = speechsdk.audio.AudioConfig(use_default_microphone=True)
# Tworzenie recognizer
speech_recognizer = speechsdk.SpeechRecognizer(
speech_config=self.speech_config,
audio_config=audio_config
)
# Event handlers
def recognizing_cb(evt):
print(f"ROZPOZNAWANIE: {evt.result.text}")
def recognized_cb(evt):
if evt.result.reason == speechsdk.ResultReason.RecognizedSpeech:
print(f"ROZPOZNANO: {evt.result.text}")
self.process_recognized_text(evt.result.text)
elif evt.result.reason == speechsdk.ResultReason.NoMatch:
print("Nie rozpoznano mowy")
def canceled_cb(evt):
print(f"ANULOWANO: {evt.reason}")
if evt.reason == speechsdk.CancellationReason.Error:
print(f"Błąd: {evt.error_details}")
# Podłączenie event handlers
speech_recognizer.recognizing.connect(recognizing_cb)
speech_recognizer.recognized.connect(recognized_cb)
speech_recognizer.canceled.connect(canceled_cb)
# Rozpoczęcie ciągłego rozpoznawania
speech_recognizer.start_continuous_recognition()
return speech_recognizer
def text_to_speech(self, text, voice_name="pl-PL-AgnieszkaNeural"):
"""Konwersja tekstu na mowę"""
# Konfiguracja syntezatora
self.speech_config.speech_synthesis_voice_name = voice_name
# Tworzenie syntezatora
speech_synthesizer = speechsdk.SpeechSynthesizer(
speech_config=self.speech_config
)
# Synteza mowy
result = speech_synthesizer.speak_text_async(text).get()
if result.reason == speechsdk.ResultReason.SynthesizingAudioCompleted:
print(f"Synteza zakończona pomyślnie dla: {text}")
return result.audio_data
elif result.reason == speechsdk.ResultReason.Canceled:
cancellation_details = result.cancellation_details
print(f"Synteza anulowana: {cancellation_details.reason}")
if cancellation_details.reason == speechsdk.CancellationReason.Error:
print(f"Błąd: {cancellation_details.error_details}")
return None
def batch_transcription(self, audio_file_urls):
"""Transkrypcja wsadowa dla długich plików audio"""
# Konfiguracja dla transkrypcji wsadowej
batch_config = {
"contentUrls": audio_file_urls,
"properties": {
"diarizationEnabled": True, # Rozpoznawanie mówców
"wordLevelTimestampsEnabled": True, # Znaczniki czasowe słów
"punctuationMode": "DictatedAndAutomatic",
"profanityFilterMode": "Masked"
},
"locale": "pl-PL",
"displayName": "Batch Transcription Job"
}
# W rzeczywistości używamy REST API do batch transcription
# To jest uproszczony przykład
return batch_config
def process_recognized_text(self, text):
"""Przetwarzanie rozpoznanego tekstu"""
# Tutaj można dodać logikę biznesową
print(f"Przetwarzanie: {text}")
# Przykład: wykrywanie komend głosowych
if "zapisz" in text.lower():
self.save_command(text)
elif "wyszukaj" in text.lower():
self.search_command(text)
elif "pomoc" in text.lower():
self.help_command()
def save_command(self, text):
"""Obsługa komendy zapisywania"""
print(f"Wykonywanie komendy zapisywania: {text}")
def search_command(self, text):
"""Obsługa komendy wyszukiwania"""
search_term = text.replace("wyszukaj", "").strip()
print(f"Wyszukiwanie: {search_term}")
def help_command(self):
"""Obsługa komendy pomocy"""
help_text = "Dostępne komendy: zapisz, wyszukaj, pomoc"
self.text_to_speech(help_text)
🗣️ Asystenci głosowi - architektura
Pipeline asystenta głosowego
GŁOS → STT → NLU → DIALOG MANAGER → NLG → TTS → AUDIO WYJŚCIE
Komponenty systemu:
-
Speech-to-Text (STT)
- Konwersja audio na tekst
- Obsługa hałasu tła
- Wielojęzyczność
-
Natural Language Understanding (NLU)
- Rozpoznawanie intencji
- Ekstrakcja encji
- Zarządzanie kontekstem
-
Dialog Management
- Kontrola przepływu rozmowy
- Zarządzanie stanem
- Logika biznesowa
-
Natural Language Generation (NLG)
- Formułowanie odpowiedzi
- Personalizacja
- Kontekst konwersacji
-
Text-to-Speech (TTS)
- Synteza mowy
- Emocje w głosie
- Naturalność brzmieniea
Praktyczny projekt: Asystent głosowy
import asyncio
from typing import Dict, List
import json
class VoiceAssistant:
def __init__(self, speech_processor, nlp_processor):
self.speech_processor = speech_processor
self.nlp_processor = nlp_processor
self.conversation_state = {}
self.commands = {
"weather": self.get_weather,
"calendar": self.check_calendar,
"notes": self.manage_notes,
"search": self.search_documents
}
async def start_conversation(self, user_id="default"):
"""Rozpoczęcie sesji konwersacyjnej"""
print("🎤 Asystent głosowy uruchomiony. Powiedz 'Hello' aby rozpocząć.")
# Inicjalizacja stanu konwersacji
self.conversation_state[user_id] = {
"active": True,
"context": {},
"last_interaction": time.time()
}
# Rozpoczęcie nasłuchiwania
recognizer = self.speech_processor.speech_to_text_continuous()
# Setup callback dla rozpoznanej mowy
def on_speech_recognized(evt):
if evt.result.text:
asyncio.create_task(
self.process_voice_input(evt.result.text, user_id)
)
recognizer.recognized.connect(on_speech_recognized)
return {"status": "listening", "user_id": user_id}
async def process_voice_input(self, spoken_text: str, user_id: str):
"""Przetwarzanie wejścia głosowego"""
print(f"👤 Użytkownik: {spoken_text}")
try:
# Analiza NLU
intent_analysis = await self.nlp_processor.analyze_intent(spoken_text)
intent = intent_analysis.get("intent", "unknown")
entities = intent_analysis.get("entities", {})
confidence = intent_analysis.get("confidence", 0.0)
print(f"🧠 Rozpoznano intencję: {intent} (pewność: {confidence:.2f})")
# Sprawdź czy pewność jest wystarczająca
if confidence < 0.6:
response = "Nie jestem pewien co masz na myśli. Czy możesz powiedzieć to inaczej?"
else:
# Wykonaj akcję na podstawie intencji
response = await self.execute_intent(intent, entities, user_id)
# Odpowiedz głosowo
await self.speak_response(response)
# Aktualizuj stan konwersacji
self.conversation_state[user_id]["last_interaction"] = time.time()
self.conversation_state[user_id]["context"].update({
"last_intent": intent,
"last_entities": entities
})
except Exception as e:
error_response = "Przepraszam, wystąpił błąd w przetwarzaniu Twojej prośby."
await self.speak_response(error_response)
print(f"❌ Błąd: {str(e)}")
async def execute_intent(self, intent: str, entities: Dict, user_id: str) -> str:
"""Wykonanie akcji na podstawie rozpoznanej intencji"""
if intent in self.commands:
command_handler = self.commands[intent]
response = await command_handler(entities, user_id)
elif intent == "greeting":
response = "Cześć! Jestem Twoim asystentem AI. Jak mogę pomóc?"
elif intent == "goodbye":
response = "Do widzenia! Miło było z Tobą rozmawiać."
self.conversation_state[user_id]["active"] = False
else:
response = f"Rozpoznałem intencję '{intent}', ale nie wiem jeszcze jak na nią odpowiedzieć. Uczę się cały czas!"
return response
async def speak_response(self, text: str):
"""Wypowiedzenie odpowiedzi"""
print(f"🤖 Asystent: {text}")
# Konwersja na mowę
audio_data = self.speech_processor.text_to_speech(text)
if audio_data:
print("🔊 Odtwarzanie odpowiedzi audio...")
# Tutaj byłby kod odtwarzania audio
return audio_data
async def get_weather(self, entities: Dict, user_id: str) -> str:
"""Handler dla zapytań o pogodę"""
location = entities.get("location", "Warsaw")
# Symulacja API pogodowego
weather_data = {
"temperature": "22°C",
"condition": "słonecznie",
"humidity": "45%"
}
return f"Pogoda w {location}: {weather_data['temperature']}, {weather_data['condition']}, wilgotność {weather_data['humidity']}"
async def check_calendar(self, entities: Dict, user_id: str) -> str:
"""Handler dla zapytań o kalendarz"""
date = entities.get("date", "dzisiaj")
# Symulacja integracji z kalendarzem
events = [
"10:00 - Spotkanie zespołu",
"14:00 - Prezentacja AI",
"16:30 - Warsztat Azure"
]
if events:
events_text = ", ".join(events)
return f"Na {date} masz zaplanowane: {events_text}"
else:
return f"Na {date} nie masz żadnych zaplanowanych wydarzeń"
async def search_documents(self, entities: Dict, user_id: str) -> str:
"""Handler dla wyszukiwania dokumentów"""
query = entities.get("query", "")
if not query:
return "Czego szukasz? Powiedz mi więcej szczegółów."
# Symulacja wyszukiwania
search_results = [
"Dokument 1: Azure AI Best Practices",
"Dokument 2: Speech Recognition Tutorial",
"Dokument 3: Voice Assistant Implementation"
]
if search_results:
return f"Znalazłem {len(search_results)} dokumentów dla zapytania '{query}'. Pierwszy to: {search_results[0]}"
else:
return f"Nie znalazłem żadnych dokumentów dla zapytania '{query}'"
🛠️ Praktyczne zastosowania
Implementacja voice-enabled aplikacji
class VoiceEnabledApp:
def __init__(self):
self.voice_assistant = None
self.is_listening = False
self.supported_languages = ["pl-PL", "en-US", "de-DE", "fr-FR"]
async def initialize_voice_features(self, config):
"""Inicjalizacja funkcji głosowych"""
# Setup Speech Services
speech_processor = AzureSpeechProcessor(
config["azure_speech_key"],
config["azure_region"]
)
# Setup NLP processor (simplified)
nlp_processor = SimpleNLPProcessor()
# Initialize voice assistant
self.voice_assistant = VoiceAssistant(speech_processor, nlp_processor)
print("✅ Voice features initialized")
return {"status": "ready", "languages": self.supported_languages}
async def start_voice_interaction(self, user_id):
"""Rozpoczęcie interakcji głosowej"""
if not self.voice_assistant:
raise ValueError("Voice assistant not initialized")
self.is_listening = True
result = await self.voice_assistant.start_conversation(user_id)
return result
def stop_voice_interaction(self):
"""Zatrzymanie interakcji głosowej"""
self.is_listening = False
print("🔇 Voice interaction stopped")
return {"status": "stopped"}
# Przykład integracji z aplikacją web
class WebVoiceInterface:
def __init__(self, voice_app):
self.voice_app = voice_app
async def handle_voice_request(self, request_data):
"""Obsługa zapytania głosowego z aplikacji web"""
audio_data = request_data.get("audio_data")
user_id = request_data.get("user_id", "web_user")
# Transkrypcja audio
if audio_data:
transcript = await self.voice_app.speech_processor.transcribe_audio(audio_data)
# Przetwarzanie jako tekst
response = await self.voice_app.voice_assistant.process_voice_input(
transcript["text"], user_id
)
# Zwróć zarówno tekst jak i audio odpowiedzi
response_audio = await self.voice_app.voice_assistant.speak_response(response)
return {
"transcript": transcript["text"],
"response_text": response,
"response_audio": response_audio,
"confidence": transcript["confidence"]
}
return {"error": "No audio data provided"}
🎯 Zaawansowane funkcje
Speaker Diarization (rozpoznawanie mówców)
class SpeakerDiarization:
def __init__(self, speech_config):
self.speech_config = speech_config
async def analyze_conversation(self, audio_file_path):
"""Analiza rozmowy z identyfikacją mówców"""
# Konfiguracja conversation transcriber
audio_config = speechsdk.audio.AudioConfig(filename=audio_file_path)
conversation_transcriber = speechsdk.transcription.ConversationTranscriber(
speech_config=self.speech_config,
audio_config=audio_config
)
transcription_results = []
def transcribed_cb(evt):
"""Callback dla transkrypcji z informacją o mówcy"""
result = {
"speaker_id": evt.result.speaker_id,
"text": evt.result.text,
"timestamp": evt.result.offset,
"duration": evt.result.duration
}
transcription_results.append(result)
print(f"Mówca {evt.result.speaker_id}: {evt.result.text}")
# Podłączenie callback
conversation_transcriber.transcribed.connect(transcribed_cb)
# Rozpoczęcie transkrypcji
await conversation_transcriber.start_transcribing_async()
# Czekaj na zakończenie (w rzeczywistości trzeba obsłużyć to lepiej)
await asyncio.sleep(10) # Przykładowy czas
await conversation_transcriber.stop_transcribing_async()
# Analiza wyników
speaker_analysis = self.analyze_speakers(transcription_results)
return {
"transcription": transcription_results,
"speaker_analysis": speaker_analysis,
"total_speakers": len(set(r["speaker_id"] for r in transcription_results))
}
def analyze_speakers(self, transcription_results):
"""Analiza zachowań mówców"""
speaker_stats = {}
for result in transcription_results:
speaker_id = result["speaker_id"]
if speaker_id not in speaker_stats:
speaker_stats[speaker_id] = {
"total_words": 0,
"total_duration": 0,
"segments": 0
}
word_count = len(result["text"].split())
duration = result["duration"] / 10000000 # Convert to seconds
speaker_stats[speaker_id]["total_words"] += word_count
speaker_stats[speaker_id]["total_duration"] += duration
speaker_stats[speaker_id]["segments"] += 1
return speaker_stats
✅ Zadania praktyczne
Zadanie 1: Basic Speech Recognition (30 min)
- Skonfiguruj Azure Speech Service
- Zaimplementuj simple STT
- Przetestuj z różnymi akcentami i prędkościami mowy
- Dodaj obsługę błędów
Zadanie 2: Text-to-Speech Implementation (30 min)
- Zaimplementuj TTS z różnymi głosami
- Przetestuj SSML dla kontroli prozodii
- Dodaj emocjonalne style głosu
- Optymalizuj jakość audio
Zadanie 3: Simple Voice Assistant (45 min)
- Stwórz prostego asystenta głosowego
- Zaimplementuj 3-5 podstawowych komend
- Dodaj conversation state management
- Przetestuj end-to-end workflow
Zadanie 4: Speaker Diarization (15 min)
- Przeanalizuj nagranie z wieloma mówcami
- Zidentyfikuj różnych uczestników rozmowy
- Wygeneruj podsumowanie dla każdego mówcy
- Stwórz raport z analizy konwersacji
📊 Metryki wydajności
- Word Error Rate (WER) < 5% dla czystego audio
- Latencja STT < 500ms dla real-time processing
- Naturalność TTS > 4.5/5 w ocenie użytkowników
- Intent recognition accuracy > 90%