Blame for app.py

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}&current=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)

@spaces.GPU(duration=5)
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()