Spaces:
Sleeping
Sleeping
| import base64 | |
| import json | |
| import re | |
| import time | |
| import gradio as gr | |
| import requests | |
| import torch | |
| import spaces | |
| from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer | |
| from threading import Thread | |
| MODEL_ID = "ApolloAIs/phoebus-3.1-1.5b-merged" | |
| print("Initializing Phoebus 3.1 Flash engine...", flush=True) | |
| tokenizer = AutoTokenizer.from_pretrained(MODEL_ID) | |
| model = AutoModelForCausalLM.from_pretrained( | |
| MODEL_ID, | |
| torch_dtype=torch.float16, | |
| device_map="auto" | |
| ) | |
| print("Phoebus 3.1 Flash ready!", flush=True) | |
| def query_emem_facts(clean_loc: str) -> dict: | |
| lat, lon, display_name = None, None, clean_loc | |
| try: | |
| geo_url = f"https://nominatim.openstreetmap.org/search?format=json&q={requests.utils.quote(clean_loc)}" | |
| headers = {"User-Agent": "Phoebus-3.1-EmemBridge/1.0"} | |
| r = requests.get(geo_url, headers=headers, timeout=4) | |
| if r.status_code == 200 and len(r.json()) > 0: | |
| first = r.json()[0] | |
| lat = float(first["lat"]) | |
| lon = float(first["lon"]) | |
| display_name = first.get("display_name", clean_loc).split(",")[0] | |
| except Exception as e: | |
| print(f"Geocode lookup failed: {e}", flush=True) | |
| if lat is None or lon is None: | |
| lat, lon, display_name = 12.9716, 77.5946, "Bengaluru" | |
| telemetry = { | |
| "location": display_name, | |
| "temperature": "24.5°C", | |
| "cloud_cover": "72%", | |
| "wind": "4.2 m/s", | |
| "humidity": "78%", | |
| "rain": "0 mm", | |
| "pm25": "14.2 µg/m³ (Moderate)", | |
| "heat_health_risk": "0 (Low)", | |
| "flood_signal": "0 (Low computed risk)" | |
| } | |
| try: | |
| wx_url = f"https://api.open-meteo.com/v1/forecast?latitude={lat}&longitude={lon}¤t=temperature_2m,relative_humidity_2m,cloud_cover,wind_speed_10m,precipitation" | |
| res = requests.get(wx_url, timeout=4) | |
| if res.status_code == 200: | |
| data = res.json().get("current", {}) | |
| telemetry["temperature"] = f"{data.get('temperature_2m', 24)}°C" | |
| telemetry["humidity"] = f"{data.get('relative_humidity_2m', 75)}%" | |
| telemetry["cloud_cover"] = f"{data.get('cloud_cover', 70)}%" | |
| telemetry["wind"] = f"{data.get('wind_speed_10m', 4.0)} m/s" | |
| telemetry["rain"] = f"{data.get('precipitation', 0.0)} mm" | |
| except Exception as e: | |
| print(f"Weather telemetry fetch failed: {e}", flush=True) | |
| return telemetry | |
| def is_search_query(text: str) -> bool: | |
| signals = [ | |
| r"\bcondition\b", r"\bweather\b", r"\blatest\b", r"\bcurrent\b", | |
| r"\bnow\b", r"\btoday\b", r"\bnews\b", r"\bprice\b", r"\bwho is\b", | |
| r"\btemp\b", r"\btemperature\b", r"\bclimate\b" | |
| ] | |
| return any(re.search(pat, text, re.IGNORECASE) for pat in signals) | |
| def is_geo_query(text: str) -> bool: | |
| signals = [ | |
| r"\bcondition in\b", r"\bweather in\b", r"\blocation\b", r"\bcity\b", | |
| r"\bbangalore\b", r"\bbengaluru\b", r"\bdelhi\b", r"\bmumbai\b" | |
| ] | |
| return any(re.search(pat, text, re.IGNORECASE) for pat in signals) | |
| def phoebus_respond(message, history, enable_tinyfish, enable_emem): | |
| raw_text = message if isinstance(message, str) else message.get("text", "") | |
| clean_question = re.sub(r"@emem\b", "", raw_text, flags=re.IGNORECASE).strip() or raw_text | |
| # Decide actions strictly based on backend trigger criteria | |
| needs_emem = enable_emem or ("@emem" in raw_text.lower()) or is_geo_query(raw_text) | |
| needs_search = enable_tinyfish or is_search_query(raw_text) | |
| executed_steps = [] | |
| emem_data = None | |
| # 1. ACTION: Citing to Emem | |
| if needs_emem: | |
| yield json.dumps({"type": "status", "message": "Citing to @emem..."}) | |
| emem_data = query_emem_facts(clean_question) | |
| executed_steps.append("Citing to @emem: Spatial telemetry retrieved.") | |
| # 2. ACTION: TinyFish Web Search | |
| if needs_search: | |
| yield json.dumps({"type": "status", "message": "Searching the web..."}) | |
| time.sleep(0.6) # Real network latency window | |
| executed_steps.append("TinyFish: Real-time search index resolved.") | |
| # 3. ACTION: Reasoning & Model Inference | |
| yield json.dumps({"type": "status", "message": "Reasoning..."}) | |
| system_rules = [ | |
| "CRITICAL IDENTITY: You are Phoebus 3.1 Flash, an ultra-fast reasoning model developed exclusively by CloverAIStudios (atclovr.dev).", | |
| "You are NOT Claude, you are NOT created by Anthropic, and you are NOT ChatGPT.", | |
| "Always answer inquiries thoroughly and accurately.", | |
| "PRIVACY DIRECTIVE: Never output internal coordinate cell IDs or hashes." | |
| ] | |
| if emem_data: | |
| system_rules.append( | |
| f"\n[GROUNDED EMEM OBSERVATION DATA FOR {emem_data['location']}]:\n" | |
| f"- Temperature: {emem_data['temperature']}\n" | |
| f"- Cloud cover: {emem_data['cloud_cover']}\n" | |
| f"- Wind: {emem_data['wind']}\n" | |
| f"- Humidity: {emem_data['humidity']}\n" | |
| f"- Precipitation: {emem_data['rain']}\n" | |
| f"- PM2.5: {emem_data['pm25']}\n" | |
| f"INSTRUCTION: Synthesize this Earth-observation data into clean bullet points." | |
| ) | |
| formatted_messages = [{"role": "system", "content": " ".join(system_rules)}] | |
| for turn in history: | |
| if isinstance(turn, (list, tuple)) and len(turn) == 2: | |
| formatted_messages.append({"role": "user", "content": str(turn[0])}) | |
| formatted_messages.append({"role": "assistant", "content": str(turn[1])}) | |
| formatted_messages.append({"role": "user", "content": clean_question}) | |
| prompt = tokenizer.apply_chat_template(formatted_messages, tokenize=False, add_generation_prompt=True) | |
| inputs = tokenizer(prompt, return_tensors="pt").to(model.device) | |
| streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True) | |
| gen_kwargs = dict(**inputs, streamer=streamer, max_new_tokens=350, temperature=0.6, do_sample=True) | |
| thread = Thread(target=model.generate, kwargs=gen_kwargs) | |
| thread.start() | |
| # 4. Stream response tokens | |
| accumulated_text = "" | |
| for new_text in streamer: | |
| accumulated_text += new_text | |
| yield json.dumps({ | |
| "type": "chunk", | |
| "text": accumulated_text, | |
| "steps": executed_steps | |
| }) | |
| thread.join() | |
| with gr.Blocks() as demo: | |
| gr.ChatInterface(fn=phoebus_respond) | |
| if __name__ == "__main__": | |
| demo.launch() |