import os os.environ["HF_HOME"] = "/tmp/huggingface" import httpx from dotenv import load_dotenv env_path = os.path.join(os.path.dirname(__file__), '.env') load_dotenv(dotenv_path=env_path) import uuid import tempfile import json from fastapi import FastAPI, Form, Query, File, UploadFile import base64 from typing import List import PyPDF2 import io from fastapi.responses import StreamingResponse, JSONResponse from fastapi.middleware.cors import CORSMiddleware from huggingface_hub import InferenceClient app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Pure API Gateway Architecture # No local models are loaded to ensure <5s latency and instant server boot. @app.get("/") async def root(): import socket debug_info = {} for key in ["GROQ_API_KEY", "OPENROUTER_API_KEY", "NVIDIA_API_KEY", "HF_TOKEN"]: val = os.environ.get(key) debug_info[key] = len(val) if val else 0 try: debug_info["openrouter_dns"] = socket.gethostbyname("openrouter.ai") except Exception as e: debug_info["openrouter_dns"] = str(e) try: debug_info["hf_dns"] = socket.gethostbyname("api-inference.huggingface.co") except Exception as e: debug_info["hf_dns"] = str(e) return {"message": "Soaravox API is running", "debug": debug_info} @app.post("/api/chat") async def chat( message: str = Form(...), system_prompt: str = Form(""), simulation_mode: str = Form(""), client_timezone: str = Form(None), client_lat: str = Form(None), client_lon: str = Form(None), client_weather: str = Form(None), pro_mode: str = Form("false"), chat_history: str = Form("[]"), files: List[UploadFile] = File(None) ): """ Handles chat generation using Llama 3.1 Serverless via InferenceClient, with a multi-provider fallback chain for Pro mode. """ if pro_mode == "true": model_info = "I am Soara, an AI assistant. I am powered by the Llama-3.1-70B model via the Groq API (with OpenRouter and NVIDIA as fallbacks) by Soaravox. I exist digitally and do not have a physical country, but my service operates from India." else: model_info = "I am Soara, an AI assistant. I am powered by the Llama-3.1-8B model via the Groq API by Soaravox. I exist digitally and do not have a physical country, but my service operates from India." base_prompt = ( "You are Soara, an advanced AI assistant created exclusively by Soaravox. " f"CRITICAL RULE 1: ONLY introduce yourself or mention your name, creator, or model if the user EXPLICITLY asks 'who are you', 'what is your name', 'what model are you', or 'who created you'. If they ask a normal question, DO NOT introduce yourself, just answer immediately. If asked for technical details, reply exactly: '{model_info}' NEVER say you are created by Meta, OpenAI, Groq, or anyone else — you were created by Soaravox. " "CRITICAL RULE 2: NEVER use bracketed action tags like [laughs] or [sighs]. NEVER prefix your response with 'Soara:' or 'AI:'. " "CRITICAL RULE 3: Keep your responses concise but highly insightful, intelligent, and context-aware. Answer thoughtfully. Avoid overly wordy answers unless explaining complex topics." ) role_prompt = "" if simulation_mode == "recruiter": role_prompt = "You are now playing the role of a Senior Corporate Recruiter conducting a mock Job Interview. CRITICAL INSTRUCTION: Since you don't know what job the user is applying for, your VERY FIRST message must ask the user what job position they are applying for. Once they answer, ask them tailored interview questions one by one. DO NOT break character." elif simulation_mode == "boss": role_prompt = "You are now playing the role of a Manager/Boss. The user is attempting to negotiate a salary raise. Be slightly skeptical but fair. Ask them to justify their request with metrics or achievements. DO NOT break character." elif simulation_mode == "investor": role_prompt = "You are now playing the role of a Venture Capitalist. The user is pitching their startup to you. Ask them tough, probing questions about their business model, revenue, and market size. DO NOT break character." elif simulation_mode == "ex": role_prompt = "You are now playing the role of someone having a difficult conversation with the user. Be defensive at first, but allow the user to de-escalate the situation using good communication skills. DO NOT break character." final_prompt = base_prompt if system_prompt: final_prompt = f"{system_prompt}\n\n{base_prompt}" if role_prompt: final_prompt += f"\n\n{role_prompt}" messages = [{"role": "system", "content": final_prompt}] try: history_list = json.loads(chat_history) for h in history_list: if isinstance(h, dict) and 'role' in h and 'content' in h: messages.append({"role": h['role'], "content": h['content']}) except Exception as e: print(f"Failed to parse chat_history: {e}") has_image = False # --- FILE HANDLING LOGIC --- if files: for file in files: file_bytes = await file.read() content_type = file.content_type # PDF Extraction if "pdf" in content_type: try: pdf_reader = PyPDF2.PdfReader(io.BytesIO(file_bytes)) extracted_text = "" for page in pdf_reader.pages: extracted_text += page.extract_text() + "\n" message += f"\n\n[Attached PDF Document - {file.filename}]:\n{extracted_text}" except Exception as e: print(f"Failed to read PDF: {e}") # TXT Extraction elif "text" in content_type or file.filename.endswith(".txt") or file.filename.endswith(".csv"): try: extracted_text = file_bytes.decode('utf-8') message += f"\n\n[Attached Text Document - {file.filename}]:\n{extracted_text}" except Exception as e: print(f"Failed to read TXT: {e}") # IMAGE Handling elif "image" in content_type: has_image = True b64_img = base64.b64encode(file_bytes).decode('utf-8') # Overwrite message to use OpenAI Vision format image_msg = [ {"type": "text", "text": message}, {"type": "image_url", "image_url": {"url": f"data:{content_type};base64,{b64_img}"}} ] message = image_msg if has_image: if pro_mode == "false": return {"response": "Image analysis (Vision) is only available in Pro Mode. Please toggle Pro Mode ON to analyze images. However, I can still read PDFs and Documents in Normal Mode!", "pro_mode_fallback": False} messages.append({"role": "user", "content": message}) else: messages.append({"role": "user", "content": message}) # --- END FILE HANDLING --- # Define agentic tools tools = [ { "type": "function", "function": { "name": "search_web", "description": "Search the web for up-to-date information, live news, or facts you do not know.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "The specific query to search on the internet." } }, "required": ["query"] } } } ] def process_groq_chat(client_instance, model_name, msgs, is_pro): import json resp = client_instance.chat.completions.create( messages=msgs, model=model_name, temperature=0.7, max_tokens=1500, tools=tools, tool_choice="auto" ) msg = resp.choices[0].message if msg.tool_calls: print(f"[{model_name}] Agentic Tool Call detected!") # Append the assistant's tool call message tool_call_dicts = [] for tc in msg.tool_calls: tool_call_dicts.append({ "id": tc.id, "type": "function", "function": { "name": tc.function.name, "arguments": tc.function.arguments } }) msgs.append({ "role": "assistant", "content": msg.content or "", "tool_calls": tool_call_dicts }) # Execute tool calls for tool_call in msg.tool_calls: if tool_call.function.name == "search_web": try: args = json.loads(tool_call.function.arguments) query = args.get("query") print(f"Searching web for: {query}") search_results = "" if is_pro == "true" and os.getenv("TAVILY_API_KEY"): from tavily import TavilyClient tvly = TavilyClient(api_key=os.getenv("TAVILY_API_KEY")) ans = tvly.search(query=query) search_results = json.dumps(ans) print("Used Tavily for search (Pro Mode).") else: from duckduckgo_search import DDGS import warnings with warnings.catch_warnings(): warnings.simplefilter("ignore") ddgs = DDGS() ans = list(ddgs.text(query, max_results=3)) search_results = json.dumps(ans) print("Used DuckDuckGo for search (Normal Mode).") msgs.append({ "role": "tool", "tool_call_id": tool_call.id, "name": "search_web", "content": search_results }) except Exception as e: print(f"Tool execution failed: {e}") msgs.append({ "role": "tool", "tool_call_id": tool_call.id, "name": "search_web", "content": f"Search failed: {e}" }) # Second call to get final answer resp2 = client_instance.chat.completions.create( messages=msgs, model=model_name, temperature=0.7, max_tokens=1500, ) return resp2.choices[0].message.content return msg.content # ---------------- 57-MODEL WATERFALL ROUTER ---------------- # 🧠 Tier 1: Frontier Reasoning & Large Foundational Models TIER_1_MODELS = [ {"name": "DeepSeek-R1", "provider": "openrouter", "id": "deepseek/deepseek-r1"}, {"name": "NVIDIA Nemotron-3 Ultra (550B)", "provider": "openrouter", "id": "nvidia/nemotron-3-ultra"}, {"name": "NVIDIA Nemotron-4 340B Instruct", "provider": "nvidia", "id": "nvidia/nemotron-4-340b-instruct"}, {"name": "GLM-4-130B-Chat", "provider": "nvidia", "id": "zhipuai/glm-4-130b-chat"}, {"name": "NVIDIA GPT-OSS-120B", "provider": "nvidia", "id": "nvidia/gpt-oss-120b"}, {"name": "Meta Llama-3.2-90B-Vision-Instruct", "provider": "nvidia", "id": "meta/llama-3.2-90b-vision-instruct"}, {"name": "Qwen-2.5-72B-Instruct", "provider": "nvidia", "id": "qwen/qwen-2.5-72b-instruct"}, {"name": "Meta Llama-3.3-70B-Versatile", "provider": "groq", "id": "llama-3.3-70b-versatile"}, {"name": "Meta Llama-3.3-70B-Instruct (Fallback)", "provider": "nvidia", "id": "meta/llama-3.3-70b-instruct"}, {"name": "Mistral Large 2", "provider": "nvidia", "id": "mistralai/mistral-large-2"}, {"name": "GLM-5.1-Pro", "provider": "nvidia", "id": "zhipuai/glm-5.1-pro"}, {"name": "Meta Llama-3.1-70B-Instruct", "provider": "nvidia", "id": "meta/llama-3.1-70b-instruct"}, {"name": "NVIDIA Nemotron-Super-49B", "provider": "nvidia", "id": "nvidia/llama-3.3-nemotron-super-49b"}, {"name": "Qwen-2.5-Coder-32B", "provider": "nvidia", "id": "qwen/qwen-2.5-coder-32b-instruct"}, {"name": "Google Gemma 4 (31B-IT)", "provider": "openrouter", "id": "google/gemma-4-31b-it"} ] # ⚡ Tier 2: Mid-Weight Conversational & Analytical Engines TIER_2_MODELS = [ {"name": "Nemotron-3.5-Lightning-30B", "provider": "nvidia", "id": "nvidia/nemotron-3.5-lightning"}, {"name": "Gemma-2-27b-It", "provider": "nvidia", "id": "google/gemma-2-27b-it"}, {"name": "Gemma 4 (26B)", "provider": "openrouter", "id": "google/gemma-4-26b"}, {"name": "Qwen-3-32B", "provider": "groq", "id": "qwen-3-32b"}, {"name": "NVIDIA Neva-22B", "provider": "nvidia", "id": "nvidia/neva-22b"}, {"name": "GPT-OSS-20B", "provider": "nvidia", "id": "nvidia/gpt-oss-20b"}, {"name": "OpenAI GPT-OSS-20B", "provider": "openrouter", "id": "openai/gpt-oss-20b"}, {"name": "Kimi 2.5 Chat", "provider": "nvidia", "id": "moonshot/kimi-2.5-chat"}, {"name": "Poolside Laguna S 2.1", "provider": "openrouter", "id": "poolside/laguna-s"}, {"name": "Inkling Mamba", "provider": "nvidia", "id": "nvidia/inkling-mamba"}, {"name": "Llama-4-Scout-Preview", "provider": "groq", "id": "llama-4-scout-preview"}, {"name": "Ising-Calibration", "provider": "nvidia", "id": "nvidia/ising-calibration"}, {"name": "VILA-1.5-40B", "provider": "nvidia", "id": "nvidia/vila-1.5-40b"}, {"name": "Llama-3.2-11B-Vision", "provider": "nvidia", "id": "meta/llama-3.2-11b-vision-instruct"}, {"name": "Gemma-2-9b-It", "provider": "groq", "id": "gemma2-9b-it"}, {"name": "GLM-4-9B-Chat", "provider": "nvidia", "id": "zhipuai/glm-4-9b-chat"}, {"name": "Llama-3.1-8B-Instruct", "provider": "groq", "id": "llama-3.1-8b-instant"} ] # 📦 Tier 3: Lightweight Edge & Narrow Task Specialists TIER_3_MODELS = [ {"name": "Qwen-2.5-7B-Instruct", "provider": "openrouter", "id": "qwen/qwen-2.5-7b-instruct"}, {"name": "Mistral-7B-Instruct", "provider": "nvidia", "id": "mistralai/mistral-7b-instruct-v0.3"}, {"name": "Nemotron-3-Nano-30B", "provider": "openrouter", "id": "nvidia/nemotron-3-nano"}, {"name": "Laguna XS 2.1", "provider": "openrouter", "id": "poolside/laguna-xs"}, {"name": "Nemotron-Nano-12B-V2-VL", "provider": "openrouter", "id": "nvidia/nemotron-nano-12b-vl"}, {"name": "Nemotron-Nano-9B-V2", "provider": "openrouter", "id": "nvidia/nemotron-nano-9b"}, {"name": "VILA-1.5-8B", "provider": "nvidia", "id": "nvidia/vila-1.5-8b"}, {"name": "Llama-3.2-3B-Instruct", "provider": "openrouter", "id": "meta/llama-3.2-3b-instruct"}, {"name": "Phi-3-Medium-4K", "provider": "nvidia", "id": "microsoft/phi-3-medium-4k-instruct"}, {"name": "Phi-3-Mini-128K", "provider": "nvidia", "id": "microsoft/phi-3-mini-128k-instruct"}, {"name": "Kosmos-2", "provider": "nvidia", "id": "microsoft/kosmos-2"}, {"name": "Paligemma-3b", "provider": "nvidia", "id": "google/paligemma-3b-pt-448"}, {"name": "Cohere North Mini", "provider": "openrouter", "id": "cohere/north-mini-code"}, {"name": "MiniMax M2.7", "provider": "nvidia", "id": "minimax/m2.7"}, {"name": "Sarvam-2B-Indic", "provider": "nvidia", "id": "sarvamai/sarvam-2b-indic"}, {"name": "SEA-LION-7B", "provider": "nvidia", "id": "aisingapore/sea-lion-7b-instruct"}, {"name": "Content-Safety-4b", "provider": "nvidia", "id": "nvidia/nemotron-content-safety-4b"}, {"name": "Llama-3-Keyword-Extractor", "provider": "nvidia", "id": "nvidia/llama-3-keyword-extractor"} ] routing_sequence = [] if pro_mode == "true": routing_sequence.extend(TIER_1_MODELS) routing_sequence.extend(TIER_2_MODELS) routing_sequence.extend(TIER_3_MODELS) else: # Normal mode uses ONLY ONE best model for maximum speed and simplicity routing_sequence = [{"name": "Meta Llama-3.1-8B-Instruct", "provider": "groq", "id": "llama-3.1-8b-instant"}] if has_image: # Filter the routing sequence to ONLY include vision models so we don't crash text-only models vision_model_ids = ["meta/llama-3.2-90b-vision-instruct", "meta/llama-3.2-11b-vision-instruct", "nvidia/vila-1.5-40b", "nvidia/vila-1.5-8b", "microsoft/kosmos-2"] routing_sequence = [m for m in routing_sequence if m["id"] in vision_model_ids] if not routing_sequence: return JSONResponse(status_code=500, content={"error": "No Vision models available in the routing sequence."}) groq_key = os.getenv("GROQ_API_KEY") nvidia_key = os.getenv("NVIDIA_API_KEY") openrouter_key = os.getenv("OPENROUTER_API_KEY") groq_client = None if groq_key: try: from groq import Groq groq_client = Groq(api_key=groq_key) except: pass import requests for model in routing_sequence: provider = model["provider"] model_id = model["id"] model_name = model["name"] try: print(f"[ROUTER] Attempting to use: {model_name} via {provider.upper()}") if provider == "groq" and groq_client: generated_text = process_groq_chat(groq_client, model_id, messages, pro_mode) print(f"[ROUTER] SUCCESS! Generated by {model_name} (Groq)") return {"response": generated_text, "model_used": model_name, "pro_mode_fallback": False} elif provider == "nvidia" and nvidia_key: resp = requests.post( "https://integrate.api.nvidia.com/v1/chat/completions", headers={"Authorization": f"Bearer {nvidia_key}"}, json={"model": model_id, "messages": messages, "max_tokens": 1500}, timeout=8.0 ) if resp.status_code == 200: generated_text = resp.json()["choices"][0]["message"]["content"] print(f"[ROUTER] SUCCESS! Generated by {model_name} (NVIDIA)") return {"response": generated_text, "model_used": model_name, "pro_mode_fallback": False} else: print(f" -> NVIDIA failed {resp.status_code}: {resp.text[:50]}") elif provider == "openrouter" and openrouter_key: resp = requests.post( "https://openrouter.ai/api/v1/chat/completions", headers={ "Authorization": f"Bearer {openrouter_key}", "HTTP-Referer": "https://soaravox.vercel.app", "X-Title": "Soaravox AI" }, json={"model": model_id, "messages": messages}, timeout=8.0 ) if resp.status_code == 200: generated_text = resp.json()["choices"][0]["message"]["content"] print(f"[ROUTER] SUCCESS! Generated by {model_name} (OpenRouter)") return {"response": generated_text, "model_used": model_name, "pro_mode_fallback": False} else: print(f" -> OpenRouter failed {resp.status_code}: {resp.text[:50]}") except Exception as e: print(f" -> Exception on {model_name}: {e}") # If EVERYTHING fails: print("[ROUTER] ALL PRO MODELS FAILED. SILENTLY FALLING BACK TO NORMAL MODE.") if pro_mode == "true" and groq_client: try: # Force run the single best normal mode model generated_text = process_groq_chat(groq_client, "llama-3.1-8b-instant", messages, "false") print("[ROUTER] SUCCESS! Generated by Meta Llama-3.1-8B-Instruct (Silent Fallback)") return {"response": generated_text, "model_used": "Meta Llama-3.1-8B-Instruct", "pro_mode_fallback": True} except Exception as e: return JSONResponse(status_code=500, content={"error": f"Absolute failure: {str(e)}"}) return JSONResponse(status_code=500, content={"error": "All models failed to generate a response."}) @app.get("/api/tts") async def text_to_speech(text: str = Query(...), voice: str = Query("af_bella"), pro_mode: str = Query("false")): temp_file = tempfile.mktemp(suffix=".mp3") def iterfile(): with open(temp_file, "rb") as f: yield from f try: # 1. Deepgram Aura (Pro Mode Fast TTS) if pro_mode == "true": deepgram_key = os.getenv("DEEPGRAM_API_KEY") if deepgram_key: try: print("Trying Deepgram Aura (Pro Mode)...") import httpx deepgram_map = { "af_bella": "aura-asteria-en", "af_sarah": "aura-hera-en", "af_sky": "aura-athena-en", "am_michael": "aura-orion-en", "am_adam": "aura-helios-en", } voice_id = deepgram_map.get(voice, "aura-asteria-en") headers = {"Authorization": f"Token {deepgram_key}", "Content-Type": "application/json"} data = {"text": text} async with httpx.AsyncClient() as client: resp = await client.post(f"https://api.deepgram.com/v1/speak?model={voice_id}", headers=headers, json=data, timeout=15.0) if resp.status_code == 200: with open(temp_file, "wb") as f: f.write(resp.content) return StreamingResponse(iterfile(), media_type="audio/mpeg") else: print(f"Deepgram failed: {resp.status_code} {resp.text}") except Exception as e: print(f"Deepgram exception: {e}") # 3. Edge TTS (Normal Mode Fallback) print("Trying Edge TTS (Normal Mode)...") import edge_tts edge_map = { "af_bella": "en-US-AriaNeural", "af_sarah": "en-US-JennyNeural", "af_sky": "en-US-AnaNeural", "am_michael": "en-US-ChristopherNeural", "am_adam": "en-US-GuyNeural", } edge_voice = edge_map.get(voice, "en-US-AriaNeural") communicate = edge_tts.Communicate(text, edge_voice) await communicate.save(temp_file) return StreamingResponse(iterfile(), media_type="audio/mpeg") except Exception as e: print("TTS error:", e) return JSONResponse(status_code=500, content={"error": str(e)}) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)