Create app.py

app.py

@@ -0,0 +1,225 @@
+import gradio as gr
+import os
+import json
+import time
+from typing import Any, Dict, List
+from openai import OpenAI
+from dotenv import load_dotenv
+load_dotenv()
+
+HF_TOKEN = os.getenv("HF_TOKEN")
+
+# Available models for selection
+AVAILABLE_MODELS = [
+    "openai/gpt-oss-120b:fireworks-ai",
+    "openai/gpt-oss-20b:fireworks-ai"
+]
+
+# Default model
+DEFAULT_MODEL = "openai/gpt-oss-120b:fireworks-ai"
+BASE_URL = "https://router.huggingface.co/v1"
+
+client = OpenAI(base_url=BASE_URL, api_key=HF_TOKEN)
+
+# OpenAI-style tool specs for function calling
+TOOLS = [
+{
+  "type": "function",
+  "function": {
+    "name": "get_weather",
+    "description": "Get the current weather information for a specified city",
+    "parameters": {
+      "type": "object",
+      "properties": {
+        "city": {
+          "type": "string",
+          "description": "The name of the city to get weather information for"
+        }
+      },
+      "required": ["city"]
+    }
+  }
+}
+]
+
+def get_weather(city: str):
+    print(f"[debug] getting weather for {city}")
+    return f"The weather in {city} is sunny."
+
+FUNCTION_MAP = {
+    "get_weather": get_weather,
+}
+
+def call_model(messages: List[Dict[str, str]], tools=None, temperature: float = 0.3, model: str = DEFAULT_MODEL):
+    """One step with tool calling support."""
+    if tools is None:
+        tools = TOOLS
+    try:
+        return client.chat.completions.create(
+            model=model,
+            temperature=temperature,
+            messages=messages,
+            tools=tools,
+            tool_choice="auto"
+        )
+    except Exception as e:
+        print(f"Error calling model: {e}")
+        raise
+
+def run_weather_agent(user_prompt: str, model: str = DEFAULT_MODEL) -> str:
+    """
+    High level prompt for a weather agent.
+    It gets weather information for cities and provides responses.
+    """
+    system = {
+        "role": "system",
+        "content": (
+            "You are a helpful weather agent. Follow these steps:\n"
+            "1. When a user asks about weather in a city, use get_weather tool\n"
+            "2. Provide a friendly response with the weather information\n"
+            "3. If no city is mentioned, ask the user to specify a city\n"
+            "4. Be conversational and helpful\n"
+        ),
+    }
+
+    messages: List[Dict[str, str]] = [system, {"role": "user", "content": user_prompt}]
+
+    for step in range(3):  # small safety cap
+        try:
+            resp = call_model(messages, tools=TOOLS, model=model)
+            msg = resp.choices[0].message
+
+            # If the model wants to call tools
+            if getattr(msg, "tool_calls", None) and msg.tool_calls:
+                # Add the assistant message with tool calls to the conversation
+                assistant_message = {
+                    "role": "assistant",
+                    "content": msg.content or "",
+                    "tool_calls": [
+                        {
+                            "id": tool_call.id,
+                            "type": "function",
+                            "function": {
+                                "name": tool_call.function.name,
+                                "arguments": tool_call.function.arguments
+                            }
+                        }
+                        for tool_call in msg.tool_calls
+                    ]
+                }
+                messages.append(assistant_message)
+                
+                # Process each tool call
+                for tool_call in msg.tool_calls:
+                    name = tool_call.function.name
+                    args = {}
+                    try:
+                        args = json.loads(tool_call.function.arguments or "{}")
+                    except json.JSONDecodeError:
+                        args = {}
+
+                    fn = FUNCTION_MAP.get(name)
+                    if not fn:
+                        messages.append({
+                            "role": "tool",
+                            "tool_call_id": tool_call.id,
+                            "name": name,
+                            "content": json.dumps({"ok": False, "error": "unknown_tool"})
+                        })
+                        continue
+
+                    try:
+                        result = fn(**args)
+                    except TypeError as e:
+                        result = {"ok": False, "error": f"bad_args: {e}"}
+                    except Exception as e:
+                        result = {"ok": False, "error": repr(e)}
+
+                    tool_response = {
+                        "role": "tool",
+                        "tool_call_id": tool_call.id,
+                        "name": name,
+                        "content": json.dumps(result),
+                    }
+                    messages.append(tool_response)
+                
+                # Continue loop so the model can see tool outputs
+                continue
+
+            # If we have a final assistant message without tool calls
+            if msg.content:
+                return msg.content
+
+            # Fallback tiny sleep then continue
+            time.sleep(0.2)
+            
+        except Exception as e:
+            # If there's an error, try to continue or return error message
+            if step == 2:  # Last step
+                return f"Error occurred during processing: {e}"
+            time.sleep(0.5)
+            continue
+
+    return "I could not complete the task within the step limit. Try refining your query."
+
+# Example usage of the weather agent
+# if __name__ == "__main__":
+#     # Test the weather agent with different queries
+#     test_queries = [
+#         "What's the weather like in New York?",
+#         "How's the weather in London?",
+#         "Tell me about the weather in Tokyo",
+#         "What's the weather like?"  # This should prompt for a city
+#     ]
+    
+#     print("=== Weather Agent Demo ===\n")
+    
+#     for query in test_queries:
+#         print(f"User: {query}")
+#         try:
+#             response = call_model(messages=[{"role": "user", "content": query}]) # Assuming run_weather_agent is removed or replaced
+#             print(f"Agent: {response}\n")
+#         except Exception as e:
+#             print(f"Error: {e}\n")
+#         print("-" * 50 + "\n")
+
+
+### GRADIO
+def weather_chat_with_agent(message, history, model):
+    """Handle weather chat messages and return agent responses."""
+    if not message.strip():
+        return history
+    
+    try:
+        response = run_weather_agent(message, model)
+        
+        history.append({"role": "user", "content": message})
+        history.append({"role": "assistant", "content": response})
+        
+        return history
+    except Exception as e:
+        error_msg = f"Sorry, I encountered an error: {str(e)}"
+        history.append({"role": "user", "content": message})
+        history.append({"role": "assistant", "content": error_msg})
+        return history
+
+def create_weather_interface():
+    with gr.Blocks(title="Weather Agent") as demo:
+        gr.Markdown("# 🌤️ Weather Agent")
+        gr.Markdown("Ask me about the weather in any city!")
+        
+        chatbot = gr.Chatbot(height=400, type="messages")
+        msg = gr.Textbox(label="Ask about weather", placeholder="e.g., What's the weather like in Paris?")
+        clear = gr.Button("Clear")
+        
+        def respond(message, chat_history):
+            return weather_chat_with_agent(message, chat_history, DEFAULT_MODEL)
+        
+        msg.submit(respond, [msg, chatbot], [chatbot])
+        clear.click(lambda: None, None, chatbot, queue=False)
+    
+    return demo
+
+# To run the weather interface:
+demo = create_weather_interface()
+demo.launch()