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()