diff --git a/week2/community-contributions/muawiya/main.py b/week2/community-contributions/muawiya/main.py new file mode 100644 index 0000000..b27ecf9 --- /dev/null +++ b/week2/community-contributions/muawiya/main.py @@ -0,0 +1,317 @@ +# imports + +import os +import json +from dotenv import load_dotenv +from openai import OpenAI +import gradio as gr +import base64 +from io import BytesIO +from PIL import Image +from IPython.display import Audio, display +import pygame +import time +from tools import price_function, get_ticket_price, make_a_booking, booking_function +import ollama +import anthropic +from anthropic import Anthropic +import whisper +import sounddevice as sd +import soundfile as sf +import numpy as np + +# And this is included in a list of tools: + +tools = [{"type": "function", "function": price_function}, {"type": "function", "function": booking_function}] +# tools = [price_function, booking_function] + +# System messages +system_message = "You are a helpful assistant for an Airline called FlightAI. " +system_message += "Give short, courteous answers, no more than 1 sentence. " +system_message += "Always be accurate. If you don't know the answer, say so." + +# Initialization + +load_dotenv(override=True) + +openai_api_key = os.getenv('OPENAI_API_KEY') +if openai_api_key: + print(f"OpenAI API Key exists and begins {openai_api_key[:8]}") +else: + print("OpenAI API Key not set") + +MODEL = "gpt-4o-mini" +openai = OpenAI() + +def chat(history): + messages = [{"role": "system", "content": system_message}] + history + response = openai.chat.completions.create(model=MODEL, messages=messages, tools=tools) + image = None + + if response.choices[0].finish_reason == "tool_calls": + message = response.choices[0].message + response, city = handle_tool_call(message) + messages.append(message) + messages.append(response) + if message.tool_calls[0].function.name == "get_ticket_price": + # image = artist(city) + pass + response = openai.chat.completions.create(model=MODEL, messages=messages) + + reply = response.choices[0].message.content + + # ✅ SAFETY CHECK: Never add empty or None replies + if reply: + history.append({"role": "assistant", "content": str(reply)}) + talker(reply) + else: + history.append({"role": "assistant", "content": "Sorry, no response available."}) + + return history, image + +# We have to write that function handle_tool_call: + +def handle_tool_call(message): + print(f"Handling tool call: {message}") + tool_call = message.tool_calls[0] + function_name = tool_call.function.name + arguments = json.loads(tool_call.function.arguments) + + if function_name == "get_ticket_price": + city = arguments.get('destination_city') + price = get_ticket_price(city) + response = { + "role": "tool", + "content": json.dumps({"destination_city": city, "price": price}), + "tool_call_id": tool_call.id + } + return response, city + + elif function_name == "make_a_booking": + city = arguments.get('destination_city') + customer_name = arguments.get('customer_name') + customer_id = arguments.get('customer_id') + booking_result = make_a_booking(city, customer_name, customer_id) + response = { + "role": "tool", + "content": json.dumps({ + "destination_city": city, + "customer_name": customer_name, + "customer_id": customer_id, + "booking_result": booking_result + }), + "tool_call_id": tool_call.id + } + return response, city + + else: + raise ValueError(f"Unknown function: {function_name}") + +def artist(city): + image_response = openai.images.generate( + model="dall-e-3", + prompt=f"An image representing a vacation in {city}, showing tourist spots and everything unique about {city}, in a vibrant pop-art style", + size="1024x1024", + n=1, + response_format="b64_json", + ) + image_base64 = image_response.data[0].b64_json + image_data = base64.b64decode(image_base64) + return Image.open(BytesIO(image_data)) + +def talker(message): + response = openai.audio.speech.create( + model="tts-1", + voice="onyx", + input=message) + + audio_stream = BytesIO(response.content) + output_filename = f"output_audio_{time.time()}.mp3" + with open(output_filename, "wb") as f: + f.write(audio_stream.read()) + + # Play the generated audio + # display(Audio(output_filename, autoplay=True)) # This code is suitable for Juopyter + print(f"Created audio file at {output_filename}") + + # Using pygame + pygame.init() + pygame.mixer.init() + pygame.mixer.music.load(output_filename) + pygame.mixer.music.play() + while pygame.mixer.music.get_busy(): + continue + +def ollama_translator(text, target_language="German"): + """ + Translates text to the specified language using Ollama. + + Args: + text (str): The text to translate + target_language (str): The language to translate to (default: Arabic) + + Returns: + str: The translated text + """ + try: + # Create a prompt that instructs the model to translate + prompt = f"Translate the following text to {target_language}. Only output the translation, nothing else:\n\n{text}" + + response = ollama.chat( + model='llama3.2:latest', # or any other model you have installed + messages=[ + {"role": "system", "content": "You are a professional translator. Translate the given text accurately."}, + {"role": "user", "content": prompt} + ] + ) + + translated_text = response['message']['content'].strip() + return translated_text + + except Exception as e: + print(f"Translation error: {str(e)}") + return f"Translation failed: {str(e)}" + +def translate_message(history): + """ + Translates the last message in the chat history. + + Args: + history (list): List of chat messages + + Returns: + str: Translated text of the last message + """ + if not history: + return "" + + # Get the last message from history + last_message = history[-1] + + # Extract the content from the last message + message_content = last_message.get('content', '') + + if message_content: + return ollama_translator(message_content) + return "" + +def clear_chat(): + return [], "" + +def convert_audio_to_text(audio_file_path): + """ + Converts audio to text using OpenAI's Whisper model. + Supports MP3, WAV, and other common audio formats. + + Args: + audio_file_path (str): Path to the audio file + + Returns: + str: Transcribed text + """ + try: + # Load the Whisper model + model = whisper.load_model("base") + + # Transcribe the audio file + result = model.transcribe(audio_file_path) + + # Return the transcribed text + return result["text"] + + except Exception as e: + print(f"Audio transcription error: {str(e)}") + return f"Transcription failed: {str(e)}" + +def handle_audio(audio_file, history): + history = history or [] + if audio_file: + try: + if not os.path.exists(audio_file): + raise Exception("Audio file not found") + + try: + transcribed_text = convert_audio_to_text(audio_file) + except Exception as e: + print(f"Transcription error: {str(e)}") + return history, None # 🛠️ match expected outputs + + if transcribed_text: + history.append({"role": "user", "content": str(transcribed_text)}) + + try: + if os.path.exists(audio_file): + os.remove(audio_file) + except Exception as e: + print(f"Warning: Could not delete audio file: {str(e)}") + + return history, None # ✅ return both expected outputs + except Exception as e: + print(f"Error processing audio: {str(e)}") + return history, None + return history, None + +if __name__ == "__main__": + # gr.ChatInterface(fn=chat, type="messages").launch() + # talker("Hello, how are you?") + # Passing in inbrowser=True in the last line will cause a Gradio window to pop up immediately. + + # print(ollama_translator("Hello, how are you?")) + # print(convert_audio_to_text("output_audio_1744898241.4550629.mp3")) + + with gr.Blocks() as ui: + with gr.Row(): + with gr.Column(): + chatbot = gr.Chatbot(height=500, type="messages") + with gr.Row(): + entry = gr.Textbox(label="Chat with our AI Assistant:") + audio_input = gr.Audio( + type="filepath", + label="Or speak your message:", + interactive=True, + format="wav", + # source="microphone" + ) + clear = gr.Button("Clear") + with gr.Column(): + translation_output = gr.Textbox(label="Translation (Arabic):", lines=5) + image_output = gr.Image(height=500) + + def do_entry(message, history): + history = history or [] + if message: + history.append({"role": "user", "content": str(message)}) + return "", history + + def translate_message(history): + if not history: + return "" + last_message = history[-1] + message_content = last_message.get('content', '') + if message_content: + return ollama_translator(message_content) + return "" + + def clear_chat(): + return [], "" + + # Handle text input + entry.submit(do_entry, inputs=[entry, chatbot], outputs=[entry, chatbot]).then( + chat, inputs=chatbot, outputs=[chatbot, image_output] + ).then( + translate_message, inputs=chatbot, outputs=translation_output + ) + + # Handle audio input + audio_input.stop_recording( + handle_audio, inputs=[audio_input, chatbot], outputs=[chatbot, image_output] + ).then( + chat, inputs=chatbot, outputs=[chatbot, image_output] + ).then( + translate_message, inputs=chatbot, outputs=translation_output + ) + + clear.click(clear_chat, inputs=None, outputs=[chatbot, translation_output]) + + ui.launch(inbrowser=False) + \ No newline at end of file diff --git a/week2/community-contributions/muawiya/requirements.txt b/week2/community-contributions/muawiya/requirements.txt new file mode 100644 index 0000000..ef945d2 Binary files /dev/null and b/week2/community-contributions/muawiya/requirements.txt differ diff --git a/week2/community-contributions/muawiya/tools.py b/week2/community-contributions/muawiya/tools.py new file mode 100644 index 0000000..28752c3 --- /dev/null +++ b/week2/community-contributions/muawiya/tools.py @@ -0,0 +1,57 @@ +# Let's start by making a useful function + +ticket_prices = {"london": "$799", "paris": "$899", "tokyo": "$1400", "berlin": "$499"} + +def get_ticket_price(destination_city): + print(f"Tool get_ticket_price called for {destination_city}") + city = destination_city.lower() + return ticket_prices.get(city, "Unknown") + +def make_a_booking(destination_city, customer_name, customer_id): + print(f"Tool make_a_booking called for {destination_city}") + city = destination_city.lower() + print(f"Customer name: {customer_name}, Customer ID: {customer_id}") + return True + +# There's a particular dictionary structure that's required to describe our function: + +price_function = { + "name": "get_ticket_price", + "description": "Get the price of a return ticket to the destination city. Call this whenever you need to know the ticket price, for example when a customer asks 'How much is a ticket to this city'", + "parameters": { + "type": "object", + "properties": { + "destination_city": { + "type": "string", + "description": "The city that the customer wants to travel to", + }, + }, + "required": ["destination_city"], + "additionalProperties": False + } +} + +booking_function = { + "name": "make_a_booking", + "description": "Make a booking for a customer to a destination city. Call this when a customer wants to book a flight. You can get the customer's name and ID by directly asking the customer. For example, you can say 'What is your name?' or 'What is your ID?'", + "parameters": { + "type": "object", + "properties": { + "destination_city": { + "type": "string", + "description": "The city that the customer wants to travel to", + }, + "customer_name": { + "type": "string", + "description": "The name of the customer making the booking", + }, + "customer_id": { + "type": "string", + "description": "The unique identifier for the customer", + } + }, + "required": ["destination_city", "customer_name", "customer_id"], + "additionalProperties": False + } +} +