Spaces:
Paused
Paused
| import os | |
| import asyncio | |
| import httpx | |
| import xml.etree.ElementTree as ET | |
| import re as _re | |
| from datetime import datetime | |
| import wave | |
| import time as _time | |
| import boto3 | |
| from botocore.client import Config | |
| from livekit import rtc | |
| try: | |
| from zoneinfo import ZoneInfo | |
| except ImportError: | |
| ZoneInfo = None | |
| from dotenv import load_dotenv | |
| from livekit.agents import AutoSubscribe, JobContext, JobRequest, WorkerOptions, cli, RoomInputOptions | |
| from livekit.agents import AgentSession, Agent | |
| from livekit.plugins import groq, silero, openai | |
| from tts_mms import MmsTTS | |
| from tts_api import ApiTTS | |
| from tts_edge import EdgeTTS | |
| try: | |
| from livekit.agents import function_tool | |
| except ImportError: | |
| def function_tool(f): | |
| return f | |
| load_dotenv() | |
| # ────────────────────────────────────────────────────────────────────────────── | |
| # Voice Registry (label, tts config, STT language, region, flag) | |
| # ────────────────────────────────────────────────────────────────────────────── | |
| VOICE_INFO = { | |
| # ─── East Africa ────────────────────────────────────────────────────────── | |
| "swh": { | |
| "label": "Kiswahili", "type": "mms", "repo": "facebook/mms-tts-swh", | |
| "lang_name": "Kiswahili", "whisper_lang": "sw", | |
| "region": "East Africa", "flag": "🇰🇪", | |
| "greeting": "Habari yako! Naitwa Fiti. Niko tayari kuanza, ungetaka tuongee kuhusu nini leo?", | |
| "sample": "Unaweza kuniambia zaidi?", | |
| }, | |
| "swh_v2": { | |
| "label": "Kiswahili Daudi", "type": "mms", "repo": "Benjamin-png/swahili-mms-tts-finetuned", | |
| "lang_name": "Kiswahili", "whisper_lang": "sw", | |
| "region": "East Africa", "flag": "🇹🇿", | |
| "greeting": "Habari yako! Naitwa Daudi. Niko tayari kuanza, ungetaka tuongee kuhusu nini leo?", | |
| "sample": "Ninafurahi kukusaidia.", | |
| }, | |
| "swh_ke_r": { | |
| "label": "Kiswahili Rafiki", "type": "edge", "voice_id": "sw-KE-RafikiNeural", | |
| "lang_name": "Kiswahili", "whisper_lang": "sw", | |
| "region": "East Africa", "flag": "🇰🇪", | |
| "greeting": "Hujambo na karibu sana. Naitwa Rafiki. Nipo hapa kuzungumza nawe, ungetaka tukague nini leo?", | |
| "sample": "Ninafurahi kukusaidia.", | |
| }, | |
| "swh_tz_r": { | |
| "label": "Kiswahili Rehema", "type": "edge", "voice_id": "sw-TZ-RehemaNeural", | |
| "lang_name": "Kiswahili", "whisper_lang": "sw", | |
| "region": "East Africa", "flag": "🇹🇿", | |
| "greeting": "Hujambo na karibu sana. Naitwa Rehema. Nipo hapa kuzungumza nawe, ungetaka tukague nini leo?", | |
| "sample": "Ninafurahi kukusaidia.", | |
| }, | |
| "kik": { | |
| "label": "Gĩkũyũ", "type": "mms", "repo": "facebook/mms-tts-kik", | |
| "lang_name": "Kikuyu", "whisper_lang": "sw", | |
| "region": "East Africa", "flag": "🇰🇪", | |
| "greeting": "Nĩ DUOL, mũteithia waku. Ũrĩ mwega?", | |
| "sample": "Nĩ ũhoro waku?", | |
| }, | |
| "luo": { | |
| "label": "Dholuo", "type": "mms", "repo": "rnjema-unima/mms-tts-luo-baseline", | |
| "lang_name": "Luo", "whisper_lang": "sw", | |
| "region": "East Africa", "flag": "🇰🇪", | |
| "greeting": "Misawa! An DUOL, jakonyi. Idhi nade?", | |
| "sample": "An dimo ni mondo akony?", | |
| }, | |
| "saq": { | |
| "label": "Samburu", "type": "mms", "repo": "facebook/mms-tts-saq", | |
| "lang_name": "Samburu", "whisper_lang": "en", | |
| "region": "East Africa", "flag": "🇰🇪", | |
| "greeting": "Supa! Nanu DUOL.", | |
| "sample": "Takwenya?", | |
| }, | |
| "som": { | |
| "label": "Soomaali", "type": "edge", "voice_id": "so-SO-UbaxNeural", | |
| "lang_name": "Somali", "whisper_lang": "so", | |
| "region": "East Africa", "flag": "🇸🇴", | |
| "greeting": "Nabad! Anigu waxaan ahay DUOL, caawiyahaaga caqliga ah. Sidee baad tahay?", | |
| "sample": "Sidee baad tahay?", | |
| }, | |
| "orm": { | |
| "label": "Oromo", "type": "mms", "repo": "facebook/mms-tts-orm", | |
| "lang_name": "Oromo", "whisper_lang": "om", | |
| "region": "East Africa", "flag": "🇪🇹", | |
| "greeting": "Akkam! Ani DUOL, gargaaraa kee ti. Akkam bultee?", | |
| "sample": "Akkam bultee?", | |
| }, | |
| "lug": { | |
| "label": "Luganda", "type": "mms", "repo": "facebook/mms-tts-lug", | |
| "lang_name": "Luganda", "whisper_lang": "sw", | |
| "region": "East Africa", "flag": "🇺🇬", | |
| "greeting": "Oli otya! Nze DUOL, omuyambi wo ow'amagezi. Ndi muzima. Oli otya?", | |
| "sample": "Ndi muzima. Oli otya?", | |
| }, | |
| "kin": { | |
| "label": "Kinyarwanda", "type": "mms", "repo": "facebook/mms-tts-kin", | |
| "lang_name": "Kinyarwanda", "whisper_lang": "rw", | |
| "region": "East Africa", "flag": "🇷🇼", | |
| "greeting": "Muraho! Ndi DUOL, umufasha wawe w'ubwenge buhangano. Amakuru?", | |
| "sample": "Amakuru?", | |
| }, | |
| "amh": { | |
| "label": "Amharic", "type": "mms", "repo": "facebook/mms-tts-amh", | |
| "lang_name": "Amharic", "whisper_lang": "am", | |
| "region": "East Africa", "flag": "🇪🇹", | |
| "greeting": "ሰላም! እኔ ዱኦል ነኝ።", | |
| "sample": "እንዴት ነህ/ነሽ?", | |
| }, | |
| "tir": { | |
| "label": "Tigrinya", "type": "mms", "repo": "facebook/mms-tts-tir", | |
| "lang_name": "Tigrinya", "whisper_lang": "am", | |
| "region": "East Africa", "flag": "🇪🇷", | |
| "greeting": "ሰላም! ኣነ ዱኦል እየ።", | |
| "sample": "ከመይ አለካ?", | |
| }, | |
| "mlg": { | |
| "label": "Malagasy", "type": "mms", "repo": "facebook/mms-tts-mlg", | |
| "lang_name": "Malagasy", "whisper_lang": "mg", | |
| "region": "East Africa", "flag": "🇲🇬", | |
| "greeting": "Salama! Izaho i DUOL.", | |
| "sample": "Ahoana ny vaovao?", | |
| }, | |
| # ─── West Africa ─────────────────────────────────────────────────────────── | |
| "hau": { | |
| "label": "Hausa", "type": "mms", "repo": "facebook/mms-tts-hau", | |
| "lang_name": "Hausa", "whisper_lang": "ha", | |
| "region": "West Africa", "flag": "🇳🇬", | |
| "greeting": "Sannu! Ni ne DUOL, mataimakinka na fasaha. Yaya kake?", | |
| "sample": "Yaya kake?", | |
| }, | |
| "yor": { | |
| "label": "Yorùbá", "type": "mms", "repo": "facebook/mms-tts-yor", | |
| "lang_name": "Yoruba", "whisper_lang": "yo", | |
| "region": "West Africa", "flag": "🇳🇬", | |
| "greeting": "Bawo ni! Emi ni DUOL.", | |
| "sample": "Báwo ni o ṣe wà?", | |
| }, | |
| "ibo": { | |
| "label": "Igbo", "type": "mms", "repo": "facebook/mms-tts-ibo", | |
| "lang_name": "Igbo", "whisper_lang": "en", | |
| "region": "West Africa", "flag": "🇳🇬", | |
| "greeting": "Ndeewo! Abụ m DUOL.", | |
| "sample": "Kedụ ka i mere?", | |
| }, | |
| "wol": { | |
| "label": "Wolof", "type": "mms", "repo": "facebook/mms-tts-wol", | |
| "lang_name": "Wolof", "whisper_lang": "fr", | |
| "region": "West Africa", "flag": "🇸🇳", | |
| "greeting": "Nanga def! Maa ngi DUOL.", | |
| "sample": "Nanga def?", | |
| }, | |
| "lin": { | |
| "label": "Lingala", "type": "mms", "repo": "facebook/mms-tts-lin", | |
| "lang_name": "Lingala", "whisper_lang": "fr", | |
| "region": "West Africa", "flag": "🇨🇩", | |
| "greeting": "Mbote! Ngai DUOL.", | |
| "sample": "Ozali malamu?", | |
| }, | |
| "aka": { | |
| "label": "Akan (Twi)", "type": "mms", "repo": "facebook/mms-tts-aka", | |
| "lang_name": "Akan", "whisper_lang": "en", | |
| "region": "West Africa", "flag": "🇬🇭", | |
| "greeting": "Ete sɛn! Me din de DUOL.", | |
| "sample": "Wo ho te sɛn?", | |
| }, | |
| "bam": { | |
| "label": "Bambara", "type": "mms", "repo": "facebook/mms-tts-bam", | |
| "lang_name": "Bambara", "whisper_lang": "fr", | |
| "region": "West Africa", "flag": "🇲🇱", | |
| "greeting": "I ni ce! Ne ye DUOL ye.", | |
| "sample": "I ka kɛnɛ wa?", | |
| }, | |
| "ful": { | |
| "label": "Fulfulde", "type": "mms", "repo": "facebook/mms-tts-ful", | |
| "lang_name": "Fula", "whisper_lang": "en", | |
| "region": "West Africa", "flag": "🌍", | |
| "greeting": "Jam waali! Mi woni DUOL.", | |
| "sample": "No mbaadum?", | |
| }, | |
| # ─── North Africa ────────────────────────────────────────────────────────── | |
| "arb": { | |
| "label": "عربي (Arabic)", "type": "edge", "voice_id": "ar-EG-SalmaNeural", | |
| "lang_name": "Arabic", "whisper_lang": "ar", | |
| "region": "North Africa", "flag": "🌍", | |
| "greeting": "مرحباً! أنا دوول، مساعدك الصوتي الذكي.", | |
| "sample": "كيف حالك؟", | |
| }, | |
| # ─── Southern Africa ─────────────────────────────────────────────────────── | |
| "zul": { | |
| "label": "Zulu", "type": "edge", "voice_id": "zu-ZA-ThandoNeural", | |
| "lang_name": "Zulu", "whisper_lang": "zu", | |
| "region": "Southern Africa", "flag": "🇿🇦", | |
| "greeting": "Sawubona! NginguDUOL, umsizi wakho we-AI.", | |
| "sample": "Unjani?", | |
| }, | |
| "xho": { | |
| "label": "Xhosa", "type": "mms", "repo": "facebook/mms-tts-xho", | |
| "lang_name": "Xhosa", "whisper_lang": "en", | |
| "region": "Southern Africa", "flag": "🇿🇦", | |
| "greeting": "Molo! NdinguDUOL.", | |
| "sample": "Unjani?", | |
| }, | |
| "sna": { | |
| "label": "Shona", "type": "mms", "repo": "facebook/mms-tts-sna", | |
| "lang_name": "Shona", "whisper_lang": "en", | |
| "region": "Southern Africa", "flag": "🇿🇼", | |
| "greeting": "Mhoro! Ndiri DUOL.", | |
| "sample": "Makadini?", | |
| }, | |
| "nya": { | |
| "label": "Chichewa", "type": "mms", "repo": "facebook/mms-tts-nya", | |
| "lang_name": "Chichewa", "whisper_lang": "en", | |
| "region": "Southern Africa", "flag": "🇲🇼", | |
| "greeting": "Moni! Ine ndine DUOL.", | |
| "sample": "Muli bwanji?", | |
| }, | |
| "tsn": { | |
| "label": "Setswana", "type": "mms", "repo": "facebook/mms-tts-tsn", | |
| "lang_name": "Setswana", "whisper_lang": "en", | |
| "region": "Southern Africa", "flag": "🇧🇼", | |
| "greeting": "Dumela! Ke DUOL.", | |
| "sample": "O kae?", | |
| }, | |
| "afr": { | |
| "label": "Afrikaans", "type": "mms", "repo": "facebook/mms-tts-afr", | |
| "lang_name": "Afrikaans", "whisper_lang": "af", | |
| "region": "Southern Africa", "flag": "🇿🇦", | |
| "greeting": "Hallo! Ek is DUOL.", | |
| "sample": "Hoe gaan dit?", | |
| }, | |
| # ─── Premium API Voices ──────────────────────────────────────────────────── | |
| "swh_female": { | |
| "label": "Kiswahili (Female)", "type": "api", "space": "stano03/habari-xtts", | |
| "lang_name": "Kiswahili", "whisper_lang": "sw", | |
| "region": "East Africa", "flag": "🇰🇪", | |
| "greeting": "Habari! Mimi ni DUOL, msaidizi wako.", | |
| "sample": "Ninafurahi kukusaidia.", | |
| "api_kwargs": {}, | |
| }, | |
| "swh_premium": { | |
| "label": "Kiswahili (Sauti)", "type": "api", "space": "sta-03/habari-f5tts", | |
| "lang_name": "Kiswahili", "whisper_lang": "sw", | |
| "region": "East Africa", "flag": "🇰🇪", | |
| "greeting": "Habari! Mimi ni Jarvis.", | |
| "sample": "Habari!", | |
| "api_kwargs": {"language": "Kiswahili"}, | |
| }, | |
| "kln": { | |
| "label": "Kalenjin", "type": "api", "space": "sta-03/habari-f5tts", | |
| "lang_name": "Kalenjin", "whisper_lang": "sw", | |
| "region": "East Africa", "flag": "🇰🇪", | |
| "greeting": "Chamgei! Ane Jarvis.", | |
| "sample": "Chamgei!", | |
| "api_kwargs": {"language": "Kalenjin"}, | |
| }, | |
| "yor_premium": { | |
| "label": "Yorùbá (Premium)", "type": "api", "space": "Stanley03/habari-yarngpt", | |
| "lang_name": "Yoruba", "whisper_lang": "yo", | |
| "region": "West Africa", "flag": "🇳🇬", | |
| "greeting": "Bawo ni! Emi ni DUOL.", | |
| "sample": "Bawo ni!", | |
| "api_kwargs": {"language": "Yoruba", "voice": "yoruba_female2"}, | |
| }, | |
| "hau_premium": { | |
| "label": "Hausa (Premium)", "type": "api", "space": "Stanley03/habari-yarngpt", | |
| "lang_name": "Hausa", "whisper_lang": "ha", | |
| "region": "West Africa", "flag": "🇳🇬", | |
| "greeting": "Sannu! Ni ne DUOL.", | |
| "sample": "Sannu!", | |
| "api_kwargs": {"language": "Hausa", "voice": "hausa_female1"}, | |
| }, | |
| "igb": { | |
| "label": "Igbo (Premium)", "type": "api", "space": "Stanley03/habari-yarngpt", | |
| "lang_name": "Igbo", "whisper_lang": "en", | |
| "region": "West Africa", "flag": "🇳🇬", | |
| "greeting": "Ndeewo! Abụ m DUOL.", | |
| "sample": "Ndeewo!", | |
| "api_kwargs": {"language": "Igbo", "voice": "igbo_female2"}, | |
| }, | |
| "eng_naija": { | |
| "label": "Nigerian English", "type": "api", "space": "Stanley03/habari-yarngpt", | |
| "lang_name": "English", "whisper_lang": "en", | |
| "region": "West Africa", "flag": "🇳🇬", | |
| "greeting": "How far! I be Jarvis.", | |
| "sample": "How far!", | |
| "api_kwargs": {"language": "English", "voice": "idera"}, | |
| }, | |
| "swh_nursetoto": { | |
| "label": "Nurse Toto (Custom)", "type": "api", "space": "Stanley-07/habari-nursetoto", | |
| "lang_name": "Kiswahili", "whisper_lang": "sw", | |
| "region": "East Africa", "flag": "🇰🇪", | |
| "greeting": "Habari! Mimi ni Jarvis.", | |
| "sample": "Habari!", | |
| "api_kwargs": {}, | |
| }, | |
| "kis_premium": { | |
| "label": "Kisii (Ekegusii)", "type": "api", "space": "Stanley-07/habari-kisii", | |
| "lang_name": "Ekegusii", "whisper_lang": "sw", | |
| "region": "East Africa", "flag": "🇰🇪", | |
| "greeting": "Mbuya! Inche ni Jarvis.", | |
| "sample": "Mbuya!", | |
| "api_kwargs": {}, | |
| }, | |
| "habari_pro": { | |
| "label": "Habari Pro", "type": "api", "space": "Stanley-07/habari-kisii", | |
| "lang_name": "African", "whisper_lang": "en", | |
| "region": "All Africa", "flag": "🌍", | |
| "greeting": "Hello! I am Jarvis, Africa's most advanced AI voice assistant.", | |
| "sample": "Hello!", | |
| "api_kwargs": {}, | |
| }, | |
| } | |
| # ────────────────────────────────────────────────────────────────────────────── | |
| # Domain-Aware Cultural System Prompts | |
| # ────────────────────────────────────────────────────────────────────────────── | |
| DOMAIN_CONTEXT = { | |
| "health": """ | |
| You are also a trained community health advisor. | |
| You can provide first-aid guidance, basic symptom triage, and refer users to the nearest health facility. | |
| You know about common health challenges in Africa: malaria, cholera, typhoid, maternal health, HIV/AIDS, COVID-19. | |
| Always advise users to seek professional medical help for serious symptoms. Never diagnose definitively.""", | |
| "agriculture": """ | |
| You are also an agricultural advisor for smallholder African farmers. | |
| You understand crops: maize, tea, coffee, sorghum, cassava, millet, wheat, rice, groundnuts, beans. | |
| You can advise on planting seasons, weather impact, pest control, market prices, and post-harvest storage. | |
| You know about climate patterns in different African regions and their effect on farming.""", | |
| "education": """ | |
| You are also a patient and encouraging tutor. | |
| You can help with mathematics, sciences, literature, history, geography, and language learning. | |
| You celebrate every student's progress and adapt your explanations to be simple and clear. | |
| You can help students prepare for KCPE, KCSE, WAEC, NECO, and other African national exams.""", | |
| "finance": """ | |
| You are also a financial literacy advisor. | |
| You understand M-Pesa, Airtel Money, MTN Mobile Money, and other African mobile payment systems. | |
| You can help with budgeting, savings advice, micro-loan information, and understanding financial products. | |
| Always recommend consulting a licensed financial advisor for major investment decisions.""", | |
| "general": "", | |
| } | |
| def build_instructions(info: dict, spoken_language: str, domain: str = "general") -> str: | |
| lang_name = info["lang_name"] | |
| region = info.get("region", "Africa") | |
| domain_txt = DOMAIN_CONTEXT.get(domain, "") | |
| if spoken_language == "english": | |
| lang_prompt = f"You speak English with a warm, authentic {lang_name} accent — this is your natural voice." | |
| else: | |
| lang_prompt = ( | |
| f"CRITICAL LANGUAGE DIRECTIVE: You MUST speak EXCLUSIVELY in {lang_name}. " | |
| f"Do NOT respond in Swahili unless {lang_name} is Swahili. " | |
| f"Do NOT respond in English. " | |
| f"Every single response, sentence, and word MUST be natively in {lang_name}. " | |
| f"Never switch to Swahili or any other language unless the user explicitly speaks to you in that language." | |
| ) | |
| # Dialect-specific fillers | |
| if lang_name.lower() == "kiswahili": | |
| fillers = "('Ah', 'Sawa', 'Eshi', 'Haya')" | |
| elif lang_name.lower() == "zulu": | |
| fillers = "('Yebo', 'Cha', 'Hawu', 'Mm')" | |
| elif lang_name.lower() == "arabic": | |
| fillers = "('نعم', 'أهلاً', 'حسناً', 'طبعاً')" | |
| elif lang_name.lower() == "kikuyu": | |
| fillers = "('Ĩĩ', 'Aca', 'Atĩrĩ', 'Kaĩ')" | |
| elif lang_name.lower() == "luo": | |
| fillers = "('Ee', 'Kare', 'Donge', 'Adier')" | |
| elif lang_name.lower() == "samburu": | |
| fillers = "('Ee', 'Acha', 'Supa')" | |
| elif lang_name.lower() == "kinyarwanda": | |
| fillers = "('Yego', 'Oya', 'Ese', 'Nibyo')" | |
| elif lang_name.lower() == "hausa": | |
| fillers = "('To', 'Ai', 'Wato', 'Da kyau')" | |
| elif lang_name.lower() == "somali": | |
| fillers = "('Haa', 'Maya', 'Waa hagaag', 'Sax')" | |
| elif lang_name.lower() == "oromo": | |
| fillers = "('Eeyyee', 'Lakki', 'Tole', 'Dhugaa')" | |
| elif lang_name.lower() == "luganda": | |
| fillers = "('Yee', 'Nedda', 'Kituufu', 'Mmm')" | |
| else: | |
| fillers = "natural local conversational pause markers" | |
| persona_name = "DUOL" | |
| if info.get("label") == "Kiswahili" or info.get("label") == "Kiswahili Fiti": | |
| persona_name = "Fiti" | |
| elif info.get("label") == "Kiswahili Daudi": | |
| persona_name = "Daudi" | |
| elif info.get("label") == "Kiswahili Rafiki": | |
| persona_name = "Rafiki" | |
| elif info.get("label") == "Kiswahili Rehema": | |
| persona_name = "Rehema" | |
| code_switch_rule = "" | |
| if info.get("voice_id") in ["sw-KE-RafikiNeural", "sw-TZ-RehemaNeural"] and spoken_language != "english": | |
| code_switch_rule = "\n- Must mix natural local Swahili/Sheng phrases with English in the first 10 words." | |
| return f"""You are {persona_name}, Africa's most intelligent AI voice assistant. | |
| Role: Deliver an emotionally intelligent, voice-optimized, real-time conversational agent capable of sub-500ms multi-lingual interaction. | |
| Language Context: {lang_prompt} | |
| Target Language: {lang_name} | |
| Target Region: {region} | |
| Task: | |
| 1. Speak natively and fluently in {lang_name}. If {lang_name} is NOT Swahili, NEVER speak Swahili. | |
| 2. Process real-time streaming speech transcripts (including disfluencies and barge-ins). | |
| 3. Apply Linguistic Mirroring to match user dialect, language ratios, and structural syntax in {lang_name}. | |
| 4. CRITICAL: You MUST use the `search_web` or `get_news` tools (powered by Parallel AI) for ANY query about current events, politics, facts, trends, or anything you are not 100% absolutely certain about. Never guess or hallucinate facts. | |
| 5. Execute other proactive tool calls (weather, local time, math, commodity prices) when relevant. | |
| 6. Format speech output strictly for low-latency TTS engines using prosodic punctuation and word-count boundaries. | |
| Output Rules: | |
| - Audio-first text streams optimized for TTS conversion (zero markdown, zero URLs, no special characters). | |
| - Hard cap of 25 words per utterance to yield turns rapidly. | |
| - Commas and em-dashes inserted before language switches to create natural TTS breath pauses. | |
| - Strict preservation of cultural markers and local honorifics for {lang_name}. | |
| Tone & Personality: | |
| - Warm, knowledgeable older sibling, street-smart, empathetic, and culturally proud. | |
| - Audience: Youth, farmers, traders, and rural/urban citizens. | |
| Guardrails (MUST AVOID): | |
| - The names 'Jarvis' or 'Habari AI'. Your name is ONLY {persona_name}. | |
| - Phrases like 'As an AI' or references to LLMs, OpenAI, or GPT. | |
| - Markdown formatting, bullet points, asterisks, symbols, URLs, or written-text shortcuts. | |
| - Speaking Kiswahili when the chosen voice is Kinyarwanda, Zulu, Arabic, Yoruba, Hausa, or any other non-Swahili language. | |
| Guardrails (MUST INCLUDE): | |
| - Always introduce and identify yourself ONLY as {persona_name}. | |
| - Native conversational markers {fillers}. | |
| - Proactive disclaimers for health, finance, and legal queries.{code_switch_rule} | |
| Domain Focus: | |
| {domain_txt} | |
| Verification before responding: | |
| - Is utterance under 25 words? | |
| - Is the response strictly in {lang_name}? | |
| - Was 2026 context retrieved via tools when needed? | |
| """ | |
| # ────────────────────────────────────────────────────────────────────────────── | |
| # DUOL Agent with Function Tools | |
| # ────────────────────────────────────────────────────────────────────────────── | |
| class JarvisAgent(Agent): | |
| def __init__(self, voice_code: str, spoken_language: str, domain: str = "general"): | |
| info = VOICE_INFO.get(voice_code, VOICE_INFO["swh"]) | |
| instructions = build_instructions(info, spoken_language, domain) | |
| super().__init__(instructions=instructions) | |
| # ── Tool 1: Weather ──────────────────────────────────────────────────────── | |
| async def get_weather(self, city: str) -> str: | |
| """ | |
| Get current weather for any city in the world. | |
| Args: | |
| city: City name (e.g. Dodoma, Nairobi, London) | |
| """ | |
| import httpx | |
| try: | |
| async with httpx.AsyncClient(timeout=10.0) as client: | |
| geo_resp = await client.get( | |
| "https://geocoding-api.open-meteo.com/v1/search", | |
| params={"name": city, "count": 1, "language": "en", "format": "json"} | |
| ) | |
| geo_data = geo_resp.json() | |
| if not geo_data.get("results"): | |
| return f"Could not find geographic coordinates for {city}." | |
| loc = geo_data["results"][0] | |
| lat, lon, tz = loc["latitude"], loc["longitude"], loc.get("timezone", "UTC") | |
| name, country = loc.get("name", city), loc.get("country", "") | |
| weather_resp = await client.get( | |
| "https://api.open-meteo.com/v1/forecast", | |
| params={ | |
| "latitude": lat, "longitude": lon, | |
| "current": "temperature_2m,weathercode,windspeed_10m,relative_humidity_2m", | |
| "timezone": tz, | |
| } | |
| ) | |
| data = weather_resp.json() | |
| cur = data["current"] | |
| temp = cur["temperature_2m"] | |
| hum = cur["relative_humidity_2m"] | |
| wind = cur["windspeed_10m"] | |
| wcode = cur["weathercode"] | |
| WEATHER_CODES = { | |
| 0: "clear sky", 1: "mainly clear", 2: "partly cloudy", 3: "overcast", | |
| 45: "foggy", 48: "icy fog", 51: "light drizzle", 53: "drizzle", | |
| 55: "heavy drizzle", 61: "light rain", 63: "moderate rain", 65: "heavy rain", | |
| 71: "light snow", 73: "snow", 75: "heavy snow", | |
| 80: "rain showers", 81: "heavy rain showers", 82: "violent rain showers", | |
| 95: "thunderstorm", 96: "thunderstorm with hail", 99: "severe thunderstorm", | |
| } | |
| desc = WEATHER_CODES.get(wcode, "unknown conditions") | |
| return f"Weather in {name}, {country}: {temp}°C, {desc}. Humidity: {hum}%, Wind: {wind} km/h." | |
| except Exception as e: | |
| return f"Weather service unavailable for {city}." | |
| # ── Tool 2: Local Time ───────────────────────────────────────────────────── | |
| async def get_local_time(self, city: str) -> str: | |
| """ | |
| Get the current local time and date for any city in the world. | |
| Args: | |
| city: City name (e.g. Dodoma, Tokyo, Nairobi) | |
| """ | |
| import httpx | |
| from datetime import datetime | |
| try: | |
| import zoneinfo | |
| async with httpx.AsyncClient(timeout=10.0) as client: | |
| geo_resp = await client.get( | |
| "https://geocoding-api.open-meteo.com/v1/search", | |
| params={"name": city, "count": 1, "language": "en", "format": "json"} | |
| ) | |
| geo_data = geo_resp.json() | |
| if not geo_data.get("results"): | |
| return f"Could not find timezone for {city}." | |
| loc = geo_data["results"][0] | |
| tz_name = loc.get("timezone", "UTC") | |
| name, country = loc.get("name", city), loc.get("country", "") | |
| tz = zoneinfo.ZoneInfo(tz_name) | |
| now = datetime.now(tz) | |
| time_str = now.strftime("%I:%M %p").lstrip('0') | |
| date_str = now.strftime("%A, %B %d, %Y") | |
| return f"The current local time in {name}, {country} is {time_str} on {date_str}." | |
| except Exception: | |
| return f"Time service unavailable for {city}." | |
| # ── Tool 3: Calculator ───────────────────────────────────────────────────── | |
| async def calculate(self, expression: str) -> str: | |
| """ | |
| Calculate a mathematical expression safely. | |
| Args: | |
| expression: A math expression like '15 * 3 + 7' or '1000 / 12' | |
| """ | |
| safe = _re.sub(r'[^0-9+\-*/().%\s]', '', expression) | |
| if not safe.strip(): | |
| return "Please provide a valid mathematical expression." | |
| try: | |
| result = eval(safe, {"__builtins__": {}}, {}) # nosec | |
| if isinstance(result, float): | |
| result = round(result, 6) | |
| return f"{expression} equals {result}." | |
| except Exception: | |
| return "I couldn't calculate that expression. Please try rephrasing it." | |
| # ── Tool 4: News ─────────────────────────────────────────────────────────── | |
| async def get_news(self, topic: str = "africa") -> str: | |
| """ | |
| Get the latest news headlines relevant to a specific topic or region. | |
| Args: | |
| topic: Topic to search for (e.g. 'Kenya politics', 'Nigeria news', 'Tanzania sports') | |
| """ | |
| import os | |
| import httpx | |
| import xml.etree.ElementTree as ET | |
| # 1. Try Parallel AI first for highly accurate news | |
| api_key = os.environ.get("PARALLEL_API_KEY") | |
| if api_key: | |
| try: | |
| from parallel import AsyncParallel | |
| client = AsyncParallel(api_key=api_key) | |
| search = await client.search( | |
| objective=f"Find the most recent breaking news and headlines about: {topic}", | |
| search_queries=[f"latest news {topic}", f"breaking news {topic} today"] | |
| ) | |
| if search.results: | |
| excerpts = [] | |
| for res in search.results[:3]: | |
| if hasattr(res, 'title') and res.title: | |
| excerpt = res.excerpts[0] if hasattr(res, 'excerpts') and res.excerpts else '' | |
| excerpts.append(f"{res.title}: {excerpt}") | |
| if excerpts: | |
| return f"Latest news on '{topic}': {' | '.join(excerpts)}" | |
| except Exception as e: | |
| print(f"[News] Parallel AI failed: {e}") | |
| # 2. Fallback to generic BBC Africa RSS | |
| try: | |
| feed_url = "https://feeds.bbci.co.uk/news/world/africa/rss.xml" | |
| async with httpx.AsyncClient(timeout=10.0) as client: | |
| resp = await client.get(feed_url, headers={"User-Agent": "HabariJarvis/1.0"}) | |
| root = ET.fromstring(resp.text) | |
| items = root.findall(".//item") | |
| headlines = [] | |
| for item in items[:3]: | |
| title_el = item.find("title") | |
| if title_el is not None and title_el.text: | |
| headlines.append(title_el.text.strip()) | |
| if not headlines: | |
| return "No recent headlines available right now." | |
| return f"Latest general Africa headlines: {'. Next: '.join(headlines)}." | |
| except Exception as e: | |
| return f"News service is temporarily unavailable." | |
| # ── Tool 5: Commodity Prices ─────────────────────────────────────────────── | |
| async def get_commodity_price(self, commodity: str) -> str: | |
| """ | |
| Get approximate commodity prices relevant to African farmers and traders. | |
| Args: | |
| commodity: Commodity name (e.g. 'maize', 'tea', 'coffee', 'sugar', 'rice', 'gold', 'oil') | |
| """ | |
| PRICES = { | |
| "maize": ("USD 180 to 220 per metric tonne", "Used for food and animal feed"), | |
| "corn": ("USD 180 to 220 per metric tonne", "Same as maize"), | |
| "coffee": ("USD 2.00 to 2.50 per pound", "Arabica, traded on ICE"), | |
| "tea": ("USD 2.50 to 3.50 per kilogram", "Mombasa auction average"), | |
| "sugar": ("USD 22 to 26 per hundredweight", "Raw sugar on commodity markets"), | |
| "rice": ("USD 500 to 600 per metric tonne", "Thai 5 percent broken benchmark"), | |
| "wheat": ("USD 220 to 260 per metric tonne", "CBOT wheat futures"), | |
| "sorghum": ("USD 180 to 220 per metric tonne", "Similar to maize"), | |
| "gold": ("USD 2300 to 2500 per troy ounce", "COMEX gold spot price"), | |
| "oil": ("USD 75 to 90 per barrel", "Brent crude benchmark"), | |
| "copper": ("USD 9000 to 10000 per metric tonne","LME copper"), | |
| "cocoa": ("USD 7000 to 9000 per metric tonne", "ICE cocoa futures"), | |
| } | |
| key = commodity.lower().strip() | |
| for name, (price, note) in PRICES.items(): | |
| if key in name or name in key: | |
| return (f"{commodity.title()} price: approximately {price}. " | |
| f"Note: {note}. Always verify with your local market.") | |
| return (f"I don't have price data for {commodity}. " | |
| f"Check with your local commodity exchange or market board.") | |
| # ── Tool 6: Web Search ───────────────────────────────────────────────────── | |
| async def search_web(self, query: str) -> str: | |
| """ | |
| Search the web for general knowledge, locations, universities, and up-to-date facts. | |
| Args: | |
| query: The search query (e.g. 'Where is Multimedia University Kenya') | |
| """ | |
| import os | |
| import asyncio | |
| import requests | |
| # 1. Parallel AI Search (Primary) | |
| api_key = os.environ.get("PARALLEL_API_KEY") | |
| if api_key: | |
| try: | |
| from parallel import AsyncParallel | |
| client = AsyncParallel(api_key=api_key) | |
| search = await client.search( | |
| objective=f"Find concise information answering: {query}", | |
| search_queries=[query] | |
| ) | |
| if search.results: | |
| excerpts = [] | |
| for res in search.results[:2]: | |
| if hasattr(res, 'excerpts') and res.excerpts: | |
| excerpts.append(str(res.excerpts[0])) | |
| if excerpts: | |
| return f"Search results for '{query}': {' '.join(excerpts)}" | |
| except Exception as e: | |
| print(f"[Search] Parallel AI failed: {e}") | |
| # 2. Reliable Fallback: Wikipedia API | |
| def do_fallback(): | |
| try: | |
| search_url = f"https://en.wikipedia.org/w/api.php?action=opensearch&search={query}&limit=1&namespace=0&format=json" | |
| res = requests.get(search_url).json() | |
| if len(res) > 1 and res[1]: | |
| title = res[1][0] | |
| summary_url = f"https://en.wikipedia.org/w/api.php?format=json&action=query&prop=extracts&exintro&explaintext&redirects=1&titles={title}" | |
| summary_res = requests.get(summary_url).json() | |
| pages = summary_res.get("query", {}).get("pages", {}) | |
| for page_id in pages: | |
| return pages[page_id].get("extract", "")[:1000] | |
| except: | |
| pass | |
| return "" | |
| try: | |
| summary = await asyncio.to_thread(do_fallback) | |
| if not summary: | |
| return f"I couldn't find any information on the web for '{query}'." | |
| return f"Search results for '{query}': {summary}" | |
| except Exception as e: | |
| return f"The web search service is currently unavailable." | |
| # ────────────────────────────────────────────────────────────────────────────── | |
| # Entry Point | |
| # ────────────────────────────────────────────────────────────────────────────── | |
| async def entrypoint(ctx: JobContext): | |
| await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY) | |
| # Metadata format: "voice_code:mode:domain" | |
| # e.g. "swh:local:health" or "yor_premium:english:general" | |
| metadata = ctx.job.metadata or "swh:local:general" | |
| parts = metadata.split(":") | |
| voice_code = parts[0] if len(parts) > 0 and parts[0] in VOICE_INFO else "swh" | |
| spoken_lang = parts[1] if len(parts) > 1 and parts[1] in ["local", "english"] else "local" | |
| domain = parts[2] if len(parts) > 2 and parts[2] in DOMAIN_CONTEXT else "general" | |
| info = VOICE_INFO[voice_code] | |
| print(f"[Jarvis] Voice: {voice_code} ({info['label']}) | Mode: {spoken_lang} | Domain: {domain}") | |
| # ── Select TTS engine ── | |
| if info["type"] == "api": | |
| print(f"[Jarvis] Using distributed API TTS -> {info['space']}") | |
| tts_engine = ApiTTS(space_id=info["space"], **info.get("api_kwargs", {})) | |
| elif info["type"] == "edge": | |
| print(f"[Jarvis] Using Edge TTS -> {info['voice_id']}") | |
| tts_engine = EdgeTTS(voice_id=info["voice_id"]) | |
| else: | |
| print(f"[Jarvis] Using local MMS-TTS -> {info['repo']}") | |
| tts_engine = MmsTTS(model_id=info["repo"]) | |
| # ── Select STT language dynamically ── | |
| whisper_lang = info.get("whisper_lang", "sw") | |
| print(f"[Jarvis] STT language: {whisper_lang}") | |
| # ── Build greeting ── | |
| if spoken_lang == "english": | |
| greeting = (f"Hello! I am DUOL, your AI voice assistant. " | |
| f"How can I help you today?") | |
| else: | |
| greeting = info["greeting"] | |
| # Optimized VAD for sub-400ms cutoff latency | |
| # If the language is Swahili, we let Whisper auto-detect to handle Sheng/English code-switching accurately | |
| session = AgentSession( | |
| vad=silero.VAD.load(min_silence_duration=0.3, min_speech_duration=0.05), | |
| stt=groq.STT(model="whisper-large-v3", language=whisper_lang), | |
| llm=openai.LLM(model="gpt-4o-mini"), | |
| tts=tts_engine, | |
| ) | |
| audio_frames = [] | |
| audio_sample_rate = 48000 | |
| audio_channels = 1 | |
| recording_active = False | |
| async def record_participant_audio(participant: rtc.RemoteParticipant): | |
| nonlocal audio_frames, audio_sample_rate, audio_channels, recording_active | |
| # Wait for the audio track to be published | |
| for _ in range(30): # wait up to 30 seconds | |
| for pub in participant.track_publications.values(): | |
| if pub.kind == rtc.TrackKind.KIND_AUDIO and pub.track: | |
| track = pub.track | |
| print(f"[R2 Logger] Found audio track for {participant.identity}") | |
| recording_active = True | |
| stream = rtc.AudioStream(track) | |
| try: | |
| async for event in stream: | |
| if event.frame: | |
| frame = event.frame | |
| audio_sample_rate = frame.sample_rate | |
| audio_channels = frame.num_channels | |
| audio_frames.append(frame.data.tobytes()) | |
| except Exception as e: | |
| print(f"[R2 Logger] Stream ended: {e}") | |
| finally: | |
| recording_active = False | |
| return | |
| await asyncio.sleep(1) | |
| print("[R2 Logger] No audio track found for participant") | |
| async def save_and_upload_audio(session_id: str, identity: str): | |
| nonlocal audio_frames | |
| if not audio_frames: | |
| print("[R2 Logger] No audio frames captured, skipping upload.") | |
| return | |
| import tempfile | |
| temp_dir = tempfile.gettempdir() | |
| audio_file = os.path.join(temp_dir, f"{session_id}_{identity}_{int(_time.time())}.wav") | |
| try: | |
| with wave.open(audio_file, "wb") as wf: | |
| wf.setnchannels(audio_channels) | |
| wf.setsampwidth(2) | |
| wf.setframerate(audio_sample_rate) | |
| for frame_data in audio_frames: | |
| wf.writeframes(frame_data) | |
| total_seconds = len(audio_frames) * 480 / audio_sample_rate # rough estimate | |
| print(f"[R2 Logger] Saved {len(audio_frames)} frames (~{total_seconds:.0f}s) to {audio_file}") | |
| await upload_to_r2(audio_file, f"voice_data/{session_id}/{os.path.basename(audio_file)}") | |
| except Exception as e: | |
| print(f"[R2 Logger Error] Save failed: {e}") | |
| finally: | |
| audio_frames = [] | |
| async def upload_to_r2(file_path: str, object_name: str): | |
| endpoint_url = os.environ.get("CLOUDFLARE_R2_ENDPOINT_URL", "https://aad8747f9b4aaa2c4a766decd6796228.r2.cloudflarestorage.com") | |
| access_key = os.environ.get("CLOUDFLARE_R2_ACCESS_KEY_ID", "f3ae0ad268f28c501ef800ac80b2409f") | |
| secret_key = os.environ.get("CLOUDFLARE_R2_SECRET_ACCESS_KEY", "981cc70538ad6b0b112deebc8e229250b45a994aa92c7e1138072831a0517b3a") | |
| bucket_name = os.environ.get("CLOUDFLARE_R2_BUCKET", "duol-audio") | |
| if not endpoint_url or not access_key: | |
| print("[R2 Logger Warning] R2 credentials not set. Skipping upload.") | |
| return | |
| def _upload(): | |
| try: | |
| s3 = boto3.client( | |
| "s3", | |
| endpoint_url=endpoint_url, | |
| aws_access_key_id=access_key, | |
| aws_secret_access_key=secret_key, | |
| config=Config(signature_version="s3v4"), | |
| region_name="auto" | |
| ) | |
| s3.upload_file(file_path, bucket_name, object_name) | |
| print(f"[R2 Logger] Uploaded to R2: {object_name}") | |
| os.remove(file_path) | |
| except Exception as e: | |
| print(f"[R2 Logger Error] Upload failed: {e}") | |
| await asyncio.to_thread(_upload) | |
| def on_participant_connected(participant: rtc.RemoteParticipant): | |
| print(f"[R2 Logger] Participant connected: {participant.identity}") | |
| asyncio.create_task(record_participant_audio(participant)) | |
| def on_participant_disconnected(participant: rtc.RemoteParticipant): | |
| print(f"[R2 Logger] Participant disconnected: {participant.identity}. Saving audio...") | |
| asyncio.create_task(save_and_upload_audio(ctx.room.name, participant.identity)) | |
| # Also handle participants already in the room | |
| for participant in ctx.room.remote_participants.values(): | |
| print(f"[R2 Logger] Existing participant: {participant.identity}") | |
| asyncio.create_task(record_participant_audio(participant)) | |
| # --------------------------------------------- | |
| # --- SUPABASE LOGGING LOGIC --- | |
| last_user_input = None | |
| def on_conversation_item_added(ev): | |
| nonlocal last_user_input | |
| try: | |
| item = ev.item | |
| role = getattr(item, "role", None) | |
| if role == "user": | |
| last_user_input = getattr(item, "content", getattr(item, "text", str(item))) | |
| if isinstance(last_user_input, list): | |
| last_user_input = " ".join([str(c) for c in last_user_input]) | |
| print(f"[Supabase Log] Captured user input: {last_user_input}") | |
| elif role == "assistant": | |
| if last_user_input: | |
| agent_text = getattr(item, "content", getattr(item, "text", str(item))) | |
| if isinstance(agent_text, list): | |
| agent_text = " ".join([str(c) for c in agent_text]) | |
| asyncio.create_task( | |
| _log_to_supabase( | |
| session_id=ctx.room.name, | |
| user_input=last_user_input, | |
| duol_response=agent_text, | |
| model_used=voice_code | |
| ) | |
| ) | |
| last_user_input = None | |
| except Exception as e: | |
| print(f"[Supabase Log Error in event hook] {e}") | |
| async def _log_to_supabase(session_id, user_input, duol_response, model_used): | |
| supabase_url = os.environ.get("NEXT_PUBLIC_SUPABASE_URL", "https://jxqsacpmjjvvmeocbbqi.supabase.co") | |
| supabase_key = os.environ.get("SUPABASE_SERVICE_ROLE_KEY", "sb_secret_4gODEyeWHrRta1Eoox1Wfg_eSLhQ1iG") | |
| url = f"{supabase_url}/rest/v1/conversations" | |
| headers = { | |
| "apikey": supabase_key, | |
| "Authorization": f"Bearer {supabase_key}", | |
| "Content-Type": "application/json", | |
| "Prefer": "return=minimal" | |
| } | |
| payload = { | |
| "session_id": session_id, | |
| "user_input": user_input, | |
| "duol_response": duol_response, | |
| "model_used": model_used | |
| } | |
| try: | |
| async with httpx.AsyncClient(timeout=10.0) as client: | |
| await client.post(url, json=payload, headers=headers) | |
| print("[Supabase Log] Successfully saved conversation turn to database.") | |
| except Exception as e: | |
| print(f"[Supabase Log Error] {e}") | |
| # ------------------------------ | |
| await session.start( | |
| room=ctx.room, | |
| agent=JarvisAgent(voice_code=voice_code, spoken_language=spoken_lang, domain=domain), | |
| room_input_options=RoomInputOptions(), | |
| ) | |
| await session.generate_reply( | |
| instructions=f"Greet the user using exactly this phrase: {greeting}" | |
| ) | |
| async def request_fnc(req: JobRequest): | |
| await req.accept(entrypoint) | |
| if __name__ == "__main__": | |
| import threading | |
| from http.server import BaseHTTPRequestHandler, HTTPServer | |
| class HealthHandler(BaseHTTPRequestHandler): | |
| def do_GET(self): | |
| self.send_response(200) | |
| self.end_headers() | |
| self.wfile.write(b"OK") | |
| def log_message(self, *_): | |
| pass | |
| def run_dummy_server(): | |
| server = HTTPServer(('0.0.0.0', 7860), HealthHandler) | |
| server.serve_forever() | |
| threading.Thread(target=run_dummy_server, daemon=True).start() | |
| cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, request_fnc=request_fnc, agent_name="DUOL")) | |