Create app.py

app.py

@@ -0,0 +1,296 @@
+import os
+import json
+import requests
+from functools import lru_cache
+import gradio as gr
+import openai
+
+# -------------------- CONFIGURATION --------------------
+# Use environment variables for security (set them before running)
+OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "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")
+OPENWEATHER_API_KEY = os.getenv("OPENWEATHER_API_KEY", "")  # Optional, for live weather
+
+client = openai.OpenAI(api_key=OPENAI_API_KEY)
+
+# Load attractions database
+with open("attractions_data.json", "r") as f:
+    ATTRACTIONS_DB = json.load(f)
+
+# -------------------- TOOL IMPLEMENTATIONS --------------------
+
+def get_weather(city: str) -> dict:
+    """Get current weather for a city (mock if no API key)."""
+    if not OPENWEATHER_API_KEY:
+        # Mock response
+        return {
+            "city": city,
+            "temperature": 22,
+            "condition": "sunny",
+            "humidity": 60,
+            "wind_speed": 10,
+            "precipitation": 0,
+            "note": "Mock data (no API key)"
+        }
+    try:
+        url = f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={OPENWEATHER_API_KEY}&units=metric"
+        resp = requests.get(url)
+        data = resp.json()
+        if resp.status_code != 200:
+            return {"error": f"Weather API error: {data.get('message', 'unknown')}"}
+        # Extract relevant fields
+        weather_info = {
+            "city": city,
+            "temperature": data["main"]["temp"],
+            "condition": data["weather"][0]["description"],
+            "humidity": data["main"]["humidity"],
+            "wind_speed": data["wind"]["speed"],
+            "precipitation": data.get("rain", {}).get("1h", 0)
+        }
+        return weather_info
+    except Exception as e:
+        return {"error": f"Weather service unavailable: {str(e)}"}
+
+@lru_cache(maxsize=1)
+def get_exchange_rates(base="USD"):
+    """Fetch latest exchange rates (cached). Fallback to mock."""
+    try:
+        url = f"https://api.exchangerate-api.com/v4/latest/{base}"
+        resp = requests.get(url)
+        return resp.json().get("rates", {})
+    except:
+        # Mock rates for common currencies
+        return {"USD": 1, "EUR": 0.85, "GBP": 0.75, "JPY": 110, "UGX": 3700, "KES": 130, "RWF": 1100}
+
+def convert_currency(amount: float, from_currency: str, to_currency: str) -> dict:
+    """Convert amount from one currency to another."""
+    rates = get_exchange_rates(from_currency.upper())
+    if to_currency.upper() not in rates:
+        return {"error": f"Currency {to_currency} not supported."}
+    converted = amount * rates[to_currency.upper()]
+    return {
+        "amount": amount,
+        "from": from_currency.upper(),
+        "to": to_currency.upper(),
+        "converted": round(converted, 2),
+        "rate": rates[to_currency.upper()]
+    }
+
+def get_attractions(city: str) -> dict:
+    """Return list of attractions for a given city."""
+    city_key = city.title()
+    if city_key in ATTRACTIONS_DB:
+        return {"city": city_key, "attractions": ATTRACTIONS_DB[city_key]}
+    else:
+        return {"error": f"No attractions data available for {city}.", "city": city}
+
+def compose_itinerary(city: str, weather: dict, attractions: list, budget: dict = None) -> str:
+    """
+    Generate a day trip itinerary using the tool outputs.
+    Calls OpenAI to format the plan based on verified data.
+    """
+    # Build context string
+    weather_text = f"{weather['condition'].capitalize()}, {weather['temperature']}°C"
+    if weather.get('precipitation', 0) > 0:
+        weather_text += f", chance of rain {weather['precipitation']}mm"
+    if "note" in weather:
+        weather_text += f" ({weather['note']})"
+
+    attractions_text = "\n".join([
+        f"- {a['name']} ({a['type']}, ~{a['duration_hours']} hrs, entry: {a['entry_fee']} {a.get('currency', '')})"
+        for a in attractions
+    ])
+
+    budget_text = ""
+    if budget and "error" not in budget:
+        budget_text = f"Budget: {budget['amount']} {budget['from']} (approx. {budget['converted']} {budget['to']} in local currency)"
+
+    prompt = f"""
+You are a travel assistant. Create a one-day itinerary for {city} based on the following verified information.
+
+Weather: {weather_text}
+Attractions available:
+{attractions_text}
+{budget_text}
+
+The itinerary should:
+- Be realistic given the weather (e.g., outdoor activities if sunny, indoor if rain).
+- Sequence attractions logically (considering location and opening hours if known).
+- Include estimated times for each activity.
+- Suggest meal times and local food options (use common knowledge, do not invent specific restaurants).
+- If a budget is provided, mention if attractions have entry fees and stay within budget.
+- Use markdown for readability (headings, bullet points, emojis).
+
+Return only the itinerary.
+"""
+    response = client.chat.completions.create(
+        model="gpt-3.5-turbo",
+        messages=[{"role": "user", "content": prompt}],
+        temperature=0.7
+    )
+    return response.choices[0].message.content
+
+# -------------------- TOOL SCHEMAS FOR OPENAI --------------------
+tools = [
+    {
+        "type": "function",
+        "function": {
+            "name": "get_weather",
+            "description": "Get current weather for a city",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "city": {"type": "string", "description": "City name"}
+                },
+                "required": ["city"]
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "convert_currency",
+            "description": "Convert an amount from one currency to another",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "amount": {"type": "number", "description": "Amount to convert"},
+                    "from_currency": {"type": "string", "description": "Source currency code (e.g., USD)"},
+                    "to_currency": {"type": "string", "description": "Target currency code (e.g., UGX)"}
+                },
+                "required": ["amount", "from_currency", "to_currency"]
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "get_attractions",
+            "description": "Get list of tourist attractions for a city",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "city": {"type": "string", "description": "City name"}
+                },
+                "required": ["city"]
+            }
+        }
+    }
+    # Note: compose_itinerary is not exposed as a tool; it's called by the agent after gathering data.
+]
+
+# -------------------- AGENT ORCHESTRATION --------------------
+def run_agent(user_message, history):
+    """
+    Main agent loop: maintains conversation, calls tools, and finally generates itinerary.
+    """
+    # System prompt that guides the agent
+    system_prompt = """You are a helpful travel planning assistant. You have access to tools that can:
+- Get current weather for a city
+- Convert currency
+- Retrieve tourist attractions for a city
+
+Your goal is to help the user plan a day trip. If the user's request is missing information (e.g., city, budget, currency), ask clarifying questions.
+When you have gathered all necessary data (city, weather, attractions, and optionally budget), call the appropriate tools to fetch the data, then use that information to compose a final itinerary.
+Do not invent any data; only use the tool outputs.
+Be friendly, concise, and helpful."""
+
+    messages = [{"role": "system", "content": system_prompt}]
+    # Add conversation history (Gradio passes history as list of [user, assistant] pairs)
+    for user_msg, asst_msg in history:
+        messages.append({"role": "user", "content": user_msg})
+        if asst_msg:
+            messages.append({"role": "assistant", "content": asst_msg})
+    messages.append({"role": "user", "content": user_message})
+
+    # We'll collect tool outputs in a dict for later use
+    tool_outputs = {}
+
+    # Max iterations to prevent infinite loops
+    for _ in range(5):
+        response = client.chat.completions.create(
+            model="gpt-3.5-turbo",
+            messages=messages,
+            tools=tools,
+            tool_choice="auto"
+        )
+        msg = response.choices[0].message
+
+        # If the model wants to call tools
+        if msg.tool_calls:
+            # Add assistant message with tool calls to history
+            messages.append(msg)
+            for tool_call in msg.tool_calls:
+                func_name = tool_call.function.name
+                args = json.loads(tool_call.function.arguments)
+                # Execute the tool
+                if func_name == "get_weather":
+                    result = get_weather(args["city"])
+                    tool_outputs["weather"] = result
+                elif func_name == "convert_currency":
+                    result = convert_currency(args["amount"], args["from_currency"], args["to_currency"])
+                    tool_outputs["budget"] = result
+                elif func_name == "get_attractions":
+                    result = get_attractions(args["city"])
+                    tool_outputs["attractions"] = result
+                else:
+                    result = {"error": "Unknown tool"}
+
+                # Append tool response
+                messages.append({
+                    "role": "tool",
+                    "tool_call_id": tool_call.id,
+                    "content": json.dumps(result)
+                })
+            # Continue loop to let model process tool responses
+            continue
+        else:
+            # No tool calls: either final answer or clarification request
+            final_text = msg.content
+            # If we have all necessary data, we might want to generate the itinerary via compose_itinerary
+            # But the model may already have done it if we instructed it to.
+            # To be safe, we can check if we have weather and attractions, and then call compose_itinerary.
+            if "weather" in tool_outputs and "attractions" in tool_outputs and "error" not in tool_outputs["attractions"]:
+                # We have the data, generate itinerary
+                city = tool_outputs["attractions"]["city"]
+                weather = tool_outputs["weather"]
+                attractions = tool_outputs["attractions"]["attractions"]
+                budget = tool_outputs.get("budget")
+                itinerary = compose_itinerary(city, weather, attractions, budget)
+                return itinerary
+            else:
+                # Model's response is a clarification or intermediate answer
+                return final_text
+
+    return "I'm having trouble processing your request. Please try again."
+
+# -------------------- GRADIO INTERFACE --------------------
+def chat_interface(message, history):
+    response = run_agent(message, history)
+    return response
+
+# Custom CSS for a nicer look
+css = """
+.gradio-container { font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif; }
+.chatbot { border-radius: 10px; }
+"""
+
+with gr.Blocks(css=css, title="TouristGuide AI Agent") as demo:
+    gr.Markdown("""
+    # 🌍 TouristGuide AI Agent
+    Your personal travel planner powered by AI and real-time tools.  
+    Ask for a day trip itinerary in any supported city (Kigali, Kampala, Nairobi, etc.) with optional budget and currency conversion.
+    """)
+    chatbot = gr.Chatbot(label="Conversation", bubble_full_width=False, height=500)
+    msg = gr.Textbox(label="Your message", placeholder="e.g., Plan a day trip in Kigali with a budget of 150 USD", lines=2)
+    clear = gr.Button("Clear conversation")
+
+    def respond(message, chat_history):
+        bot_message = run_agent(message, chat_history)
+        chat_history.append((message, bot_message))
+        return "", chat_history
+
+    msg.submit(respond, [msg, chatbot], [msg, chatbot])
+    clear.click(lambda: None, None, chatbot, queue=False)
+
+if __name__ == "__main__":
+    demo.launch(share=False, server_name="0.0.0.0")