import streamlit as st import torch from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification import torchaudio import os import jieba import magic # Device setup: automatically selects CUDA or CPU device = "cuda" if torch.cuda.is_available() else "cpu" # Load Whisper model for Cantonese audio transcription MODEL_NAME = "alvanlii/whisper-small-cantonese" language = "zh" pipe = pipeline(task="automatic-speech-recognition", model=MODEL_NAME, chunk_length_s=60, device=device) pipe.model.config.forced_decoder_ids = pipe.tokenizer.get_decoder_prompt_ids(language=language, task="transcribe") # Transcription function (supports long audio) def transcribe_audio(audio_path): waveform, sample_rate = torchaudio.load(audio_path) duration = waveform.shape[1] / sample_rate if duration > 60: results = [] for start in range(0, int(duration), 50): end = min(start + 60, int(duration)) chunk = waveform[:, start * sample_rate:end * sample_rate] temp_filename = f"temp_chunk_{start}.wav" torchaudio.save(temp_filename, chunk, sample_rate) result = pipe(temp_filename)["text"] results.append(result) os.remove(temp_filename) return " ".join(results) return pipe(audio_path)["text"] # Load sentiment analysis model sentiment_pipe = pipeline("text-classification", model="Leo0129/CustomModel-multilingual-sentiment-analysis", device=device) # Text splitting function (using jieba for Chinese text) def split_text(text, max_length=512): words = list(jieba.cut(text)) chunks, current_chunk = [], "" for word in words: if len(current_chunk) + len(word) < max_length: current_chunk += word else: chunks.append(current_chunk) current_chunk = word if current_chunk: chunks.append(current_chunk) return chunks # Function to rate sentiment quality based on most frequent result def rate_quality(text): chunks = split_text(text) results = [] for chunk in chunks: result = sentiment_pipe(chunk)[0] label_map = {"Very Negative": "Very Poor", "Negative": "Poor", "Neutral": "Neutral", "Positive": "Good", "Very Positive": "Very Good"} results.append(label_map.get(result["label"], "Unknown")) return max(set(results), key=results.count) # Streamlit main interface def main(): st.set_page_config(page_title="Customer Service Quality Analyzer", page_icon="🎙️") # Custom CSS styling st.markdown(""" """, unsafe_allow_html=True) # Header st.markdown("""
Evaluate the service quality with simple uploading!