app.py

import json
import os

import gradio as gr
import requests
import spaces
from dotenv import load_dotenv
from groq import Groq

load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env"))
load_dotenv()

MODEL = "llama-3.3-70b-versatile"


@spaces.GPU
def _zerogpu_noop():
    """Unused — satisfies the ZeroGPU hardware startup check. This app is CPU-only."""
    return None

_client = None


def get_client() -> Groq:
    global _client
    if _client is None:
        _client = Groq(api_key=os.environ.get("GROQ_API_KEY"))
    return _client

WEATHER_CODES = {
    0: "clear sky", 1: "mostly clear", 2: "partly cloudy", 3: "cloudy",
    45: "fog", 48: "depositing rime fog",
    51: "light drizzle", 53: "drizzle", 55: "dense drizzle",
    61: "light rain", 63: "rain", 65: "heavy rain",
    71: "light snow", 73: "snow", 75: "heavy snow",
    80: "light showers", 81: "showers", 82: "violent showers",
    95: "thunderstorm", 96: "thunderstorm w/ hail", 99: "severe thunderstorm w/ hail",
}


def get_weather(city: str) -> dict:
    """Given a city name, geocode it and fetch current weather from Open-Meteo (no API key needed)."""
    geo = requests.get(
        "https://geocoding-api.open-meteo.com/v1/search",
        params={"name": city, "count": 1, "language": "en"},
        timeout=10,
    ).json()
    results = geo.get("results")
    if not results:
        return {"error": f"City '{city}' not found."}
    place = results[0]
    lat, lon = place["latitude"], place["longitude"]

    weather = requests.get(
        "https://api.open-meteo.com/v1/forecast",
        params={"latitude": lat, "longitude": lon, "current_weather": True},
        timeout=10,
    ).json()
    cw = weather["current_weather"]
    code = cw.get("weathercode")

    return {
        "city": place.get("name", city),
        "country": place.get("country", ""),
        "temp_c": cw["temperature"],
        "sky": WEATHER_CODES.get(code, "unknown"),
        "wind_kmh": cw.get("windspeed"),
    }


def convert_temperature(value: float, to_unit: str) -> dict:
    """Convert a temperature value to 'F' (Fahrenheit) or 'C' (Celsius). Input value is assumed Celsius when converting to F, and Fahrenheit when converting to C."""
    to_unit = to_unit.upper()
    if to_unit == "F":
        result = value * 9 / 5 + 32
    elif to_unit == "C":
        result = (value - 32) * 5 / 9
    else:
        return {"error": f"Unsupported unit '{to_unit}'. Use 'C' or 'F'."}
    return {"value": round(result, 1), "unit": to_unit}


TOOLS_SCHEMA = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get the current weather (temperature in Celsius, sky condition, wind) for a city, using the free Open-Meteo API.",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name, e.g. 'Ankara' or 'London'"},
                },
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "convert_temperature",
            "description": "Convert a numeric temperature value between Celsius and Fahrenheit.",
            "parameters": {
                "type": "object",
                "properties": {
                    "value": {"type": "number", "description": "The temperature value to convert"},
                    "to_unit": {"type": "string", "enum": ["C", "F"], "description": "Target unit: 'C' or 'F'"},
                },
                "required": ["value", "to_unit"],
            },
        },
    },
]

AVAILABLE_FUNCTIONS = {
    "get_weather": get_weather,
    "convert_temperature": convert_temperature,
}

SYSTEM_PROMPT = (
    "You are a helpful assistant with access to tools: get_weather and convert_temperature. "
    "Use them whenever the user's question needs live weather data or unit conversion. "
    "When converting, pass the exact temp_c value a tool returned, never a rounded or guessed one. "
    "Call tools as needed, then give a final, direct answer."
)


def run_agent(user_message: str, history: list):
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    for turn in history:
        messages.append({"role": turn["role"], "content": turn["content"]})
    messages.append({"role": "user", "content": user_message})

    trace_lines = []
    turn_num = 1
    max_turns = 6

    while turn_num <= max_turns:
        for attempt in range(3):
            try:
                response = get_client().chat.completions.create(
                    model=MODEL,
                    messages=messages,
                    tools=TOOLS_SCHEMA,
                    tool_choice="auto",
                )
                break
            except Exception as e:
                if "tool_use_failed" in str(e) and attempt < 2:
                    continue
                raise
        msg = response.choices[0].message

        if not msg.tool_calls:
            final_text = msg.content or ""
            if trace_lines:
                trace = "\n".join(trace_lines)
                return f"```\n{trace}\n```\n\n**Yanıt:**\n{final_text}"
            return final_text

        messages.append(
            {
                "role": "assistant",
                "content": msg.content,
                "tool_calls": [
                    {
                        "id": tc.id,
                        "type": "function",
                        "function": {"name": tc.function.name, "arguments": tc.function.arguments},
                    }
                    for tc in msg.tool_calls
                ],
            }
        )

        trace_lines.append(f"[Tur {turn_num}] Araç Çağrıları:")
        for tc in msg.tool_calls:
            name = tc.function.name
            args = json.loads(tc.function.arguments)
            func = AVAILABLE_FUNCTIONS.get(name)
            result = func(**args) if func else {"error": f"Unknown tool {name}"}

            args_str = ", ".join(f"{k}={v!r}" for k, v in args.items())
            trace_lines.append(f"   -> {name}({args_str})")
            trace_lines.append(f"   <- {result}")

            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "name": name,
                    "content": json.dumps(result, ensure_ascii=False),
                }
            )

        turn_num += 1

    return "Üzgünüm, çok fazla araç çağrısı yapıldı, yanıt üretilemedi."


def chat_fn(message, history):
    if not os.environ.get("GROQ_API_KEY"):
        return "GROQ_API_KEY ortam değişkeni ayarlanmamış. Lütfen Space secrets kısmına ekleyin."
    try:
        return run_agent(message, history)
    except Exception as e:
        return f"Hata: {e}"


demo = gr.ChatInterface(
    fn=chat_fn,
    title="🛠️ Tool Calling Demo — Open-Meteo + Groq",
    description=(
        "Model, hava durumu sorularında `get_weather` ve birim çevirisi gerektiğinde "
        "`convert_temperature` araçlarını otomatik çağırır. Arka planda hangi araçların "
        "hangi parametrelerle çağrıldığı yanıtın üstünde gösterilir.\n\n"
        "Örnek: *'Ankara mı daha sıcak Londra mı, ve bu değerler Fahrenheit olarak kaç eder?'*"
    ),
    examples=[
        "Ankara mı daha sıcak Londra mı, ve bu değerler Fahrenheit olarak kaç eder?",
        "İstanbul'da hava nasıl?",
        "Tokyo'daki sıcaklık kaç Fahrenheit?",
    ],
)

if __name__ == "__main__":
    demo.launch()