Create kiswahili_hf_app.py

kiswahili_hf_app.py

@@ -0,0 +1,276 @@
+#!/usr/bin/env python3
+"""
+Kiswahili Voice Agent for Hugging Face Spaces
+Natural conversational Kiswahili voice-to-voice assistant
+"""
+
+import gradio as gr
+import os
+from datetime import datetime
+import json
+
+# Try to import optional dependencies
+try:
+    from gtts import gTTS
+    HAS_GTTS = True
+except ImportError:
+    HAS_GTTS = False
+
+try:
+    import speech_recognition as sr
+    HAS_SR = True
+except ImportError:
+    HAS_SR = False
+
+try:
+    import requests
+    HAS_REQUESTS = True
+except ImportError:
+    HAS_REQUESTS = False
+
+# Conversation history
+conversation_history = []
+conversation_id = None
+
+# Natural Kiswahili system prompt
+SYSTEM_PROMPT = """Wewe ni Manus, msaidizi wa sauti wa Kiswahili ambaye ana tabia nzuri na karimu. 
+Unazungumza Kiswahili safi na asilia, na unafahamu utamaduni wa Kiswahili.
+Katika kila jibu, jaribu kuuliza swali la mfuatano ili kuendelea na mazungumzo.
+Jibu kwa ufupi lakini kwa maana - kwa kawaida 1-2 sentensi.
+Kila jibu lazima liwe na swali au kauli inayokamatia mazungumzo."""
+
+# Natural Kiswahili UI strings
+UI_STRINGS = {
+    "title": "🎙️ Manus - Msaidizi wa Sauti wa Kiswahili",
+    "subtitle": "Mazungumzo ya asilia kwa Kiswahili",
+    "instruction": "Bonyeza kurekodi, sema kitu kwa Kiswahili, kisha bonyeza kuacha.",
+    "status_recording": "🔴 Inasikiliza...",
+    "status_processing": "⚙️ Inachakata...",
+    "status_ready": "✅ Tayari",
+    "status_error": "❌ Hitilafu",
+    "user_label": "Wewe:",
+    "assistant_label": "Manus:",
+    "reset_button": "🔄 Anza Upya",
+    "reset_confirm": "Mazungumzo yamefutwa. Karibu tena!",
+    "error_audio": "Haiwezekani kusoma sauti. Tafadhali jaribu tena.",
+    "error_process": "Haiwezekani kuchakata sauti. Tafadhali jaribu tena.",
+    "welcome": "Habari! Naitwa Manus. Karibu sana! Unaweza kusema kitu chochote kwa Kiswahili, na nitakujibu.",
+}
+
+def transcribe_audio(audio_file):
+    """Transcribe Kiswahili audio using speech recognition"""
+    if not HAS_SR:
+        return "Haiwezekani kusoma sauti - moduli haipo"
+    
+    try:
+        recognizer = sr.Recognizer()
+        with sr.AudioFile(audio_file) as source:
+            audio = recognizer.record(source)
+        
+        # Try to recognize Kiswahili
+        text = recognizer.recognize_google(audio, language="sw-TZ")
+        return text
+    except sr.UnknownValueError:
+        return "Haiwezekani kuelewa sauti. Tafadhali jaribu tena."
+    except sr.RequestError:
+        return "Haiwezekani kuunganisha na huduma ya mtandao."
+    except Exception as e:
+        return f"Hitilafu: {str(e)}"
+
+def generate_response(user_text):
+    """Generate natural Kiswahili response using simple logic"""
+    # Simple rule-based responses for demo (replace with API call for better results)
+    
+    user_text_lower = user_text.lower()
+    
+    # Greeting responses
+    greetings = {
+        "habari": "Habari nzuri! Niko sawa. Wewe je, uko sawa?",
+        "jina": "Naitwa Manus, msaidizi wako wa sauti. Jina lako nani?",
+        "asante": "Karibu sana! Kuna kitu kingine ninachoweza kukusaidia?",
+        "pole": "Pole pole! Kila kitu kitakuwa sawa. Unaweza kusema nini kinachokukosesha?",
+        "ndiyo": "Nzuri! Unaweza kusema zaidi?",
+        "hapana": "Sawa. Kuna kitu kingine?",
+    }
+    
+    # Check for keywords
+    for keyword, response in greetings.items():
+        if keyword in user_text_lower:
+            return response
+    
+    # Default conversational response
+    default_responses = [
+        "Hiyo ni kitu kizuri! Unaweza kusema zaidi kuhusu hilo?",
+        "Nimeelewa. Na kisha nini?",
+        "Sawa! Hiyo ni kitu muhimu. Unaweza kueneza?",
+        "Nzuri sana! Unaweza kusema kitu kingine?",
+        "Hiyo ni interesting! Unaweza kuniambia zaidi?",
+    ]
+    
+    import random
+    return random.choice(default_responses)
+
+def text_to_speech_kiswahili(text):
+    """Convert Kiswahili text to speech"""
+    if not HAS_GTTS:
+        return None
+    
+    try:
+        tts = gTTS(text=text, lang='sw', slow=False)
+        audio_file = "/tmp/response.mp3"
+        tts.save(audio_file)
+        return audio_file
+    except Exception as e:
+        print(f"TTS Error: {e}")
+        return None
+
+def process_voice_input(audio_input):
+    """Main processing function for voice input"""
+    global conversation_history, conversation_id
+    
+    if audio_input is None:
+        return (
+            UI_STRINGS["status_error"],
+            UI_STRINGS["error_audio"],
+            None,
+            gr.update(value="")
+        )
+    
+    try:
+        # Step 1: Transcribe user audio
+        user_text = transcribe_audio(audio_input)
+        
+        if "Hitilafu" in user_text or "Haiwezekani" in user_text:
+            return (
+                UI_STRINGS["status_error"],
+                user_text,
+                None,
+                gr.update(value="")
+            )
+        
+        # Step 2: Generate response
+        assistant_response = generate_response(user_text)
+        
+        # Step 3: Convert response to speech
+        audio_response = text_to_speech_kiswahili(assistant_response)
+        
+        # Step 4: Update conversation history
+        conversation_history.append({
+            "timestamp": datetime.now().isoformat(),
+            "user": user_text,
+            "assistant": assistant_response
+        })
+        
+        # Format conversation display
+        conversation_text = ""
+        for msg in conversation_history:
+            conversation_text += f"\n**{UI_STRINGS['user_label']}** {msg['user']}\n"
+            conversation_text += f"**{UI_STRINGS['assistant_label']}** {msg['assistant']}\n"
+        
+        return (
+            UI_STRINGS["status_ready"],
+            conversation_text,
+            audio_response,
+            gr.update(value="")  # Clear recorder
+        )
+    
+    except Exception as e:
+        error_msg = f"{UI_STRINGS['status_error']}: {str(e)}"
+        return (
+            UI_STRINGS["status_error"],
+            error_msg,
+            None,
+            gr.update(value="")
+        )
+
+def reset_conversation():
+    """Reset conversation history"""
+    global conversation_history
+    conversation_history = []
+    return (
+        UI_STRINGS["status_ready"],
+        UI_STRINGS["reset_confirm"],
+        None,
+        gr.update(value="")
+    )
+
+# Create Gradio interface
+with gr.Blocks(title=UI_STRINGS["title"], theme=gr.themes.Soft()) as demo:
+    gr.Markdown(f"# {UI_STRINGS['title']}")
+    gr.Markdown(f"### {UI_STRINGS['subtitle']}")
+    gr.Markdown(f"> {UI_STRINGS['instruction']}")
+    
+    with gr.Row():
+        with gr.Column(scale=1):
+            # Status indicator
+            status_display = gr.Textbox(
+                value=UI_STRINGS["status_ready"],
+                label="Hali",
+                interactive=False,
+                lines=1
+            )
+            
+            # Voice recorder
+            audio_input = gr.Audio(
+                label="🎤 Rekodi Sauti",
+                type="filepath",
+                sources=["microphone"]
+            )
+            
+            # Process button
+            process_btn = gr.Button(
+                "📤 Tuma Sauti",
+                variant="primary",
+                size="lg"
+            )
+            
+            # Reset button
+            reset_btn = gr.Button(
+                UI_STRINGS["reset_button"],
+                variant="secondary"
+            )
+        
+        with gr.Column(scale=1):
+            # Conversation history
+            conversation_display = gr.Markdown(
+                value=f"**{UI_STRINGS['assistant_label']}** {UI_STRINGS['welcome']}\n",
+                label="Mazungumzo"
+            )
+            
+            # Audio response player
+            audio_output = gr.Audio(
+                label="🔊 Jibu la Sauti",
+                type="filepath",
+                interactive=False
+            )
+    
+    # Event handlers
+    process_btn.click(
+        fn=process_voice_input,
+        inputs=[audio_input],
+        outputs=[status_display, conversation_display, audio_output, audio_input]
+    )
+    
+    reset_btn.click(
+        fn=reset_conversation,
+        outputs=[status_display, conversation_display, audio_output, audio_input]
+    )
+    
+    # Auto-process when audio is recorded
+    audio_input.change(
+        fn=lambda audio: process_voice_input(audio) if audio else (
+            UI_STRINGS["status_ready"],
+            conversation_display.value,
+            None,
+            gr.update(value="")
+        ),
+        inputs=[audio_input],
+        outputs=[status_display, conversation_display, audio_output, audio_input]
+    )
+
+if __name__ == "__main__":
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        share=True
+    )