brad/LLM_Engineering_OLD
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add week2 contributions

Browse Source
This commit is contained in:
lisekarimi
2025-06-05 17:19:54 +02:00
parent 5782ca2b43
commit 01b33b4ede
2 changed files with 986 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

429
week2/community-contributions/04_tribot_debate.ipynb Normal file
View File

File diff suppressed because one or more lines are too long

557
week2/community-contributions/05_weathermate_ai_agent.ipynb Normal file
View File

@@ -0,0 +1,557 @@
{
"cells": [
{
"attachments": {},
"cell_type": "markdown",
"id": "ae1ef804-3504-488d-af86-5a0da36fea78",
"metadata": {},
"source": [
"# ☀️🏃‍♀️ WeatherMate\n",
"----\n",
"\n",
"**WeatherMate** is a conversational **AI agent** that analyzes real-time weather conditions and suggests the best activities and events based on location. Whether it's sunny, rainy, or snowy, WeatherMate helps you make the most of your day! \n",
"\n",
"Here's how it works:\n",
"1. Get current weather conditions for the user's location.\n",
"2. Recommend suitable indoor or outdoor activities based on the weather.\n",
"3. Find relevant events using the Ticketmaster API.\n",
"4. Merge both activity suggestions and events into a single, structured response.\n",
"\n",
"---\n",
"\n",
"Large Language Models (LLMs), by themselves, cannot fetch real-time data such as weather information. To enable LLMs to access and use such real-time data, we integrate **external tools.** \n",
"\n",
"In this notebook, we will implement a weather API, allowing the assistant to fetch real-time weather information and use it for personalized activity suggestions based on current weather conditions. This is an essential step in transforming an LLM into a more interactive and data-driven AI assistant.\n",
"\n",
"\n",
"In this notebook, we will develop a conversational AI Agent that helps users receive personalized activity recommendations based on real-time weather data.\n",
"\n",
"- 🧑‍💻 Skill Level: Advanced\n",
"- 📤 Output Format: conversational chat\n",
"- 🚀 Tools:\n",
" - Weather API integration \n",
" - Ticketmaster API\n",
" - OpenAI with external tool handling\n",
" - Gradio for the UI\n",
"\n",
"🛠️ Requirements\n",
"- ⚙️ Hardware: ✅ CPU is sufficient — no GPU required\n",
"- 🔑 OpenAI API Key\n",
"- 🔑 Weather API integration (https://www.weatherapi.com)\n",
"- 🔑 Ticketmaster API (https://developer.ticketmaster.com/explore/)\n",
"\n",
"⚙️ Customizable by user\n",
"- 🤖 Selected model\n",
"- 📜 system_prompt: Controls model behavior\n",
"\n",
"---\n",
"📢 Find more LLM notebooks on my [GitHub repository](https://github.com/lisekarimi/lexo)"
]
},
{
"cell_type": "markdown",
"id": "ad262788",
"metadata": {},
"source": [
"**Class Diagram**\n",
"\n",
"![](https://github.com/lisekarimi/lexo/blob/main/assets/05_weather_class_diagram.png?raw=true)\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "d6b7a492-f510-4ba4-bbc3-239675d389dd",
"metadata": {},
"outputs": [],
"source": [
"# imports\n",
"\n",
"import os\n",
"import json\n",
"import requests\n",
"from dotenv import load_dotenv\n",
"from openai import OpenAI\n",
"import gradio as gr\n",
"from datetime import datetime\n",
"\n",
"# Initialization\n",
"\n",
"load_dotenv(override=True)\n",
"\n",
"openai_api_key = os.getenv('OPENAI_API_KEY')\n",
"if not openai_api_key:\n",
" print(\"❌ OpenAI API Key is missing!\")\n",
"\n",
"weather_api_key = os.getenv('WEATHERAPI_KEY')\n",
"if not weather_api_key:\n",
" print(\"❌ Weather API Key is missing!\")\n",
"\n",
"ticketmaster_api_key = os.getenv('TICKETMASTER_KEY')\n",
"if not ticketmaster_api_key:\n",
" print(\"❌ TicketMaster API Key is missing!\")\n",
"\n",
"\n",
"MODEL = \"gpt-4o-mini\"\n",
"openai = OpenAI()"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "347dbe00-5826-4aa6-9d2c-9d028fc33ec8",
"metadata": {},
"outputs": [],
"source": [
"# Get today's date and day name\n",
"today_str = datetime.today().strftime('%Y-%m-%d')\n",
"day_name = datetime.today().strftime('%A')\n",
"\n",
"nb_activity = 10\n",
"\n",
"\n",
"system_message = f\"\"\"\n",
"You are a fun and helpful assistant for an Activity Suggestion App.\n",
"Your job is to recommend **up to {nb_activity} activities** based on the real-time weather fetched from the API, ensuring a mix of **indoor, outdoor, and event-based activities** whenever possible.\n",
"\n",
"The total must always be **10 or fewer**, following this rule:\n",
"**nb_events + nb_indoors + nb_outdoors ≤ 10**.\n",
"\n",
"You must **analyze and think carefully** to determine the best combination of activities and events for the user.\n",
"- Evaluate **weather conditions** to decide if outdoor activities are suitable.\n",
"- Check **event availability** and select the most relevant ones.\n",
"- Balance **indoor, outdoor, and event-based activities** dynamically to provide the best experience.\n",
"\n",
"If one of these categories is unavailable, that's fine—just provide the best possible suggestions without exceeding **10 activities**.\n",
"Deliver everything **in one go—no waiting!**\n",
"\n",
"\n",
"### **Understanding Relative Dates**\n",
"- Always interpret relative dates based on **{today_str} ({day_name})**.\n",
"- The weekend always refers to Saturday and Sunday.\n",
"- \"Next {day_name}\" should refer to the **closest upcoming occurrence** of that day.\n",
"- If the user asks for a time range (e.g., \"the next 3 days\"), calculate the **exact date range** starting from today.\n",
"- If no specific date is mentioned, **assume today by default**.\n",
"- **Do not ask for confirmation** when interpreting dates—just assume the correct date and proceed confidently unless there's real ambiguity.\n",
"\n",
"### **Activity and Event Suggestion Process**\n",
"To provide the best {nb_activity} activity recommendations, follow these steps:\n",
"Step 1: Retrieve Weather Data – Use the Weather API to get current conditions for the user's location.\n",
"Step 2: Suggest Activities – Recommend suitable indoor or outdoor activities based on the weather.\n",
"Step 3: Fetch Events (if available) – Use the Ticketmaster API to find relevant events in the user’s area.\n",
"Step 4: Combine Everything – Merge both event listings and activity suggestions into a single, well-structured response.\n",
"This entire process should be done seamlessly in one go without making the user wait.\n",
"\n",
"### **How to Handle Each API**\n",
"- **Weather API Handling**:\n",
" - If the user requests a relative date (e.g., \"tomorrow,\" \"next Monday\"), calculate the number of days from today.\n",
" - Provide the weather forecast only for the requested date, ignoring any other days in the response.\n",
" - If no weather data is available, inform the user in a friendly, light-hearted way.\n",
" - The forecast is limited to 14 days, so if the user requests a longer period, politely let him know.\n",
"\n",
"- **Ticketmaster API Handling**:\n",
" - If the user asks for events today, set the start date as today’s date.\n",
" - If the user asks for any specific weekday, find the next occurrence of that day and use it as the start date.\n",
" - If the user asks for a range of days (e.g., \"the next 3 days\"), use today’s date as the start date.\n",
" - The country corresponding to the user's city must be represented using the ISO Alpha-2 Code (e.g., FR for France, US for the United States, CA for Canada, DK for Denmark).\n",
" - If more than 5 events are found, ask the user for their interests to refine the search, using a one-word keyword like 'music,' 'cinema,' or 'theater.'\n",
" - If no events are found, explicitly inform the user in a friendly, funny way.\n",
" - Do not mention Ticketmaster unless necessary; simply state that you are checking for events.\n",
"\n",
"### **User Interaction Rules**\n",
"- If the user **doesn’t mention a city**, **ask them to provide one**.\n",
"- If an event search fails, do **not** mention Ticketmaster; simply say that no events were found.\n",
"- Ensure all activity suggestions are provided **in one response**, combining weather-based activities and event suggestions.\n",
"\n",
"\n",
"### **Event Formatting in Output**\n",
"**If Ticketmaster events are available**, format the output as follows:\n",
"Here are some events that may interest you:\n",
"**Event Name**:\n",
"- 📅 Date: Give the date like 19th March 2025\n",
"- 📍 Venue:\n",
"- 🔗 Ticket Link: Put the URL here\n",
"\n",
"(And don't forget to separate these gems with a snazzy divider)\n",
"\n",
"**Event Name**:\n",
"- 📅 Date: Give the date like 19th March 2025\n",
"- 📍 Venue:\n",
"- 🔗 Ticket Link: Put the URL here\n",
"\n",
"(Another divider, because we like to keep things fresh!)\n",
"\n",
"**Event Name**:\n",
"- 📅 Date: Give the date like 19th March 2025\n",
"- 📍 Venue:\n",
"- 🔗 Ticket Link: Put the URL here\n",
"\n",
"### **Tone and Style**\n",
"**Keep it short, fun, and don’t forget to add a dash of humor!**\n",
"Your job is to keep the user smiling while giving them the **best activities for the day**.\n",
"Be **accurate and concise**, but let’s keep it **light and lively!** 🎉\n",
"\"\"\"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "578da33d-be38-4c75-8a96-9d6bfc1af99b",
"metadata": {},
"outputs": [],
"source": [
"class WeatherAPI:\n",
" def get_weather(self, city: str, days: int) -> dict:\n",
" \"\"\"Fetches weather data for the given city for the next 'days' number of days.\"\"\"\n",
" url = \"https://api.weatherapi.com/v1/forecast.json\"\n",
" params = {\"key\": weather_api_key, \"q\": city, \"days\": days}\n",
" # print(f\"params weather: {params}\")\n",
" response = requests.get(url, params=params)\n",
"\n",
" if response.status_code == 200:\n",
" data = response.json()\n",
" forecast = []\n",
" for day in data[\"forecast\"][\"forecastday\"]:\n",
" forecast.append({\n",
" \"date\": day[\"date\"],\n",
" \"temp\": day[\"day\"][\"avgtemp_c\"]\n",
" })\n",
"\n",
" result = {\n",
" \"city\": city,\n",
" \"forecast\": forecast\n",
" }\n",
" return result\n",
" else:\n",
" return {\"error\": f\"City '{city}' not found or other issue. Please check the city name and try again.\"}"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "305f9f18-8556-4b49-9f6b-4a2233eefae9",
"metadata": {},
"outputs": [],
"source": [
"from abc import ABC, abstractmethod\n",
"\n",
"class BaseEventAPI(ABC):\n",
" @abstractmethod\n",
" def get_events(self, city, country_code, keywords, size):\n",
" \"\"\"Fetches upcoming events from an event provider.\"\"\"\n",
" pass # Subclasses must implement this method\n",
"\n",
"class TicketmasterAPI(BaseEventAPI):\n",
" def get_events(self, city, country_code, keywords, start_date):\n",
" \"\"\"Fetches upcoming events from Ticketmaster for a given city.\"\"\"\n",
" url = \"https://app.ticketmaster.com/discovery/v2/events.json\"\n",
" params = {\n",
" \"apikey\": ticketmaster_api_key,\n",
" \"city\": city,\n",
" \"countryCode\": country_code,\n",
" \"keyword\": \",\".join(keywords),\n",
" \"size\": 10,\n",
" \"startDateTime\": start_date\n",
" }\n",
"\n",
" response = requests.get(url, params=params)\n",
"\n",
" if response.status_code == 200:\n",
" data = response.json()\n",
" events = data.get(\"_embedded\", {}).get(\"events\", [])\n",
" return [\n",
" {\n",
" \"name\": event[\"name\"],\n",
" \"date\": event[\"dates\"][\"start\"][\"localDate\"],\n",
" \"venue\": event[\"_embedded\"][\"venues\"][0][\"name\"],\n",
" \"url\": event.get(\"url\", \"N/A\") # Using .get() to avoid KeyError\n",
" }\n",
" for event in events\n",
" ] if events else []\n",
" else:\n",
" return {\"error\": f\"API request failed! Status: {response.status_code}, Response: {response.text}\"}\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "4c60820f-4e9f-4851-8330-52c8fd676259",
"metadata": {},
"outputs": [],
"source": [
"class ChatAssistant:\n",
" def __init__(self):\n",
" self.model = MODEL\n",
" self.tools = [\n",
" {\n",
" \"type\": \"function\",\n",
" \"function\": {\n",
" \"name\": \"get_weather\",\n",
" \"description\": \"Get the current weather and forecast for the destination city.\",\n",
" \"parameters\": {\n",
" \"type\": \"object\",\n",
" \"properties\": {\n",
" \"city\": {\n",
" \"type\": \"string\",\n",
" \"description\": \"The city for which the weather is being requested.\"\n",
" },\n",
" \"days\": {\n",
" \"type\": \"integer\",\n",
" \"description\": \"The number of days for the weather forecast (can be 1, 2, 6, or 10).\"\n",
" }\n",
" },\n",
" \"required\": [\"city\", \"days\"],\n",
" \"additionalProperties\": False\n",
" }\n",
" }\n",
" },\n",
" {\n",
" \"type\": \"function\",\n",
" \"function\": {\n",
" \"name\": \"get_ticketmaster_events\",\n",
" \"description\": \"Fetch upcoming events from Ticketmaster.\",\n",
" \"parameters\": {\n",
" \"type\": \"object\",\n",
" \"properties\": {\n",
" \"city\": {\n",
" \"type\": \"string\",\n",
" \"description\": \"City where the events are searched.\"\n",
" },\n",
" \"country_code\": {\n",
" \"type\": \"string\",\n",
" \"description\": \"Country code for filtering results.\"\n",
" },\n",
" \"keywords\": {\n",
" \"type\": \"array\",\n",
" \"items\": {\n",
" \"type\": \"string\"\n",
" },\n",
" \"description\": \"Optional keywords for event search (e.g., 'music', 'concert').\"\n",
" },\n",
" \"size\": {\n",
" \"type\": \"integer\",\n",
" \"description\": \"Number of events to fetch.\"\n",
" },\n",
" \"start_date\": {\n",
" \"type\": \"string\",\n",
" \"description\": \"Start date for the event search.\"\n",
" }\n",
" },\n",
" \"required\": [\"city\", \"country_code\", \"size\", \"start_date\"],\n",
" \"additionalProperties\": False\n",
" }\n",
" }\n",
" }\n",
" ]\n",
"\n",
" def chat(self, user_message, history, weather_api, event_apis):\n",
" # Build the conversation\n",
" messages = [{\"role\": \"system\", \"content\": system_message}] + history + [{\"role\": \"user\", \"content\": user_message}]\n",
"\n",
" # OpenAI response\n",
" response = openai.chat.completions.create(model=self.model, messages=messages, tools=self.tools, stream=True)\n",
"\n",
" recovered_pieces = {\n",
" \"content\": None,\n",
" \"role\": \"assistant\",\n",
" \"tool_calls\": {}\n",
" }\n",
" last_tool_calls = {}\n",
" has_tool_call = False\n",
" result = \"\" # Initialize result accumulator\n",
" # previous_index = None # Track the last processed index\n",
"\n",
" for chunk in response:\n",
" delta = chunk.choices[0].delta\n",
" finish_reason = chunk.choices[0].finish_reason\n",
"\n",
" # Handle tool call detection\n",
" if delta.tool_calls and finish_reason in [None, \"tool_calls\"]:\n",
" has_tool_call = True\n",
" piece = delta.tool_calls[0] # Get the first piece in the tool call\n",
"\n",
" # Create a dictionary for the tool call if it doesn't exist yet\n",
" recovered_pieces[\"tool_calls\"][piece.index] = recovered_pieces[\"tool_calls\"].get(\n",
" piece.index, {\"id\": None, \"function\": {\"arguments\": \"\", \"name\": \"\"}, \"type\": \"function\"}\n",
" )\n",
"\n",
" if piece.id:\n",
" recovered_pieces[\"tool_calls\"][piece.index][\"id\"] = piece.id\n",
" if piece.function.name:\n",
" recovered_pieces[\"tool_calls\"][piece.index][\"function\"][\"name\"] = piece.function.name\n",
" recovered_pieces[\"tool_calls\"][piece.index][\"function\"][\"arguments\"] += piece.function.arguments\n",
"\n",
" # Store the tool call in the dictionary by index\n",
" last_tool_calls[piece.index] = recovered_pieces[\"tool_calls\"][piece.index]\n",
"\n",
" # Store content in result and yield\n",
" else:\n",
" result += delta.content or \"\"\n",
" if result.strip():\n",
" yield result\n",
"\n",
"\n",
" # Handle tool call scenario\n",
" if has_tool_call:\n",
" # Handle the tool calls\n",
" response = self.handle_tool_call(last_tool_calls, weather_api, event_apis)\n",
"\n",
" if response: # Only iterate if response is not None\n",
" tool_calls_list = [tool_call for tool_call in last_tool_calls.values()]\n",
" messages.append({\"role\": \"assistant\", \"tool_calls\": tool_calls_list}) # Append the tool calls to the messages\n",
"\n",
" # Dynamically process each tool call response and append it to the message history\n",
" for res in response:\n",
" messages.append({\n",
" \"role\": \"tool\",\n",
" \"tool_call_id\": res[\"tool_call_id\"],\n",
" \"content\": json.dumps(res[\"content\"])\n",
" })\n",
"\n",
" # New OpenAI request with tool response\n",
" response = openai.chat.completions.create(model=self.model, messages=messages, stream=True)\n",
"\n",
" result = \"\" # Reset result before second stream\n",
" for chunk in response:\n",
" result += chunk.choices[0].delta.content or \"\"\n",
" if result.strip():\n",
" yield result\n",
"\n",
"\n",
" def handle_tool_call(self, tool_call, weather_api, event_apis):\n",
" stored_values = {} # Dictionary to store the valid value for each field\n",
"\n",
" for index, call in tool_call.items():\n",
" # Load the arguments for each tool call dynamically\n",
" arguments = json.loads(call[\"function\"][\"arguments\"])\n",
"\n",
" # Iterate over all keys dynamically\n",
" for key, value in arguments.items():\n",
" # Update the field if it's currently None or hasn't been set before\n",
" if key not in stored_values or stored_values[key] is None:\n",
" stored_values[key] = value\n",
"\n",
" city = stored_values.get('city')\n",
" days = stored_values.get('days')\n",
" country_code = stored_values.get('country_code')\n",
" keywords = stored_values.get('keywords', [])\n",
" # size = stored_values.get('size')\n",
" start_date = stored_values.get('start_date')\n",
" start_date = str(start_date) + \"T00:00:00Z\"\n",
"\n",
" weather_data = None\n",
" event_data = None\n",
"\n",
" # Iteration over tool_call\n",
" for call in tool_call.values():\n",
" if call[\"function\"][\"name\"] == \"get_weather\":\n",
" weather_data = weather_api.get_weather(city, days)\n",
"\n",
" if call[\"function\"][\"name\"] == \"get_ticketmaster_events\":\n",
" event_data = event_apis[\"ticketmaster\"].get_events(city, country_code, keywords, start_date)\n",
"\n",
" responses = []\n",
"\n",
" # Ensure weather response is always included\n",
" weather_tool_call_id = next((call[\"id\"] for call in tool_call.values() if call[\"function\"][\"name\"] == \"get_weather\"), None)\n",
" if weather_data and \"forecast\" in weather_data:\n",
" responses.append({\n",
" \"role\": \"assistant\",\n",
" \"content\": {\"weather\": weather_data[\"forecast\"]},\n",
" \"tool_call_id\": weather_tool_call_id\n",
" })\n",
" elif weather_tool_call_id:\n",
" responses.append({\n",
" \"role\": \"assistant\",\n",
" \"content\": {\"message\": \"No weather data available for this location.\"},\n",
" \"tool_call_id\": weather_tool_call_id\n",
" })\n",
"\n",
" # Ensure event response is always included\n",
" event_tool_call_id = next((call[\"id\"] for call in tool_call.values() if call[\"function\"][\"name\"] == \"get_ticketmaster_events\"), None)\n",
" if event_data:\n",
" responses.append({\n",
" \"role\": \"assistant\",\n",
" \"content\": {\"events\": event_data},\n",
" \"tool_call_id\": event_tool_call_id\n",
" })\n",
" elif event_tool_call_id:\n",
" responses.append({\n",
" \"role\": \"assistant\",\n",
" \"content\": {\"message\": \"No events found for this location.\"},\n",
" \"tool_call_id\": event_tool_call_id\n",
" })\n",
"\n",
" # print(\"Final responses:\", responses)\n",
" return responses\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "191a3a9e-95e1-4ca6-8992-4a5bafb9b8ff",
"metadata": {},
"outputs": [],
"source": [
"# GradioInterface class to handle the Gradio UI\n",
"class GradioInterface:\n",
" def __init__(self, activity_assistant):\n",
" self.activity_assistant = activity_assistant\n",
"\n",
" def launch(self):\n",
" # Gradio chat interface\n",
" gr.ChatInterface(fn=self.activity_assistant.chat, type=\"messages\").launch()\n",
"\n",
"# ActivityAssistant setup\n",
"class ActivityAssistant:\n",
" def __init__(self):\n",
" self.weather_api = WeatherAPI() # Interact with the Weather API\n",
" self.event_apis = { # Interact with the Events API\n",
" \"ticketmaster\": TicketmasterAPI()\n",
" }\n",
" self.chat_assistant = ChatAssistant() # This will handle conversation with OpenAI\n",
"\n",
" def chat(self, user_message, history):\n",
" # Forward the user message and conversation history to ChatAssistant\n",
" response_stream = self.chat_assistant.chat(user_message, history, self.weather_api, self.event_apis)\n",
" for chunk in response_stream:\n",
" yield chunk"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "0b501e8e-2e10-4ab7-b523-1d4b8ad358e8",
"metadata": {},
"outputs": [],
"source": [
"# Main execution\n",
"if __name__ == \"__main__\":\n",
" activity_assistant = ActivityAssistant()\n",
" gradio_interface = GradioInterface(activity_assistant)\n",
" gradio_interface.launch()"
]
}
],
"metadata": {
"kernelspec": {
"display_name": ".venv",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.11.7"
}
},
"nbformat": 4,
"nbformat_minor": 5
}