# llm_server.py # Main FastAPI server — Multilingual Groq LLM Call Center Voice Orchestrator # Exposed on port 7860 on HF Spaces # # v3 — Issues 4 + 5 fixed (mirrors voice_pipeline.py v4) # Applied to the /chat/stream SSE path for consistency with the # primary voice path (the-brain's WS voice flow uses voice_pipeline.py, # not this file — but /chat/stream is a secondary entry point that # should behave identically): # Issue 4: first chunk of a turn uses clause-level (comma) TTS # splitting instead of waiting for a full sentence, reducing # time-to-first-audio on long opening sentences. # Issue 5: _strip_foreign_scripts() runs on every sentence before # TTS — hard safety net removing CJK/Devanagari/Cyrillic/Hangul/ # Thai characters regardless of what the LLM produced. # # v2 — URDU-ONLY MODE (matches voice_pipeline.py) # _fire_tts() previously re-detected language PER SENTENCE via # detect_language_from_content(), and used that (not the session's # locked language) to pick the TTS voice. Since this product is # Urdu-only for now, tts_lang is now ALWAYS normalize_language(language) # — the session's locked language (always "ur"). The per-sentence # content-detection override is preserved as a comment for when # per-agent language selection is reintroduced. This only affects the # /chat/stream SSE path; the primary voice path is voice_pipeline.py. import os import time import logging import asyncio import re import base64 import json import threading import httpx from typing import List, Optional, Dict, Any from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, StreamingResponse from pydantic import BaseModel # Internal Domain Layer Imports from client import groq, fireworks from conversation_manager import ConversationManager, ConversationSession from utils import detect_language_from_content, SUPPORTED_LANGUAGES from language_config import normalize_language from prompt_builder import build_system_prompt from summary import generate_summary from greeting_handler import ( classify_opening_message, get_short_reply, should_short_circuit, ) from utils import ( estimate_payload_tokens, strip_formatting, ) from stt_client import SttClient from voice_orchestrator import VoiceOrchestratorSession import neon_client # ── Logging System Initialization ─────────────────────────────────────────── logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(name)s] %(levelname)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) logger = logging.getLogger("llm_server") # Verify core execution environment states if not os.getenv("GROQ_API_KEY"): logger.critical("CRITICAL: GROQ_API_KEY environment configuration variable is missing.") # ── Architectural Threshold Variables ─────────────────────────────────────── TOKEN_THRESHOLD = 4000 DURATION_THRESHOLD_MINUTES = 30 MODEL_FAST = "llama-3.1-8b-instant" MODEL_HEAVY = "llama-3.3-70b-versatile" FALLBACK_MODEL_FIREWORKS = "accounts/fireworks/models/llama-v3p1-8b-instruct" # ── TTS Space Bridge Configuration ────────────────────────────────────────── # URL is read from env so it can be overridden without a code deploy. TTS_SPACE_URL = os.getenv("TTS_SPACE_URL", "https://voice-tts-tts.hf.space") TTS_SYNTHESISE_TIMEOUT = 15.0 # seconds per sentence request # Multi-script sentence boundary detector. # Captures a run of non-terminator characters followed by one or more # sentence-ending punctuation marks, then optional whitespace / end-of-string. # Terminators covered: # . ! ? — English # \u06d4 \u061f — Urdu (۔) / Arabic-Urdu (؟) SENTENCE_END_RE = re.compile( r'[^.!?\u06d4\u061f\u0964\n]+[.!?\u06d4\u061f\u0964]+(?:\s|$)' r'|[^.!?\u06d4\u061f\u0964\n]*\n', re.UNICODE, ) # Issue 4: clause boundary (comma OR sentence end) — used ONLY for the # first chunk of a /chat/stream turn, to reduce time-to-first-audio on # long opening sentences. Mirrors voice_pipeline.py's fix for the # primary voice path. CLAUSE_END_RE = re.compile( r'[^.!?,\u060c\u06d4\u061f\u0964\n]+[.!?,\u060c\u06d4\u061f\u0964]+(?:\s|$)' r'|[^.!?,\u060c\u06d4\u061f\u0964\n]*\n', re.UNICODE, ) MIN_FIRST_CLAUSE_CHARS = 15 # Issue 5: known foreign (non-Urdu, non-Latin) script ranges. Latin and # digits are intentionally excluded — legitimate in Urdu call speech # (times, prices, acronyms) and handled by the TTS-side transliterator. FOREIGN_SCRIPT_RE = re.compile( r'[\u4E00-\u9FFF\u3400-\u4DBF' # CJK Unified Ideographs + Ext A (Chinese) r'\u3040-\u309F\u30A0-\u30FF' # Hiragana, Katakana (Japanese) r'\uAC00-\uD7A3' # Hangul syllables (Korean) r'\u0900-\u097F' # Devanagari (Hindi) r'\u0400-\u04FF' # Cyrillic (Russian etc.) r'\u0E00-\u0E7F' # Thai r']+' ) # ── FastAPI App Configuration ──────────────────────────────────────────────── app = FastAPI( title="The Brain — Voice LLM Call Center Agent", description="English/Urdu bilingual voice pipeline orchestration layer powered by Groq and Fireworks AI.", version="1.0.0", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["GET", "POST", "OPTIONS"], allow_headers=["*"], expose_headers=["X-Session-Id", "X-Language", "X-Turn-Count"], ) # Global memory state allocations manager = ConversationManager() class CallerFieldSpec(BaseModel): key: str label: str label_ur: str = "" type: str = "text" # ── Unified Pydantic Request Models ───────────────────────────────────────── class StartSessionRequest(BaseModel): company_name: str = "Our Company" agent_name: str = "Sara" agent_gender: str = "male" # Default changed to male custom_rules: List[str] = [] document_context: str = "" # RAG text block container caller_collection_fields: List[CallerFieldSpec] = [] session_id: Optional[str] = None # Generated if missing class ChatRequest(BaseModel): session_id: str message: str company_name: Optional[str] = None agent_name: Optional[str] = None agent_gender: Optional[str] = "male" # Re-supplied for emergency session fallback currentSummary: Optional[str] = "" sessionStart: Optional[int] = None # Epoch milliseconds payload timestamp class EndSessionRequest(BaseModel): session_id: str # ── TTS Bridge Helpers ────────────────────────────────────────────────────── def _extract_sentences(buffer: str) -> tuple[list[str], str]: """ Scan the rolling token buffer for complete sentences using SENTENCE_END_RE. Returns (completed_sentences, leftover_text) so the caller can continue accumulating tokens into the leftover portion. """ sentences: list[str] = [] last_end = 0 for m in SENTENCE_END_RE.finditer(buffer): sentence = buffer[m.start():m.end()].strip() if sentence: sentences.append(sentence) last_end = m.end() return sentences, buffer[last_end:] def _extract_first_clause(buffer: str) -> tuple[Optional[str], str]: """ Issue 4: for the FIRST chunk of a turn only. Finds the earliest point where accumulated text reaches a clause boundary (comma or sentence end) AND is at least MIN_FIRST_CLAUSE_CHARS long. Returns (clause_text, remainder) or (None, buffer) if no qualifying boundary has been reached yet. Mirrors voice_pipeline.py. """ matches = list(CLAUSE_END_RE.finditer(buffer)) if not matches: return None, buffer for m in matches: candidate = buffer[0:m.end()].strip() if len(candidate) >= MIN_FIRST_CLAUSE_CHARS: return candidate, buffer[m.end():] return None, buffer def _strip_foreign_scripts(text: str) -> str: """ Issue 5: hard safety-net filter. Removes characters from scripts that have no place in Urdu-only TTS output (Chinese, Japanese, Korean, Hindi/Devanagari, Cyrillic, Thai) regardless of what the LLM produced. Latin letters and digits are NOT touched. Mirrors voice_pipeline.py. """ if not text: return text cleaned = FOREIGN_SCRIPT_RE.sub('', text) cleaned = re.sub(r'\s{2,}', ' ', cleaned).strip() if cleaned != text.strip(): logger.warning( f"[Issue 5] Stripped foreign-script characters from LLM output: " f"original={text[:80]!r} cleaned={cleaned[:80]!r}" ) return cleaned async def _fire_tts(sentence: str, language: str, gender: str) -> tuple[Optional[bytes], str]: """ Asynchronously POST a single sentence to the deployed TTS Space. Returns (audio_bytes, tts_language) on success, (None, tts_language) on failure. v2 (Urdu-only mode): tts_lang is always the session's language (normalize_language(language)) — no per-sentence content-based re-detection. See module docstring. """ sentence = sentence.strip() if not sentence: return None, normalize_language(language) # Issue 5: strip any foreign-script characters before synthesis. sentence = _strip_foreign_scripts(sentence) if not sentence: return None, normalize_language(language) tts_lang = normalize_language(language) # DISABLED (Urdu-only mode) — per-sentence content-based language # detection. Re-enable for multi-language support: # # tts_lang = normalize_language(detect_language_from_content(sentence)) # session_lang = normalize_language(language) # if tts_lang != session_lang: # logger.info( # "TTS lang from content: session=%s → sentence=%s preview=%r", # session_lang, # tts_lang, # sentence[:60], # ) try: async with httpx.AsyncClient(timeout=TTS_SYNTHESISE_TIMEOUT) as client: resp = await client.post( f"{TTS_SPACE_URL}/synthesise", json={"text": sentence, "language": tts_lang, "gender": gender}, ) if resp.status_code == 200: logger.info(f"TTS ✓ lang={tts_lang} chars={len(sentence)}") return resp.content, tts_lang logger.warning( f"TTS Space returned HTTP {resp.status_code}: {resp.text[:120]}" ) return None, tts_lang except httpx.TimeoutException: logger.error( f"TTS request timed out ({TTS_SYNTHESISE_TIMEOUT}s): {sentence[:60]!r}" ) return None, tts_lang except Exception as exc: logger.error(f"TTS request error: {exc}") return None, tts_lang def _start_groq_stream_thread( params: dict, q: asyncio.Queue, loop: asyncio.AbstractEventLoop, ) -> None: """ Spawns a daemon thread that calls Groq's synchronous streaming API and forwards each content chunk onto the asyncio Queue via loop.call_soon_threadsafe so the async event generator can consume it. Events pushed onto the queue: ("token", str) — a streamed content chunk from the LLM ("done", None) — stream finished cleanly ("error", str) — exception message from the stream """ def _run() -> None: try: stream = groq.chat.completions.create(**{**params, "stream": True}) for chunk in stream: delta = chunk.choices[0].delta if delta and delta.content: loop.call_soon_threadsafe( q.put_nowait, ("token", delta.content) ) loop.call_soon_threadsafe(q.put_nowait, ("done", None)) except Exception as exc: loop.call_soon_threadsafe(q.put_nowait, ("error", str(exc))) threading.Thread(target=_run, daemon=True, name="groq-stream-worker").start() def _select_primary_model(session: ConversationSession, detected_language: str) -> str: ld = session.language_detector if detected_language == "ur" and (ld.is_locked() or ld.confidence >= 0.70): return MODEL_HEAVY return MODEL_FAST def _fields_spec_dicts(session: ConversationSession) -> List[Dict[str, Any]]: return list(session.caller_collection_fields or []) async def _roll_summary(session: ConversationSession, current_summary: str) -> str: if len(session.messages) <= 6: return current_summary or "" messages_to_summarize = session.messages[:-6] session.messages = session.messages[-6:] return await generate_summary( current_summary or "", messages_to_summarize, caller_info=session.caller_info.to_dict(), caller_fields_spec=_fields_spec_dicts(session), ) def _build_prompt(session: ConversationSession, detected_language: str, input_text: str) -> str: greeting_phase = ( session.turn_count <= 3 and not session.language_detector.is_locked() and session.greeting_streak > 0 ) return build_system_prompt( company_name=session.company_name, agent_name=session.agent_name, agent_gender=session.agent_gender, language=detected_language, document_context=session.document_context, custom_rules=session.custom_rules, collected_caller_info=session.caller_info.to_dict(), input_text=input_text, # REMOVED: arabic_clarification (Urdu-only mode) caller_collection_fields=_fields_spec_dicts(session), turn_count=session.turn_count, greeting_phase=greeting_phase, company_name_ur=getattr(session, "company_name_ur", None), agent_name_ur=getattr(session, "agent_name_ur", None), agent_role=getattr(session, "agent_role", None), agent_role_ur=getattr(session, "agent_role_ur", None), overview_ur=getattr(session, "overview_ur", None), ) # ── REST API Router Endpoints ──────────────────────────────────────────────── @app.post("/session/start") async def start_session(req: StartSessionRequest): """ Allocates and registers a stateful conversation block for an incoming call. Returns tracking identifiers required for subsequent transactional steps. """ fields = [f.model_dump() for f in req.caller_collection_fields] session = manager.create_session( company_name=req.company_name, agent_name=req.agent_name, agent_gender=req.agent_gender, # Pass parameter custom_rules=req.custom_rules, document_context=req.document_context, caller_collection_fields=fields, session_id=req.session_id, ) logger.info(f"Communication sequence initialized: {session.session_id}") return { "session_id": session.session_id, "company_name": req.company_name, "agent_name": req.agent_name, "status": "ready", "message": f"Agent {req.agent_name} online and monitoring streaming data frames." } @app.post("/chat") async def chat(req: ChatRequest): """ Processes voice transcript inputs, executes dynamic RAG context assembly, evaluates historical payload token depth, and executes low-latency model inference. """ if not req.message or not req.message.strip(): raise HTTPException(status_code=400, detail="Transaction rejected: Content payload string cannot be empty.") if len(req.message) > 2000: raise HTTPException(status_code=400, detail="Transaction rejected: Maximum allowed buffer exceeds 2000 characters.") # 1. Fetch target state context, fallback to automatic creation if missing session = manager.get_session(req.session_id) if not session: session = manager.create_session( company_name=req.company_name or "Our Company", agent_name=req.agent_name or "Sara", agent_gender=req.agent_gender or "male", # Use request gender, not hardcoded female session_id=req.session_id ) logger.warning(f"Target memory space missing. Initialized emergency backup stack instance: {session.session_id}") start_timestamp = time.time() # 2. Update metadata tracking matrices detected_language = normalize_language(session.process_user_message(req.message)) session.add_message("user", req.message) opening_tier = classify_opening_message( req.message, session.turn_count, session.language_detector.is_locked(), ) # 3. Greeting short-circuit — no full LLM intro on bare greetings if should_short_circuit(opening_tier): session.greeting_streak += 1 short_reply = get_short_reply(opening_tier, req.message, session.greeting_streak) session.add_message("assistant", short_reply) elapsed_ms = int((time.time() - start_timestamp) * 1000) logger.info( f"Greeting short-circuit. ID={session.session_id} | tier={opening_tier} | " f"Latency={elapsed_ms}ms" ) return JSONResponse({ "session_id": session.session_id, "provider": "template-greeting", "response": short_reply, "language": detected_language, "language_locked": session.language_detector.is_locked(), "turn_count": session.turn_count, "caller_info": session.caller_info.to_dict(), "activeSummary": req.currentSummary or "", "elapsed_ms": elapsed_ms, "tokens_used": 0, }) session.greeting_streak = 0 # 4. Rolling window summary new_summary = await _roll_summary(session, req.currentSummary or "") # 5. Build system prompt system_prompt = _build_prompt(session, detected_language, req.message) # 5. Build standard layout configurations optimized_payload = [{"role": "system", "content": system_prompt}] if new_summary: optimized_payload.append({ "role": "system", "content": f"Summary of earlier conversation context for your memory bank: {new_summary}" }) optimized_payload.extend(session.get_history_for_api()) # 6. Evaluate Token Governor Constraints raw_session_start = req.sessionStart or 0 # Guard: clients should send epoch milliseconds (>1e12). If seconds were sent # instead (value < 1e10), convert to avoid a 1000x error that would immediately # trigger the Token Governor on every single request. if 0 < raw_session_start < 1e10: raw_session_start *= 1000 duration_minutes = max(0.0, (time.time() * 1000 - raw_session_start) / 60000) if raw_session_start else 0 estimated_tokens = estimate_payload_tokens(optimized_payload) max_response_tokens: Optional[int] = None if duration_minutes >= DURATION_THRESHOLD_MINUTES or estimated_tokens > TOKEN_THRESHOLD: logger.warning(f"Token Governor triggered. Window size: {estimated_tokens} tokens. Call duration: {duration_minutes:.2f} mins.") # Purge markdown/formatting configurations to preserve strict token structures cleaned_history = [ {"role": m["role"], "content": strip_formatting(m["content"])} for m in session.get_history_for_api() ] concise_controls = { "role": "system", "content": "Token Governor Active: Reply ultra-concisely. No markdown, no filler words. Limit output strictly to 1-2 short conversational sentences." } # Isolate baseline identity parameters to decrease structural overhead parameters short_system_messages = [ m for m in optimized_payload if m.get("role") == "system" and len(m.get("content", "")) < 400 ] optimized_payload = [concise_controls] + short_system_messages if new_summary: optimized_payload.append({ "role": "system", "content": f"Summary of earlier conversation context for your memory bank: {new_summary}" }) optimized_payload.extend(cleaned_history) max_response_tokens = 120 # 7. Model Router (Urdu 70B when locked or confident) primary_model = _select_primary_model(session, detected_language) inference_parameters: Dict[str, Any] = { "model": primary_model, "messages": optimized_payload, "temperature": 0.4, # Marginally cooled to further guarantee script structure adherence "top_p": 0.95 } if max_response_tokens: inference_parameters["max_tokens"] = max_response_tokens # 8. Main Inference Execution Loop with Resilient High-Availability Fallback provider_tag = f"Groq ({primary_model})" loop = asyncio.get_running_loop() try: response = await loop.run_in_executor( None, lambda: groq.chat.completions.create(**inference_parameters) ) assistant_message = response.choices[0].message.content.strip() tokens_used = response.usage.total_tokens except Exception as groq_error: # Check for HTTP/API Status Code 429 Rate Limit Errors status_code = getattr(groq_error, "status_code", None) or getattr(groq_error, "status", None) if status_code == 429 and fireworks is not None: logger.warning("Groq rate limited — falling back to Fireworks") inference_parameters["model"] = FALLBACK_MODEL_FIREWORKS provider_tag = "Fireworks AI (Fallback)" fallback_response = await loop.run_in_executor( None, lambda: fireworks.chat.completions.create(**inference_parameters) ) assistant_message = fallback_response.choices[0].message.content.strip() tokens_used = fallback_response.usage.total_tokens if (hasattr(fallback_response, "usage") and fallback_response.usage) else 0 else: logger.error(f"Unmanaged runtime error during network execution loop: {str(groq_error)}") raise HTTPException(status_code=502, detail=f"Upstream provider failure: {str(groq_error)}") # 9. Register output sequence history session.add_message("assistant", assistant_message) elapsed_ms = int((time.time() - start_timestamp) * 1000) logger.info(f"Execution wrapped. Metrics: ID={session.session_id} | Provider={provider_tag} | Latency={elapsed_ms}ms | Language={detected_language}") return JSONResponse({ "session_id": session.session_id, "provider": provider_tag, "response": assistant_message, "language": detected_language, "language_locked": session.language_detector.is_locked(), "turn_count": session.turn_count, "caller_info": session.caller_info.to_dict(), "activeSummary": new_summary, "elapsed_ms": elapsed_ms, "tokens_used": tokens_used }) @app.post("/chat/stream") async def chat_stream(req: ChatRequest): """ Streaming LLM → TTS bridge endpoint using Server-Sent Events (SSE). Architecture ------------ 1. Groq streaming runs in a daemon thread (_start_groq_stream_thread). 2. Tokens are forwarded onto an asyncio.Queue so the async generator can consume them without blocking the event loop. 3. Each token is appended to a rolling buffer. When SENTENCE_END_RE finds a complete sentence, _fire_tts() immediately fires an async httpx POST to the TTS Space — while the LLM is still generating the rest of the response (the "live speaking" effect). 4. On Groq failure the endpoint transparently falls back to Fireworks (non-streaming) before surfacing an error to the client. SSE event schema (each event is a JSON object on a `data:` line): {"type": "token", "text": "..."} — raw LLM token {"type": "audio", "sentence": "...", "audio_b64": "...", "lang": "..."} — base64 audio/mpeg {"type": "done", "full_text": "...", "language": "...", ...} — final metadata {"type": "error", "detail": "..."} — fatal error """ if not req.message or not req.message.strip(): raise HTTPException(status_code=400, detail="Content payload cannot be empty.") if len(req.message) > 2000: raise HTTPException(status_code=400, detail="Maximum payload exceeds 2000 characters.") # ── 1. Session resolution (mirrors /chat) ───────────────────────────────── session = manager.get_session(req.session_id) if not session: session = manager.create_session( company_name=req.company_name or "Our Company", agent_name=req.agent_name or "Sara", agent_gender=req.agent_gender or "male", # Use request gender, not hardcoded female session_id=req.session_id, ) logger.warning(f"[stream] Emergency session created: {session.session_id}") # ── 2. Language detection + greeting gate ─────────────────────────────── detected_language = normalize_language(session.process_user_message(req.message)) session.add_message("user", req.message) opening_tier = classify_opening_message( req.message, session.turn_count, session.language_detector.is_locked(), ) short_circuit = should_short_circuit(opening_tier) if short_circuit: session.greeting_streak += 1 else: session.greeting_streak = 0 new_summary = await _roll_summary(session, req.currentSummary or "") system_prompt = _build_prompt(session, detected_language, req.message) optimized_payload = [{"role": "system", "content": system_prompt}] if new_summary: optimized_payload.append({ "role": "system", "content": f"Summary of earlier conversation context for your memory bank: {new_summary}", }) optimized_payload.extend(session.get_history_for_api()) # ── 3. Token Governor (mirrors /chat) ───────────────────────────────────── raw_session_start = req.sessionStart or 0 if 0 < raw_session_start < 1e10: raw_session_start *= 1000 duration_minutes = max(0.0, (time.time() * 1000 - raw_session_start) / 60000) if raw_session_start else 0 estimated_tokens = estimate_payload_tokens(optimized_payload) max_response_tokens: Optional[int] = None if duration_minutes >= DURATION_THRESHOLD_MINUTES or estimated_tokens > TOKEN_THRESHOLD: logger.warning( f"[stream] Token Governor triggered. " f"Tokens={estimated_tokens}, Duration={duration_minutes:.1f}m" ) cleaned_history = [ {"role": m["role"], "content": strip_formatting(m["content"])} for m in session.get_history_for_api() ] concise_control = { "role": "system", "content": ( "Token Governor Active: Reply ultra-concisely. " "No markdown, no filler. 1-2 short sentences max." ), } short_sys = [ m for m in optimized_payload if m.get("role") == "system" and len(m.get("content", "")) < 400 ] optimized_payload = [concise_control] + short_sys if new_summary: optimized_payload.append({ "role": "system", "content": f"Summary of earlier conversation context for your memory bank: {new_summary}", }) optimized_payload.extend(cleaned_history) max_response_tokens = 120 # ── 4. Model selection ──────────────────────────────────────────────────── primary_model = _select_primary_model(session, detected_language) inference_params: Dict[str, Any] = { "model": primary_model, "messages": optimized_payload, "temperature": 0.4, "top_p": 0.95, } if max_response_tokens: inference_params["max_tokens"] = max_response_tokens # Resolve TTS gender from session (TTS Space accepts "female" | "male") tts_gender = session.agent_gender if session.agent_gender in ("female", "male") else "female" start_timestamp = time.time() # ── 5. SSE event generator ──────────────────────────────────────────────── async def event_generator(): loop = asyncio.get_running_loop() q: asyncio.Queue = asyncio.Queue() full_text_parts: list[str] = [] buffer = "" # Issue 4: first chunk of the turn uses clause-level splitting # to reduce time-to-first-audio; subsequent chunks use full # sentence-only splitting for natural prosody. first_chunk_dispatched = False def _sse(data: dict) -> str: """Format a dict as an SSE data frame.""" return f"data: {json.dumps(data, ensure_ascii=False)}\n\n" async def _synthesise_and_emit(sentence: str) -> Optional[str]: """ Fire TTS for one sentence. Returns a ready-to-yield SSE audio event string, or None if TTS failed. """ audio_bytes, tts_lang = await _fire_tts(sentence, detected_language, tts_gender) if audio_bytes: return _sse({ "type": "audio", "sentence": sentence, "audio_b64": base64.b64encode(audio_bytes).decode("utf-8"), "lang": tts_lang, }) return None # Greeting short-circuit path — template reply + TTS, no Groq if short_circuit: short_reply = get_short_reply(opening_tier, req.message, session.greeting_streak) full_text_parts.append(short_reply) yield _sse({"type": "token", "text": short_reply}) audio_event = await _synthesise_and_emit(short_reply) if audio_event: yield audio_event else: _start_groq_stream_thread(inference_params, q, loop) try: if short_circuit: pass # skip Groq loop while not short_circuit: event_type, data = await q.get() # ── Groq stream error: attempt Fireworks non-stream fallback ── if event_type == "error": if fireworks is not None: logger.warning( "[stream] Groq stream error — activating Fireworks fallback" ) try: fb_params = {**inference_params, "model": FALLBACK_MODEL_FIREWORKS} fb_loop = asyncio.get_running_loop() fb_resp = await fb_loop.run_in_executor( None, lambda: fireworks.chat.completions.create(**fb_params), ) fallback_text = fb_resp.choices[0].message.content.strip() full_text_parts.append(fallback_text) yield _sse({"type": "token", "text": fallback_text}) # Dispatch TTS for each sentence in the fallback block fb_sentences, _ = _extract_sentences(fallback_text + " ") for sent in fb_sentences: audio_event = await _synthesise_and_emit(sent) if audio_event: yield audio_event except Exception as fb_exc: logger.error(f"[stream] Fireworks fallback failed: {fb_exc}") yield _sse({"type": "error", "detail": str(fb_exc)}) return else: yield _sse({"type": "error", "detail": data}) return break # ── Stream finished: flush any trailing buffer content ───────── elif event_type == "done": remainder = buffer.strip() if remainder: full_text_parts.append(remainder) yield _sse({"type": "token", "text": remainder}) audio_event = await _synthesise_and_emit(remainder) if audio_event: yield audio_event first_chunk_dispatched = True break # ── Normal token: accumulate + detect sentence boundaries ────── else: token: str = data full_text_parts.append(token) yield _sse({"type": "token", "text": token}) buffer += token # Issue 4: first chunk uses clause-level (comma) # splitting; every chunk after that uses full-sentence # boundaries only. if not first_chunk_dispatched: clause, buffer = _extract_first_clause(buffer) if clause: audio_event = await _synthesise_and_emit(clause) if audio_event: yield audio_event first_chunk_dispatched = True else: completed_sentences, buffer = _extract_sentences(buffer) for sentence in completed_sentences: audio_event = await _synthesise_and_emit(sentence) if audio_event: yield audio_event except asyncio.CancelledError: logger.warning( f"[stream] Client disconnected mid-stream: {session.session_id}" ) return # ── 6. Finalise session state ───────────────────────────────────────── full_text = "".join(full_text_parts).strip() session.add_message("assistant", full_text) elapsed_ms = int((time.time() - start_timestamp) * 1000) provider_label = "template-greeting" if short_circuit else primary_model logger.info( f"[stream] Complete. " f"ID={session.session_id} | Model={provider_label} " f"| Latency={elapsed_ms}ms | Lang={detected_language}" ) # Final metadata event — same shape as /chat JSON response yield _sse({ "type": "done", "session_id": session.session_id, "full_text": full_text, "language": detected_language, "language_locked": session.language_detector.is_locked(), "turn_count": session.turn_count, "caller_info": session.caller_info.to_dict(), "activeSummary": new_summary, "elapsed_ms": elapsed_ms, }) return StreamingResponse( event_generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", # Disable Nginx response buffering "Connection": "keep-alive", }, ) @app.post("/session/end") async def end_session(req: EndSessionRequest): """Closes down session trackers and outputs permanent call summaries.""" summary = manager.end_session(req.session_id) if not summary: return { "status": "already_ended", "summary": None, } return { "status": "ended", "summary": summary, } @app.get("/health") async def health(): """Runs data sweeps and returns systems heartbeat indicators.""" manager.cleanup_expired() neon_connected = False if neon_client._pool is not None: try: async with neon_client._pool.acquire() as conn: await conn.execute("SELECT 1") neon_connected = True except Exception: pass # Dynamically check RAG Space reachability instead of hardcoding True rag_ready = False RAG_SPACE_URL = os.getenv("RAG_SPACE_URL", "") if RAG_SPACE_URL: try: async with httpx.AsyncClient(timeout=2.0) as client: r = await client.get(f"{RAG_SPACE_URL}/health") rag_ready = r.status_code == 200 except Exception: pass return { "status": "ok", "version": "1.0.0", "active_sessions": manager.active_count(), "supported_languages": list(SUPPORTED_LANGUAGES), "rag_ready": rag_ready, "neon_connected": neon_connected } @app.get("/") async def root(): return { "service": "The Brain — Voice LLM Core Infrastructure Server Module", "status": "online", "documentation": "/docs" } @app.on_event("startup") async def startup_event(): await neon_client.init_db_pool() @app.on_event("shutdown") async def shutdown_event(): await neon_client.close_db_pool() @app.websocket("/ws/voice") async def websocket_voice(websocket: WebSocket): await websocket.accept() logger.info("New client WebSocket connection request to /ws/voice") try: start_message = await websocket.receive_json() is_setup = start_message.get("event") == "setup" is_session_start = start_message.get("type") == "session_start" if not is_setup and not is_session_start: logger.error("First WebSocket message was not type session_start or event setup") await websocket.send_json({"type": "error", "detail": "Session must be initialized with setup or session_start."}) await websocket.close() return session_id = start_message.get("session_id") if not session_id: logger.error("Handshake message missing session_id") await websocket.send_json({"type": "error", "detail": "session_id is required."}) await websocket.close() return stt_client = SttClient() if is_setup: company_id = start_message.get("company_id") line_token = start_message.get("line_token") if not company_id: logger.error("Setup handshake missing company_id") await websocket.send_json({"type": "error", "detail": "company_id is required."}) await websocket.close() return # Create a placeholder session first to construct the orchestrator session = ConversationSession(session_id, "Loading...", "Loading...") orchestrator = VoiceOrchestratorSession(websocket, session, stt_client) orchestrator.is_configured = False # Fire database fetch and STT pre-connect concurrently config = await neon_client.fetch_tenant_config(company_id, line_token) if not config: logger.error(f"Tenant config lookup failed: company_id={company_id}, line_token={line_token}") await websocket.send_json({"type": "error", "detail": "Company or line not found or inactive."}) await stt_client.close() await websocket.close() return # Verify line token matching if present in phone line constraints if line_token and config.get("line_token") != line_token: logger.error(f"Line token mismatch: company_id={company_id}") await websocket.send_json({"type": "error", "detail": "Invalid line token configuration."}) await stt_client.close() await websocket.close() return # Hydrate session from database configuration session = ConversationSession.from_tenant_config(config, session_id) manager.sessions[session_id] = session orchestrator.session = session orchestrator.is_configured = True await websocket.send_json({ "type": "configured", "company_id": company_id, "agent_name": session.agent_name }) else: company_name = start_message.get("company_name", "Our Company") agent_name = start_message.get("agent_name", "Sara") agent_gender = start_message.get("agent_gender", "male") session = manager.get_session(session_id) if not session: session = manager.create_session( company_name=company_name, agent_name=agent_name, agent_gender=agent_gender, session_id=session_id ) orchestrator = VoiceOrchestratorSession(websocket, session, stt_client) orchestrator.is_configured = True # 4. Start background loops await orchestrator.start() # 5. Enter ingress loop to process client messages await orchestrator.run_ingress_loop() except WebSocketDisconnect: logger.info("Client WebSocket disconnected from /ws/voice") except Exception as e: logger.exception(f"Unhandled error in voice WebSocket connection: {e}") try: await websocket.close() except Exception: pass