Spaces:
Running
Running
| 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() | |