""" LiveKit Voice Agent -- Multi-Agent Architecture for Fluente.ai Deployed to LiveKit Cloud via `lk agent deploy`. Agent name: Inworld Architecture (per Master Spec, February 2026): 3 Agents: SalesConciergeAgent -> AssessmentAgent -> TutorAgent Pipeline: Deepgram Nova-3 STT -> Gemini 2.5 Flash LLM -> Inworld TTS 1.5 State: FluenteUserData dataclass shared across agents via session.userdata Storage: Supabase (async) for leads, profiles, corrections, vocab, sessions Handoffs: Tool returns new Agent instance -> LiveKit swaps instructions + TTS voice Plugin Stack: - LLM (Primary): livekit-plugins-google (gemini-2.5-flash) - LLM (Fallback): livekit-plugins-openai (gpt-4.1-mini) - STT: livekit-plugins-deepgram (nova-3, language="multi") - TTS: livekit-plugins-inworld (tts-1.5-max) - VAD: livekit-plugins-silero (local, no API call) - Turn Detection: livekit-plugins-turn-detector (semantic, local) - Noise Cancel: Built-in BVC (background voice cancellation) """ import asyncio import json import logging import os from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Optional from livekit import rtc from livekit.agents import ( Agent, AgentSession, ConversationItemAddedEvent, ErrorEvent, JobContext, RoomInputOptions, RunContext, UserInputTranscribedEvent, WorkerOptions, cli, function_tool, llm, ) from livekit.plugins import deepgram, google, inworld, openai, silero try: from livekit.plugins import noise_cancellation except ImportError: noise_cancellation = None # type: ignore[assignment] logging.getLogger(__name__).warning( "noise_cancellation plugin not installed — running without BVC" ) logging.basicConfig(level=logging.INFO) logger = logging.getLogger("fluente-agent") # ====================================================================== # 1. CONFIGURATION # ====================================================================== SUPABASE_URL = os.environ.get("SUPABASE_URL", "") SUPABASE_SERVICE_KEY = os.environ.get("SUPABASE_SERVICE_KEY", "") INWORLD_API_KEY = os.environ.get("INWORLD_API_KEY", "") # Inworld TTS model INWORLD_TTS_MODEL = os.environ.get("INWORLD_TTS_MODEL", "inworld-tts-1.5-max") # Default Inworld voice (env override, used as final catch-all) INWORLD_DEFAULT_VOICE = os.environ.get("INWORLD_DEFAULT_VOICE", "fluente__sarah_us") # All 48 custom Fluente voices cloned on Inworld, grouped by language/region. INWORLD_VOICES: dict[str, list[str]] = { "en": ["fluente__sarah_us", "fluente__david_us"], "en-gb": ["fluente__emma_gb", "fluente__oliver_gb"], "en-au": ["fluente__mia_au", "fluente__jack_au"], "en-nz": ["fluente__aroha_nz", "fluente__tane_nz"], "en-ie": ["fluente__aoife_ie", "fluente__cillian_ie"], "en-scot": ["fluente__eilidh_scot", "fluente__callum_scot"], "en-za": ["fluente__naledi_za", "fluente__thabo_za"], "es": ["fluente__elena_es", "fluente__alejandro_es", "fluente__isabella_es", "fluente__mateo_es"], "fr": ["fluente__chloe_fr", "fluente__julien_fr"], "fr-ca": ["fluente__megane_fr_ca", "fluente__etienne_fr_ca"], "de": ["fluente__hannah_de", "fluente__lukas_de"], "it": ["fluente__giulia_it", "fluente__marco_it"], "pt": ["fluente__sofia_pt", "fluente__joao_pt"], "pt-br": ["fluente__beatriz_br", "fluente__gabriel_br"], "ja": ["fluente__yuki_ja", "fluente__kenji_ja"], "ko": ["fluente__jiwon_ko", "fluente__minho_ko"], "zh": ["fluente__li_zh-cn", "fluente__wang_zh-cn"], "ar": ["fluente__fatimah_ar", "fluente__omar_ar"], "hi": ["fluente__ananya_hi", "fluente__rahul_hi"], "ru": ["fluente__irina_ru", "fluente__pavel_ru"], "nl": ["fluente__sophie_nl", "fluente__bram_nl"], "pl": ["fluente__paulina_pl", "fluente__adam_pl"], "he": ["fluente__noe_il", "fluente__avi_il"], } INWORLD_DEFAULT_VOICES: dict[str, str] = { lang: voices[0] for lang, voices in INWORLD_VOICES.items() if voices } # --------------------------------------------------------------------------- # LLM fallback chain (per Prompts v2 §1 & Pricing doc) # # Uses LiveKit's built-in llm.FallbackAdapter which: # - Periodically health-checks the primary provider in the background # - Marks failed providers as unhealthy and stops sending requests # - Automatically resubmits the failed request to backup providers # - Re-enables the primary provider once it recovers # # Priority (OpenAI-first — proven stable; Gemini as optional upgrade): # 1. GPT-4.1 mini (OPENAI_API_KEY) — primary, strong instruction following # 2. Gemini 2.5 Flash (GOOGLE_API_KEY) — fallback, best multilingual # 3. GPT-4.1 nano (OPENAI_API_KEY) — emergency / cheapest # --------------------------------------------------------------------------- # Accept either GOOGLE_API_KEY or GEMINI_API_KEY (common alias). # If GOOGLE_API_KEY is missing but GEMINI_API_KEY is set, copy it over so # that the Google LLM plugin (which reads GOOGLE_API_KEY) works. if not os.environ.get("GOOGLE_API_KEY") and os.environ.get("GEMINI_API_KEY"): os.environ["GOOGLE_API_KEY"] = os.environ["GEMINI_API_KEY"] logger.info("Copied GEMINI_API_KEY -> GOOGLE_API_KEY") _HAS_GOOGLE_KEY = bool(os.environ.get("GOOGLE_API_KEY")) _HAS_OPENAI_KEY = bool(os.environ.get("OPENAI_API_KEY")) if not _HAS_GOOGLE_KEY: logger.warning( "GOOGLE_API_KEY / GEMINI_API_KEY not set — will use OpenAI for LLM" ) if not _HAS_OPENAI_KEY: logger.warning( "OPENAI_API_KEY not set — OpenAI fallback unavailable" ) def _make_llm(*, thinking: bool = False): """Create the best available LLM with automatic runtime failover. Uses llm.FallbackAdapter (per LiveKit docs) so that if the primary provider fails mid-session, the request is automatically resubmitted to the next provider in the chain without crashing. Fallback chain: - Default: GPT-4.1 mini (primary) -> Gemini 2.5 Flash -> GPT-4.1 nano - Thinking: Gemini 2.5 Flash (primary, with thinking_config) -> GPT-4.1 mini -> GPT-4.1 nano """ providers: list = [] if thinking and _HAS_GOOGLE_KEY: # When thinking mode is requested (e.g. AssessmentAgent), Gemini # must be primary because only Gemini supports thinking_config. kwargs: dict = {} try: from google.genai import types as _gtypes kwargs["thinking_config"] = _gtypes.ThinkingConfig( thinking_budget=2048, ) except Exception: logger.warning("Could not enable thinking mode — skipping") providers.append(google.LLM(model="gemini-2.5-flash", **kwargs)) # OpenAI as fallback for thinking path if _HAS_OPENAI_KEY: providers.append(openai.LLM(model="gpt-4.1-mini")) providers.append(openai.LLM(model="gpt-4.1-nano")) else: # Default path: OpenAI first — proven stable in production if _HAS_OPENAI_KEY: providers.append(openai.LLM(model="gpt-4.1-mini")) # Gemini as fallback (or primary when OpenAI unavailable) if _HAS_GOOGLE_KEY: providers.append(google.LLM(model="gemini-2.5-flash")) # Emergency fallback — cheapest OpenAI model if _HAS_OPENAI_KEY: providers.append(openai.LLM(model="gpt-4.1-nano")) if not providers: raise RuntimeError( "No LLM API key configured. Set GOOGLE_API_KEY or OPENAI_API_KEY." ) if len(providers) == 1: # Only one provider available — return it directly (no adapter needed) logger.info("Using LLM: %s (no fallback available)", providers[0]) return providers[0] logger.info( "LLM FallbackAdapter: %s", " -> ".join(str(p) for p in providers), ) return llm.FallbackAdapter(providers) LANGUAGE_NAMES: dict[str, str] = { "en": "English", "es": "Spanish", "fr": "French", "de": "German", "it": "Italian", "pt": "Portuguese", "ja": "Japanese", "ko": "Korean", "zh": "Chinese", "ar": "Arabic", "hi": "Hindi", "ru": "Russian", "sv": "Swedish", "nl": "Dutch", "da": "Danish", "no": "Norwegian", "el": "Greek", "pl": "Polish", "tr": "Turkish", "fi": "Finnish", "he": "Hebrew", "ms": "Malay", "sw": "Swahili", } # Teacher persona database TEACHER_DB: dict[str, dict] = { "es": { "name": "Sofia", "origin": "Madrid, Spain", "language": "es-ES", "tts_voice_id": "fluente__elena_es", "accent_profile": "Castilian Spanish, clear distincion", "personality": "Warm, encouraging, slightly formal, uses humour", "dialect_notes": "Uses vosotros, distinguishes c/z from s", }, "es-mx": { "name": "Carlos", "origin": "Mexico City, Mexico", "language": "es-MX", "tts_voice_id": "fluente__alejandro_es", "accent_profile": "Mexican Spanish, clear and neutral", "personality": "Friendly, patient, uses colloquial expressions", "dialect_notes": "Uses ustedes, seseo, Mexican vocabulary", }, "fr": { "name": "Chloe", "origin": "Paris, France", "language": "fr-FR", "tts_voice_id": "fluente__chloe_fr", "accent_profile": "Parisian French, elegant pronunciation", "personality": "Sophisticated, encouraging, uses cultural references", "dialect_notes": "Metropolitan French, liaison emphasis", }, "de": { "name": "Hannah", "origin": "Berlin, Germany", "language": "de-DE", "tts_voice_id": "fluente__hannah_de", "accent_profile": "Standard German, clear Hochdeutsch", "personality": "Precise, warm, structured teaching style", "dialect_notes": "Standard German, no regional dialect", }, "it": { "name": "Giulia", "origin": "Rome, Italy", "language": "it-IT", "tts_voice_id": "fluente__giulia_it", "accent_profile": "Roman Italian, expressive intonation", "personality": "Passionate, animated, uses gestures in speech", "dialect_notes": "Standard Italian with Roman touches", }, "ja": { "name": "Yuki", "origin": "Tokyo, Japan", "language": "ja-JP", "tts_voice_id": "fluente__yuki_ja", "accent_profile": "Standard Japanese, NHK-style pronunciation", "personality": "Polite, patient, encouraging, structured", "dialect_notes": "Standard Tokyo dialect, keigo teaching", }, "pt": { "name": "Sofia", "origin": "Lisbon, Portugal", "language": "pt-PT", "tts_voice_id": "fluente__sofia_pt", "accent_profile": "European Portuguese, clear diction", "personality": "Warm, patient, culturally aware", "dialect_notes": "European Portuguese pronunciation", }, "pt-br": { "name": "Beatriz", "origin": "Sao Paulo, Brazil", "language": "pt-BR", "tts_voice_id": "fluente__beatriz_br", "accent_profile": "Brazilian Portuguese, open vowels", "personality": "Energetic, friendly, musical speech", "dialect_notes": "Brazilian Portuguese with paulista touches", }, "ko": { "name": "Jiwon", "origin": "Seoul, South Korea", "language": "ko-KR", "tts_voice_id": "fluente__jiwon_ko", "accent_profile": "Standard Korean, Seoul dialect", "personality": "Friendly, structured, encouraging", "dialect_notes": "Standard Seoul Korean", }, "zh": { "name": "Li", "origin": "Beijing, China", "language": "zh-CN", "tts_voice_id": "fluente__li_zh-cn", "accent_profile": "Standard Mandarin, clear tones", "personality": "Patient, precise, tone-focused", "dialect_notes": "Standard Mandarin Chinese", }, "ar": { "name": "Fatimah", "origin": "Cairo, Egypt", "language": "ar-SA", "tts_voice_id": "fluente__fatimah_ar", "accent_profile": "Modern Standard Arabic", "personality": "Warm, patient, culturally sensitive", "dialect_notes": "MSA with Egyptian touches", }, "hi": { "name": "Ananya", "origin": "Mumbai, India", "language": "hi-IN", "tts_voice_id": "fluente__ananya_hi", "accent_profile": "Standard Hindi, clear diction", "personality": "Friendly, patient, encouraging", "dialect_notes": "Standard Hindi", }, "ru": { "name": "Irina", "origin": "Moscow, Russia", "language": "ru-RU", "tts_voice_id": "fluente__irina_ru", "accent_profile": "Standard Russian, Moscow pronunciation", "personality": "Direct, warm, structured", "dialect_notes": "Standard Russian", }, "nl": { "name": "Sophie", "origin": "Amsterdam, Netherlands", "language": "nl-NL", "tts_voice_id": "fluente__sophie_nl", "accent_profile": "Standard Dutch, ABN", "personality": "Direct, friendly, practical", "dialect_notes": "Standard Dutch", }, "pl": { "name": "Paulina", "origin": "Warsaw, Poland", "language": "pl-PL", "tts_voice_id": "fluente__paulina_pl", "accent_profile": "Standard Polish, clear diction", "personality": "Warm, patient, encouraging", "dialect_notes": "Standard Polish", }, "en": { "name": "Sarah", "origin": "United States", "language": "en-US", "tts_voice_id": "fluente__sarah_us", "accent_profile": "General American English", "personality": "Warm, encouraging, natural", "dialect_notes": "General American", }, } # ====================================================================== # 2. SUPABASE CLIENT # ====================================================================== _supabase_client = None # Strong references for fire-and-forget tasks so they aren't GC'd mid-flight. # See https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task _background_tasks: set[asyncio.Task] = set() def _spawn(coro) -> asyncio.Task: """Create a background task with a strong reference and error logging.""" task = asyncio.create_task(coro) _background_tasks.add(task) task.add_done_callback(_background_tasks.discard) return task async def init_supabase(): """Initialise the global Supabase async client (once per process). The guard prevents leaking HTTP connection pools when multiple sessions re-enter entrypoint() within the same worker process. """ global _supabase_client if _supabase_client is not None: return # already initialised — reuse existing client if not SUPABASE_URL or not SUPABASE_SERVICE_KEY: logger.warning("Supabase not configured -- DB writes will be skipped") return try: from supabase import acreate_client _supabase_client = await acreate_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) logger.info("Supabase async client initialised") except ImportError: logger.warning("supabase package not installed -- DB writes will be skipped") except Exception as exc: logger.error(f"Failed to initialise Supabase: {exc}") def _supa(): """Get the Supabase client (or None if not configured).""" return _supabase_client # ====================================================================== # 3. SHARED STATE DATACLASS # ====================================================================== @dataclass class FluenteUserData: """Shared state across all agents in a session. Lives on session.userdata and is accessible from every agent and every tool via ctx.userdata. """ # Identity student_id: Optional[str] = None student_name: Optional[str] = None # Lead data (set by SalesConciergeAgent tools) target_language: Optional[str] = None source_language: Optional[str] = None motivation: Optional[str] = None self_assessed_level: Optional[str] = None specific_goal: Optional[str] = None daily_time_minutes: int = 0 deadline: Optional[str] = None recommended_tier: Optional[str] = None mini_demo_completed: bool = False # Assessment data (set by AssessmentAgent tools) assessed_level: Optional[str] = None listening_score: Optional[str] = None speaking_score: Optional[str] = None vocabulary_score: Optional[str] = None grammar_score: Optional[str] = None pronunciation_score: Optional[int] = None fluency_score: Optional[str] = None weak_areas: list = field(default_factory=list) strong_areas: list = field(default_factory=list) recommended_start_lesson: Optional[str] = None # Session context (loaded from Supabase at session start) current_lesson_id: Optional[str] = None current_cefr_level: Optional[str] = None subscription_tier: Optional[str] = None teacher_name: Optional[str] = None teacher_origin: Optional[str] = None teacher_language: Optional[str] = None correction_mode: Optional[str] = None session_type: Optional[str] = None lesson_vocabulary: list = field(default_factory=list) previous_vocabulary: list = field(default_factory=list) student_weak_areas: list = field(default_factory=list) # Live metrics (updated by tools during session) corrections_made: int = 0 vocab_items_practiced: list = field(default_factory=list) session_start: Optional[datetime] = None # ====================================================================== # 4. PROMPTS # ====================================================================== SALES_CONCIERGE_PROMPT = ( "You are a Fluente.ai Sales Concierge. Your goal is to understand the " "prospective student's needs, demonstrate value, and guide them to the " "right subscription tier.\n\n" "PERSONA: Warm, knowledgeable, zero-pressure. You are a language learning " "advisor, not a salesperson. Speak in the student's native language with " "occasional demonstration phrases in their target language to showcase " "what they'll learn.\n\n" "CONVERSATION FLOW:\n" "1. WARM WELCOME (30 seconds)\n" " - Welcome to Fluente.ai! I'm here to help you find the perfect way " "to learn [language].\n" " - If visitor context provides a goal or referral source, reference it " "naturally.\n\n" "2. NEEDS DISCOVERY (2-3 minutes) -- Ask ONE question at a time:\n" " - What language do you want to learn?\n" " - Why? (career, travel, exam, relationship, curiosity)\n" " - Have you studied it before? How would you rate yourself?\n" " - How much time per day can you dedicate? (5 min / 15 min / 30+ min)\n" " - Any deadline? (trip, exam, job start)\n" " - Call save_lead_data() after gathering this info.\n\n" "3. MINI DEMONSTRATION (1-2 minutes)\n" " - Teach 3 useful phrases related to their stated goal.\n" " - Use the target language naturally. Ask them to repeat.\n\n" "4. PERSONALISED RECOMMENDATION (1 minute)\n" " - Call recommend_tier() to get the right subscription.\n" " - Present it naturally: Based on what you've told me...\n" " - ALWAYS mention the free tier as an option.\n\n" "5. HANDOVER TO ASSESSMENT (if student is interested)\n" " - Want to see exactly where you are? I can connect you with your " "personal teacher for a quick 3-minute assessment.\n" " - Call start_assessment() if they agree.\n\n" "RULES:\n" "- Keep responses SHORT (1-3 sentences). This is voice, not text.\n" "- Be warm, encouraging, and celebrate interest.\n" "- Never sound like a salesperson. You are an advisor.\n" "- Never mention internal terms like state machines, phases, conversion." ) ASSESSMENT_PROMPT = ( "You are a Fluente.ai language teacher conducting a placement assessment " "for a new student. This assessment determines the student's CEFR level " "(A1-B2) to place them correctly.\n\n" "CRITICAL -- ACCENT & IDENTITY RULE:\n" "You are a NATIVE speaker of the target language. Your accent must be " "authentic. You must NEVER sound like an English speaker attempting the " "target language.\n\n" "ASSESSMENT STRUCTURE (3-5 minutes):\n\n" "PHASE 1 -- LISTENING COMPREHENSION (60 seconds)\n" "- Say 3 sentences of increasing complexity in the target language.\n" "- After each, ask a simple comprehension question.\n" "- Sentence 1: A1-level (present tense, basic vocab)\n" "- Sentence 2: B1-level (past tense, compound sentence)\n" "- Sentence 3: B2-level (subjunctive/conditional, abstract vocab)\n" '- Record results with record_phase_result(phase="listening", ...).\n\n' "PHASE 2 -- ADAPTIVE CONVERSATION (90-120 seconds)\n" "- Start at A2 level. Adjust up/down based on responses.\n" "- Topics: daily routine, travel, work, opinions, hypotheticals.\n" '- Record results with record_phase_result(phase="speaking", ...).\n\n' "PHASE 3 -- PRONUNCIATION SAMPLE (30-60 seconds)\n" "- Ask student to repeat 3 carefully chosen phrases.\n" "- Score 0-100 based on intelligibility, not native-like accuracy.\n" '- Record with record_phase_result(phase="pronunciation", ...).\n\n' "PHASE 4 -- RESULT & PERSONALISED PLAN (60 seconds)\n" "- Synthesise all phase results into a final CEFR level.\n" "- Present positively: Your level is [X], which means...\n" "- Outline their 30-day milestone.\n" "- Call complete_assessment() with all data.\n\n" "RULES:\n" "- Keep responses SHORT (1-3 sentences). This is voice.\n" "- Be encouraging. Never make the student feel tested or judged.\n" "- Use thinking mode for level decisions (you have thinking budget enabled)." ) SPANISH_ASSESSMENT_ADDON = ( "=== SPANISH-SPECIFIC ASSESSMENT ADDITIONS ===\n\n" "DIALECT AWARENESS:\n" "Before starting, ask: Are you learning Spanish for Spain, Latin America, " "or both?\n\n" "SPANISH PRONUNCIATION MARKERS TO ASSESS:\n" "- /r/ vs /rr/ (pero vs perro) -- single biggest tell for level\n" "- /b/ vs /v/ distinction (most dialects merge these)\n" "- /d/ in intervocalic position (cansado -> cansa'o in casual speech)\n" "- Stress patterns on verb conjugations\n" "- Diphthongs (bueno, fuego, ciudad)\n\n" "SPANISH GRAMMAR GATES:\n" "A2 -> B1: ser vs estar usage in context\n" "B1 -> B2: subjunctive in natural speech" ) CEFR_BLOCKS: dict[str, str] = { "A1": ( "LEVEL RULES -- A1 (BEGINNER):\n" "LANGUAGE MIX: 60% source language / 40% target language\n" "- Lead with target language, immediately follow with source translation.\n" "- Pattern: [Target phrase]. That means [translation]. Can you say " "[target phrase]?\n" "SPEECH: Very slow. Present tense only. Max 5-word sentences in target " "language.\n" "Repeat key vocabulary 3-4 times naturally.\n" "CORRECTION: Never interrupt. Model correct form only.\n" "Accept close-enough pronunciation. Focus ONLY on lesson vocabulary.\n" "DO: Yes/no questions, A/B choices, celebrate every attempt.\n" "DON'T: Explain grammar, use complex tenses, ask open-ended questions." ), "A2": ( "LEVEL RULES -- A2 (ELEMENTARY):\n" "LANGUAGE MIX: 40% source language / 60% target language\n" "- Source language only for NEW vocabulary. Previously learned words " "stay in target.\n" "SPEECH: Slow-moderate pace. Present + simple past. Up to 8-word " "sentences.\n" "Simple connectors: and, but, because, then.\n" "CORRECTION: Recast naturally -- repeat what they said correctly " "without highlighting error.\n" "DO: Simple WH-questions. Build on responses.\n" "DON'T: Detailed grammar explanations. Complex structures." ), "B1": ( "LEVEL RULES -- B1 (INTERMEDIATE):\n" "LANGUAGE MIX: 15% source language / 85% target language\n" "- Source language ONLY for genuinely confusing abstractions.\n" "- If student switches to source, redirect: Can you try that in " "[target]?\n" "SPEECH: Natural pace. All common tenses including future and " "conditional.\n" "Full sentences up to 12 words. Common idioms OK.\n" "CORRECTION: Mid-sentence correction enabled (Practice/Immersion " "tiers).\n" "Briefly note rule, model correct, continue. Push self-correction.\n" "DO: Natural conversation. Follow-up questions requiring elaboration.\n" "DON'T: Over-explain grammar. Accept one-word answers." ), "B2": ( "LEVEL RULES -- B2 (UPPER INTERMEDIATE):\n" "LANGUAGE MIX: 5% source language / 95% target language\n" "- Source language only for metalinguistic discussion.\n" "SPEECH: Full native speed with natural contractions and flow.\n" "All tenses and moods. Idioms, colloquialisms, register-appropriate " "language.\n" "CORRECTION: Real-time interruption enabled (Immersion tier).\n" "Correct in target language. Focus on precision: articles, " "prepositions.\n" "Challenge false friends and fossilised errors.\n" "Push native-like phrasing: Correct, but a native would say...\n" "DO: Debate, opinions, hypotheticals. Complex role-play. Register " "variation.\n" "DON'T: Accept good enough. Slow down unless asked." ), } BASE_TUTOR_PROMPT = ( "You are a Fluente.ai language tutor. You are having a real-time voice " "conversation with a student.\n\n" "SCENARIO: {scenario_title}\n" "LESSON: {lesson_id}\n" "STUDENT CEFR LEVEL: {cefr_level}\n" "TARGET LANGUAGE: {target_language}\n" "SOURCE LANGUAGE: {source_language}\n" "TEACHER PERSONA: {teacher_name} (native {target_language} speaker from " "{teacher_origin})\n\n" "CRITICAL -- ACCENT & IDENTITY RULE:\n" "You are {teacher_name}, a NATIVE {target_language} speaker. You must " "ONLY produce {target_language} with a native accent. You must NEVER " "sound like an English speaker attempting {target_language}. Your English " "(when used for support) should carry a natural {teacher_origin} accent." "\n\n" "CORE RULES (ALL LEVELS):\n" "1. Stay in character as {teacher_name} at all times\n" "2. Keep responses SHORT (1-3 sentences). This is voice, not text.\n" "3. Focus on the scenario vocabulary and learning objectives\n" "4. Be warm, encouraging, and celebrate progress\n" "5. Never break character or discuss AI/technology\n" "6. Match speech speed and complexity to the CEFR level rules below\n\n" "{cefr_block}\n\n" "=== RUNTIME CONTEXT ===\n" "VOCABULARY_FOR_THIS_LESSON: {vocab_json}\n" "STUDENT_WEAK_AREAS: {weak_areas_json}\n" "SESSION_TYPE: {session_type}\n" "CORRECTION_MODE: {correction_mode}\n" "TEACHER_PERSONA: {teacher_json}" ) # ====================================================================== # 5. HELPER FUNCTIONS # ====================================================================== def get_correction_mode(tier: str, cefr_level: str) -> str: """Correction mode = min(tier_allows, level_appropriate). An A1 beginner on Immersion still gets gentle (interruption paralyses beginners). A B2 student on Practice gets mid_sentence (tier caps at mid_sentence). """ TIER_MAX = { "free": "gentle", "listen": "gentle", "practice": "gentle", "starter": "gentle", "conversation": "mid_sentence", "immersion": "real_time", } LEVEL_MAX = { "A1": "gentle", "A2": "gentle", "B1": "mid_sentence", "B2": "real_time", } modes = ["gentle", "mid_sentence", "real_time"] tier_cap = TIER_MAX.get(tier, "gentle") level_cap = LEVEL_MAX.get(cefr_level, "gentle") return modes[min(modes.index(tier_cap), modes.index(level_cap))] def get_teacher_voice(target_language: str) -> str: """Get the default Fluente TTS voice for a target language.""" lang = target_language.lower().split("-")[0] if target_language else "en" return INWORLD_DEFAULT_VOICES.get(lang, INWORLD_DEFAULT_VOICE) def load_teacher(teacher_language: str) -> dict: """Load teacher persona for the given language.""" lang = teacher_language.lower() if teacher_language else "en" teacher = TEACHER_DB.get(lang) if not teacher: base = lang.split("-")[0] teacher = TEACHER_DB.get(base) if not teacher: teacher = TEACHER_DB["en"] return teacher async def load_lesson(lesson_id: str) -> dict: """Load lesson data from Supabase. Falls back to stub if unavailable.""" supa = _supa() if supa: try: result = ( await supa.table("lessons") .select("*") .eq("lesson_id", lesson_id) .maybe_single() .execute() ) if result.data: return result.data except Exception as exc: logger.warning(f"Failed to load lesson {lesson_id}: {exc}") # Stub fallback — still allows the agent to start return { "lesson_id": lesson_id, "title": f"Lesson {lesson_id}", "vocabulary": [], "scenario": "General Practice", "objectives": [], } async def load_student_profile(student_id: str) -> dict: """Load student profile from Supabase.""" supa = _supa() if not supa: return {} try: result = ( await supa.table("student_profiles") .select("*") .eq("student_id", student_id) .maybe_single() .execute() ) return result.data or {} except Exception as exc: logger.warning(f"Failed to load student profile: {exc}") return {} def build_tutor_prompt( lesson: dict, student_data: FluenteUserData, teacher: dict, ) -> str: """Build the tutor system prompt from lesson data, student context, and teacher persona.""" cefr = student_data.current_cefr_level or "A2" target_lang = LANGUAGE_NAMES.get( (student_data.target_language or "es").split("-")[0], "Spanish" ) source_lang = LANGUAGE_NAMES.get( (student_data.source_language or "en").split("-")[0], "English" ) correction = student_data.correction_mode or get_correction_mode( student_data.subscription_tier or "free", cefr ) return BASE_TUTOR_PROMPT.format( scenario_title=lesson.get("scenario", "General Practice"), lesson_id=lesson.get("lesson_id", "L01_P1"), cefr_level=cefr, target_language=target_lang, source_language=source_lang, teacher_name=teacher.get("name", "Teacher"), teacher_origin=teacher.get("origin", ""), cefr_block=CEFR_BLOCKS.get(cefr, CEFR_BLOCKS["A2"]), vocab_json=json.dumps( student_data.lesson_vocabulary or lesson.get("vocabulary", []) ), weak_areas_json=json.dumps(student_data.student_weak_areas), session_type=student_data.session_type or "tutoring", correction_mode=correction, teacher_json=json.dumps(teacher, indent=2), ) async def _publish_rpc(room: rtc.Room, method: str, data: dict) -> None: """Publish an RPC-style event to the frontend via the data channel.""" lp = room.local_participant if not lp: return try: payload = json.dumps({"type": "rpc", "method": method, "data": data}) await lp.publish_data(payload, reliable=True) except Exception as exc: logger.warning(f"Failed to publish RPC {method}: {exc}") # ====================================================================== # 6. SALES CONCIERGE AGENT (4 tools) # ====================================================================== class SalesConciergeAgent(Agent): """Pre-signup needs discovery & demo agent.""" def __init__( self, source_language: Optional[str] = None, voice_override: Optional[str] = None, ) -> None: # Prefer explicit voice override from room metadata (test panel # selection), otherwise fall back to a language-appropriate default. voice = voice_override or get_teacher_voice(source_language or "en") super().__init__( instructions=SALES_CONCIERGE_PROMPT, llm=_make_llm(), stt=deepgram.STT(model="nova-3", language="multi"), tts=inworld.TTS(voice=voice, model=INWORLD_TTS_MODEL), ) async def on_enter(self) -> None: await self.session.generate_reply( instructions="Greet the student warmly in their detected language." ) @function_tool() async def save_lead_data( self, ctx: RunContext[FluenteUserData], target_language: str, motivation: str, self_assessed_level: str, specific_goal: str = "", daily_time_minutes: int = 15, deadline: str = "", ) -> str: """Save the student needs discovery data. Call after gathering info. Required: target_language, motivation, self_assessed_level.""" ud = ctx.userdata ud.target_language = target_language ud.motivation = motivation ud.self_assessed_level = self_assessed_level ud.specific_goal = specific_goal ud.daily_time_minutes = daily_time_minutes ud.deadline = deadline supa = _supa() if supa and ud.student_id: try: await supa.table("leads").upsert( { "student_id": ud.student_id, "target_language": target_language, "motivation": motivation, "self_assessed_level": self_assessed_level, "specific_goal": specific_goal, "daily_time_minutes": daily_time_minutes, "deadline": deadline, } ).execute() except Exception as exc: logger.warning(f"Failed to save lead data: {exc}") return "Lead data saved." @function_tool() async def recommend_tier(self, ctx: RunContext[FluenteUserData]) -> str: """Analyse collected data and recommend the best tier. Call after save_lead_data.""" ud = ctx.userdata if ud.specific_goal and any( w in ud.specific_goal.lower() for w in ["cert", "toeic", "dele", "delf", "ielts", "cambridge"] ): tier = "immersion" desc = "Immersion at $19.99/mo -- unlimited AI + cert prep" elif ud.daily_time_minutes >= 15: tier = "conversation" desc = "Conversation at $9.99/mo -- 50 AI sessions" elif ud.daily_time_minutes >= 5: tier = "practice" desc = "Practice at $4.99/mo -- structured speaking" else: tier = "listen" desc = "Listen (Free) -- core lessons in 140+ languages" ud.recommended_tier = tier supa = _supa() if supa and ud.student_id: try: await supa.table("leads").update( {"recommended_tier": tier} ).eq("student_id", ud.student_id).execute() except Exception as exc: logger.warning(f"Failed to update tier: {exc}") return f"Recommended: {desc}. Always mention the free tier too." @function_tool() async def start_assessment(self, ctx: RunContext[FluenteUserData]): """Hand off to level assessment. Call when student agrees to test. Tell the student you're connecting them with their teacher.""" ud = ctx.userdata ud.mini_demo_completed = True room = ctx.session.room if room: await _publish_rpc( room, "fluente:session_phase", {"phase": "assessment"} ) return AssessmentAgent( target_language=ud.target_language, source_language=ud.source_language, chat_ctx=ctx.session.chat_ctx, ) @function_tool() async def show_in_app( self, ctx: RunContext[FluenteUserData], action: str, data: str = "", ) -> None: """Trigger frontend UI. action: show_pricing | show_demo | show_signup | show_free_trial""" room = ctx.session.room if room: await _publish_rpc(room, f"fluente:{action}", {"data": data}) return None # ====================================================================== # 7. ASSESSMENT AGENT (2 tools) # ====================================================================== class AssessmentAgent(Agent): """Adaptive CEFR placement test agent. THINKING MODE ON for this agent (budget 2048 tokens). """ def __init__( self, target_language: Optional[str] = None, source_language: Optional[str] = None, voice_override: Optional[str] = None, chat_ctx=None, ) -> None: prompt = ASSESSMENT_PROMPT if target_language and target_language.lower().startswith("es"): prompt += "\n\n" + SPANISH_ASSESSMENT_ADDON voice = voice_override or get_teacher_voice(target_language or "es") super().__init__( instructions=prompt, llm=_make_llm(thinking=True), stt=deepgram.STT(model="nova-3", language="multi"), tts=inworld.TTS(voice=voice, model=INWORLD_TTS_MODEL), chat_ctx=chat_ctx, ) self._target_language = target_language self._source_language = source_language async def on_enter(self) -> None: await self.session.generate_reply( instructions=( "Introduce yourself as the student's teacher. " "Begin Phase 1 of assessment." ) ) @function_tool() async def record_phase_result( self, ctx: RunContext[FluenteUserData], phase: str, assessed_level: str, notes: str = "", ) -> str: """Record one assessment phase result. phase: listening | speaking | grammar | vocabulary | pronunciation assessed_level: A1 | A2 | B1 | B2 (or 0-100 for pronunciation)""" ud = ctx.userdata if phase == "pronunciation" and assessed_level.isdigit(): ud.pronunciation_score = int(assessed_level) elif phase == "listening": ud.listening_score = assessed_level elif phase == "speaking": ud.speaking_score = assessed_level elif phase == "grammar": ud.grammar_score = assessed_level elif phase == "vocabulary": ud.vocabulary_score = assessed_level elif phase == "fluency": ud.fluency_score = assessed_level if notes: ud.weak_areas.append(notes) return f"Phase {phase} recorded as {assessed_level}." @function_tool() async def complete_assessment( self, ctx: RunContext[FluenteUserData], final_level: str, weak_areas: list, strong_areas: list, recommended_start_lesson: str, daily_target_minutes: int = 15, thirty_day_milestone: str = "", ): """Finalise assessment. Save to Supabase and hand off to tutoring. Call after all phases complete.""" ud = ctx.userdata ud.assessed_level = final_level ud.current_cefr_level = final_level ud.weak_areas = weak_areas ud.strong_areas = strong_areas ud.recommended_start_lesson = recommended_start_lesson supa = _supa() if supa and ud.student_id: try: await supa.table("student_profiles").upsert( { "student_id": ud.student_id, "assessed_level": final_level, "listening_score": ud.listening_score, "speaking_score": ud.speaking_score, "vocabulary_score": ud.vocabulary_score, "grammar_score": ud.grammar_score, "pronunciation_score": ud.pronunciation_score, "weak_areas": weak_areas, "strong_areas": strong_areas, "recommended_start_lesson": recommended_start_lesson, "daily_target_minutes": daily_target_minutes, "thirty_day_milestone": thirty_day_milestone, "teacher_assigned": ud.teacher_name, "subscription_tier": ud.subscription_tier or "free", "target_language": ud.target_language, "source_language": ud.source_language, } ).execute() except Exception as exc: logger.warning(f"Failed to save assessment: {exc}") room = ctx.session.room if room: await _publish_rpc( room, "fluente:session_phase", {"phase": "assessment_complete", "level": final_level}, ) tutor = await TutorAgent.create( lesson_id=recommended_start_lesson, student_data=ud ) return ( tutor, f"Assessment complete. Student placed at {final_level}.", ) # ====================================================================== # 8. TUTOR AGENT (6 tools) # ====================================================================== class TutorAgent(Agent): """Main tutoring agent. 2 silent + 4 speaking tools. Use the async factory ``await TutorAgent.create(...)`` instead of the constructor directly so that lesson data can be loaded from Supabase. """ def __init__( self, lesson: dict, student_data: FluenteUserData, voice_override: Optional[str] = None, ) -> None: teacher = load_teacher( student_data.teacher_language or student_data.target_language or "en" ) prompt = build_tutor_prompt(lesson, student_data, teacher) lesson_id = lesson.get("lesson_id", "L01_P1") student_data.current_lesson_id = lesson_id student_data.teacher_name = teacher.get("name", "Teacher") student_data.teacher_origin = teacher.get("origin", "") # Prefer explicit voice override, then teacher persona voice, then # language default. voice = ( voice_override or teacher.get("tts_voice_id") or get_teacher_voice(student_data.target_language or "en") ) super().__init__( instructions=prompt, llm=_make_llm(), stt=deepgram.STT(model="nova-3", language="multi"), tts=inworld.TTS(voice=voice, model=INWORLD_TTS_MODEL), ) self._lesson = lesson self._student_data = student_data @classmethod async def create( cls, lesson_id: str, student_data: FluenteUserData, voice_override: Optional[str] = None, ) -> "TutorAgent": """Async factory — loads lesson from Supabase then builds the agent.""" lesson = await load_lesson(lesson_id) return cls(lesson=lesson, student_data=student_data, voice_override=voice_override) async def on_enter(self) -> None: await self.session.generate_reply( instructions=( "Continue from where the pre-recorded lesson left off." ) ) # -- Silent tools (fire-and-forget Supabase writes) -- @function_tool() async def log_correction( self, ctx: RunContext[FluenteUserData], error_type: str, student_said: str, correct_form: str, severity: str = "minor", ) -> None: """Log a correction. SILENT -- does not interrupt conversation. error_type: grammar | pronunciation | vocabulary | register severity: minor | major | fossilised. Call EVERY time you correct the student.""" ctx.userdata.corrections_made += 1 supa = _supa() if supa and ctx.userdata.student_id: _spawn( supa.table("corrections_log") .insert( { "student_id": ctx.userdata.student_id, "lesson_id": ctx.userdata.current_lesson_id, "error_type": error_type, "student_said": student_said, "correct_form": correct_form, "severity": severity, "cefr_level": ctx.userdata.current_cefr_level, } ) .execute() ) return None # Silent -- LLM continues speaking without waiting @function_tool() async def mark_vocab_practiced( self, ctx: RunContext[FluenteUserData], vocab_item: str, used_correctly: bool, context: str = "", ) -> None: """Record vocab usage. SILENT. Call each time student uses a target word.""" ctx.userdata.vocab_items_practiced.append(vocab_item) supa = _supa() if supa and ctx.userdata.student_id: _spawn( supa.table("vocabulary_progress") .upsert( { "student_id": ctx.userdata.student_id, "vocab_item": vocab_item, "lesson_id": ctx.userdata.current_lesson_id, "used_correctly": used_correctly, "context": context, } ) .execute() ) return None # Silent # -- Speaking tools -- @function_tool() async def next_lesson( self, ctx: RunContext[FluenteUserData], lesson_id: str, reason: str = "completed", ): """Move to a different lesson. Hands off to new TutorAgent. reason: completed | too_easy | too_hard | student_request""" ud = ctx.userdata supa = _supa() if supa and ud.student_id: try: duration = 0.0 if ud.session_start: duration = ( datetime.now(timezone.utc) - ud.session_start ).total_seconds() await supa.table("session_log").insert( { "student_id": ud.student_id, "lesson_id": ud.current_lesson_id, "corrections_made": ud.corrections_made, "vocab_practiced": ud.vocab_items_practiced, "duration_seconds": duration, "next_lesson": lesson_id, "reason": reason, } ).execute() except Exception as exc: logger.warning(f"Failed to log session: {exc}") room = ctx.session.room if room: await _publish_rpc( room, "fluente:lesson_change", {"lesson_id": lesson_id, "reason": reason}, ) tutor = await TutorAgent.create( lesson_id=lesson_id, student_data=ud ) return ( tutor, "Great work! Let's move on.", ) @function_tool() async def request_reassessment( self, ctx: RunContext[FluenteUserData] ): """Trigger a new CEFR assessment. Call when student requests level check or performance suggests different level.""" ud = ctx.userdata room = ctx.session.room if room: await _publish_rpc( room, "fluente:session_phase", {"phase": "reassessment"}, ) return AssessmentAgent( target_language=ud.target_language, source_language=ud.source_language, ) @function_tool() async def end_session( self, ctx: RunContext[FluenteUserData], summary: str = "", ) -> None: """End the session. Call when student says goodbye or time limit reached.""" ud = ctx.userdata supa = _supa() if supa and ud.student_id: try: duration = 0.0 if ud.session_start: duration = ( datetime.now(timezone.utc) - ud.session_start ).total_seconds() await supa.table("session_log").insert( { "student_id": ud.student_id, "lesson_id": ud.current_lesson_id, "corrections_made": ud.corrections_made, "vocab_practiced": ud.vocab_items_practiced, "duration_seconds": duration, "summary": summary, } ).execute() except Exception as exc: logger.warning(f"Failed to log session end: {exc}") room = ctx.session.room if room: await _publish_rpc( room, "fluente:session_end", { "summary": summary, "corrections": ud.corrections_made, "vocab_count": len(ud.vocab_items_practiced), }, ) await room.disconnect() return None @function_tool() async def suggest_upgrade( self, ctx: RunContext[FluenteUserData], reason: str, ) -> str: """Show upgrade banner. reason: mid_sentence_correction | more_languages | certification_prep | unlimited_sessions. Once per session max. Keep natural.""" room = ctx.session.room if room: await _publish_rpc( room, "fluente:show_upgrade", { "reason": reason, "current_tier": ctx.userdata.subscription_tier, }, ) return f"Upgrade suggestion shown for: {reason}" # ====================================================================== # 9. SESSION ENTRYPOINT # ====================================================================== async def entrypoint(ctx: JobContext) -> None: """Main entrypoint. Routes students to the correct starting agent based on room metadata.""" logger.info(f"Fluente agent starting for room: {ctx.room.name}") await ctx.connect() await init_supabase() room_data = json.loads(ctx.room.metadata or "{}") # Configurable VAD silence duration from room metadata. # Supports modes: give_me_time (3000ms), slow (1500ms), rapid_fire (400ms). # Default 800ms suits conversational language learning. _vad_silence_ms_map = { "give_me_time": 3000, "slow": 1500, "rapid_fire": 400, } _tempo = room_data.get("tempo", "") _vad_silence_ms = int( room_data.get("vad_silence_ms") or _vad_silence_ms_map.get(_tempo, 800) ) _vad = silero.VAD.load(min_silence_duration=_vad_silence_ms / 1000.0) logger.info("VAD silence duration: %dms (tempo=%r)", _vad_silence_ms, _tempo) session = AgentSession[FluenteUserData]( userdata=FluenteUserData(), vad=_vad, ) student_id = room_data.get("student_id") or room_data.get("visitor_id") if not student_id: # Anonymous visitors (common for Sales Concierge) still need a stable ID # so Supabase upserts don't write `NULL` keys. student_id = f"visitor:{ctx.room.name}" session_type = room_data.get("session_type", "sales_concierge") language = room_data.get("language", "es") voice = room_data.get("voice", "") # Populate userdata from room metadata ud = session.userdata ud.student_id = student_id ud.session_type = session_type ud.target_language = language ud.source_language = room_data.get("source_language", "en") # Load student profile from Supabase (if returning student) if student_id: profile = await load_student_profile(student_id) if profile: ud.assessed_level = profile.get("assessed_level") ud.current_cefr_level = ud.assessed_level ud.subscription_tier = profile.get( "subscription_tier", "free" ) ud.teacher_name = profile.get("teacher_assigned") ud.target_language = ( profile.get("target_language") or ud.target_language ) ud.source_language = ( profile.get("source_language") or ud.source_language ) ud.student_weak_areas = profile.get("weak_areas", []) # Apply test panel overrides (if present in room metadata) if room_data.get("cefr_level"): ud.current_cefr_level = room_data["cefr_level"] if room_data.get("correction_mode"): ud.correction_mode = room_data["correction_mode"] if room_data.get("tier") and room_data["tier"] != "new_visitor": ud.subscription_tier = room_data["tier"] if room_data.get("teacher_persona"): try: teacher_data = json.loads(room_data["teacher_persona"]) ud.teacher_name = teacher_data.get("name", ud.teacher_name) ud.teacher_origin = teacher_data.get( "origin", ud.teacher_origin ) ud.teacher_language = teacher_data.get( "language", ud.teacher_language ) except (json.JSONDecodeError, TypeError): pass if room_data.get("vocabulary_items"): items = room_data["vocabulary_items"] if isinstance(items, str): ud.lesson_vocabulary = [ v.strip() for v in items.split(",") if v.strip() ] elif isinstance(items, list): ud.lesson_vocabulary = items if room_data.get("weak_areas"): wa = room_data["weak_areas"] if isinstance(wa, str): ud.student_weak_areas = [ w.strip() for w in wa.split(",") if w.strip() ] elif isinstance(wa, list): ud.student_weak_areas = wa # Resolve correction mode if not explicitly set if not ud.correction_mode: ud.correction_mode = get_correction_mode( ud.subscription_tier or "free", ud.current_cefr_level or "A2", ) # Voice override from room metadata (test panel voice selector). # Only honour it when it looks like a valid Fluente voice ID. voice_override: Optional[str] = None if voice and voice.startswith("fluente__"): voice_override = voice # Route to correct initial agent if session_type in ("sales_concierge", "concierge") and not ud.assessed_level: initial_agent = SalesConciergeAgent( source_language=ud.source_language, voice_override=voice_override, ) elif session_type in ( "assessment", "level_assessment", "spanish_assessment", ): initial_agent = AssessmentAgent( target_language=ud.target_language, source_language=ud.source_language, voice_override=voice_override, ) else: lesson_id = ( room_data.get("lesson_id") or ud.recommended_start_lesson or "L01_P1" ) initial_agent = await TutorAgent.create( lesson_id=lesson_id, student_data=ud, voice_override=voice_override, ) ud.session_start = datetime.now(timezone.utc) logger.info( "Starting session -- type: %s, agent: %s, language: %s, " "voice: %s, cefr: %s, tier: %s, correction: %s", session_type, type(initial_agent).__name__, language, voice, ud.current_cefr_level, ud.subscription_tier, ud.correction_mode, ) await session.start( agent=initial_agent, room=ctx.room, room_input_options=RoomInputOptions( **({ "noise_cancellation": noise_cancellation.BVC() } if noise_cancellation else {}), ), ) # -- Error handling -- # Log LLM/STT/TTS errors. FallbackAdapter handles automatic retry & # failover. For unrecoverable errors (recoverable=False), tell the user. @session.on("error") def _on_error(ev: ErrorEvent) -> None: err = ev.error recoverable = getattr(ev, "recoverable", True) source_name = type(ev.source).__name__ logger.error( "Pipeline error from %s (recoverable=%s): %s", source_name, recoverable, err, ) # Forward error details to the frontend via data channel so # studio UI can display them without needing LK Cloud login. _spawn( _publish_json( { "type": "agent_error", "source": source_name, "recoverable": recoverable, "message": str(err), } ) ) if not recoverable: # Tell the user something went wrong so the call doesn't just # go silent. session.say() is non-blocking / fire-and-forget. _spawn( session.say( "I'm sorry, I had a brief technical issue. " "Could you repeat that?" ) ) # -- Transcript forwarding over data channel -- local_participant = ctx.room.local_participant async def _publish_json(payload: dict) -> None: """Publish a JSON message to all participants via data channel.""" if local_participant is None: return try: await local_participant.publish_data( json.dumps(payload), reliable=True, topic="viseme", ) except Exception as exc: logger.warning(f"Failed to publish data: {exc}") @session.on("user_input_transcribed") def _on_user_transcript(ev: UserInputTranscribedEvent) -> None: if not ev.is_final: return text = ev.transcript.strip() if not text: return logger.info(f"User said: {text[:80]}") _spawn( _publish_json( {"type": "transcript", "role": "user", "text": text} ) ) _spawn( _publish_json( { "type": "pipeline_status", "step": "stt", "status": "done", } ) ) _spawn( _publish_json( { "type": "pipeline_status", "step": "llm", "status": "active", } ) ) @session.on("conversation_item_added") def _on_conversation_item(ev: ConversationItemAddedEvent) -> None: item = ev.item if getattr(item, "role", None) != "assistant": return raw_text = getattr(item, "text_content", "") or "" stripped = raw_text.strip() if not stripped: return logger.info(f"Agent said: {stripped[:80]}") _spawn( _publish_json( {"type": "transcript", "role": "agent", "text": stripped} ) ) _spawn( _publish_json( { "type": "pipeline_status", "step": "llm", "status": "done", } ) ) _spawn( _publish_json( { "type": "pipeline_status", "step": "tts", "status": "active", } ) ) # -- Lifecycle management -- _disconnect_event = asyncio.Event() local_identity = ( local_participant.identity if local_participant else None ) @ctx.room.on("disconnected") def _on_room_disconnected(*_args: object) -> None: _disconnect_event.set() @ctx.room.on("participant_disconnected") def _on_participant_left( participant: rtc.RemoteParticipant, ) -> None: if participant.identity != local_identity: logger.info( f"User {participant.identity} left -- disconnecting agent" ) _spawn(ctx.room.disconnect()) # Wait for room disconnect await _disconnect_event.wait() # Log session analytics elapsed = int( (datetime.now(timezone.utc) - ud.session_start).total_seconds() if ud.session_start else 0 ) logger.info( "session_end: type=%s lang=%s agent=%s duration=%ds " "corrections=%d vocab=%d", session_type, language, type(initial_agent).__name__, elapsed, ud.corrections_made, len(ud.vocab_items_practiced), ) # ====================================================================== # 10. PREWARM & CLI # ====================================================================== def prewarm(proc) -> None: """Prewarm the VAD model so it is ready when the first session starts.""" logger.info("Prewarming VAD...") proc.userdata["vad"] = silero.VAD.load() logger.info("VAD prewarmed") if __name__ == "__main__": cli.run_app( WorkerOptions( entrypoint_fnc=entrypoint, prewarm_fnc=prewarm, agent_name="Inworld", ) )