Update app.py

app.py

@@ -11,12 +11,12 @@ import onnxruntime as ort
 from huggingface_hub import hf_hub_download, InferenceClient
 import requests
 from bs4 import BeautifulSoup
-import urllib.parse
+import urllib
 import random
 import re
 
-# List of user agents for requests
-user_agents = [
+# List of user agents to choose from for requests
+_useragent_list = [
     'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:66.0) Gecko/20100101 Firefox/66.0',
     'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36',
     'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36',
@@ -26,122 +26,128 @@ user_agents = [
     'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/111.0'
 ]
 
-def get_random_user_agent():
-    """Selects a random user agent from the list."""
-    return random.choice(user_agents)
+def get_useragent():
+    """Returns a random user agent from the list."""
+    return random.choice(_useragent_list)
 
-def extract_visible_text(html):
-    """Extracts visible text from HTML content, removing unwanted tags."""
-    soup = BeautifulSoup(html, "html.parser")
+def extract_text_from_webpage(html_content):
+    """Extracts visible text from HTML content using BeautifulSoup."""
+    soup = BeautifulSoup(html_content, "html.parser")
+    # Remove unwanted tags
     for tag in soup(["script", "style", "header", "footer", "nav"]):
-        tag.decompose()
-    visible_text = soup.get_text(separator=' ', strip=True)
-    return visible_text[:8000]  # Truncate to a reasonable length
-
-def perform_search(query, num_results=2, timeout=5, ssl_verify=None):
-    """Conducts a Google search and retrieves the visible text from the results."""
-    escaped_query = urllib.parse.quote_plus(query)
-    search_results = []
-    response = requests.get(
-        url="https://www.google.com/search",
-        headers={"User-Agent": get_random_user_agent()},
-        params={"q": escaped_query, "num": num_results},
-        timeout=timeout,
-        verify=ssl_verify,
-    )
-    response.raise_for_status()
-    soup = BeautifulSoup(response.text, "html.parser")
-    result_blocks = soup.find_all("div", class_="g")
-    
-    for block in result_blocks:
-        link_tag = block.find("a", href=True)
-        if link_tag:
-            link = link_tag["href"]
+        tag.extract()
+    # Get the remaining visible text
+    visible_text = soup.get_text(strip=True)
+    visible_text = visible_text[:8000]
+    return visible_text
+
+def search(term, num_results=2, timeout=5, ssl_verify=None):
+    """Performs a Google search and returns the results."""
+    escaped_term = urllib.parse.quote_plus(term)
+    all_results = []
+    resp = requests.get(
+            url="https://www.google.com/search",
+            headers={"User-Agent": get_useragent()}, # Set random user agent
+            params={
+                "q": term,
+                "num": num_results,
+                "udm": 14,
+            },
+            timeout=timeout,
+            verify=ssl_verify,
+        )
+    resp.raise_for_status() # Raise an exception if request fails
+    soup = BeautifulSoup(resp.text, "html.parser")
+    result_block = soup.find_all("div", attrs={"class": "g"})
+    for result in result_block:
+        link = result.find("a", href=True)
+        if link:
+            link = link["href"]
             try:
-                webpage_response = requests.get(link, headers={"User-Agent": get_random_user_agent()})
-                webpage_response.raise_for_status()
-                visible_text = extract_visible_text(webpage_response.text)
-                search_results.append({"link": link, "text": visible_text})
+                    # Fetch webpage content
+                webpage = requests.get(link, headers={"User-Agent": get_useragent()})
+                webpage.raise_for_status()
+                    # Extract visible text from webpage
+                visible_text = extract_text_from_webpage(webpage.text)
+                all_results.append({"link": link, "text": visible_text})
             except requests.exceptions.RequestException as e:
-                print(f"Error processing {link}: {e}")
-                search_results.append({"link": link, "text": None})
+                print(f"Error fetching or processing {link}: {e}")
+                all_results.append({"link": link, "text": None})
         else:
-            search_results.append({"link": None, "text": None})
-
-    return search_results
+            all_results.append({"link": None, "text": None})
+    print(all_results)
+    return all_results
 
-# Speech Recognition Model Setup
+# Speech Recognition Model Configuration
 model_name = "neongeckocom/stt_en_citrinet_512_gamma_0_25"
 sample_rate = 16000
 
-# Download necessary components
+# Download preprocessor, encoder and tokenizer
 preprocessor = torch.jit.load(hf_hub_download(model_name, "preprocessor.ts", subfolder="onnx"))
 encoder = ort.InferenceSession(hf_hub_download(model_name, "model.onnx", subfolder="onnx"))
 tokenizer = spm.SentencePieceProcessor(hf_hub_download(model_name, "tokenizer.spm", subfolder="onnx"))
 
-# Mistral Model Setup
-client = InferenceClient("mistralai/Mistral-7B-Instruct-v0.2")
-system_prompt = "<s>[SYSTEM] Respond as OpenGPT 4o, a friendly, concise virtual assistant developed by 'KingNish'. Please keep your responses clear, direct, and supportive.[USER]"
+# Mistral Model Configuration
+client1 = InferenceClient("mistralai/Mistral-7B-Instruct-v0.2")
+system_instructions1 = "<s>[SYSTEM] Answer as OpenGPT 4o, Made by 'KingNish', Keep conversation very short, clear, friendly and concise. The text provided is a request for a specific type of response from you, the virtual assistant. The request asks you to provide friendly responses. The expectation is that I will avoid introductions and start answering the query directly, elaborating on all aspects of the request. As an AI-powered assistant, your task is to generate responses that appear as if they are created by the user and not an AI-powered assistant. [USER]"
 
-def resample_audio(audio_data, original_sr):
-    """Resamples the audio to the desired sample rate."""
-    return soxr.resample(audio_data, original_sr, sample_rate)
+def resample(audio_fp32, sr):
+    return soxr.resample(audio_fp32, sr, sample_rate)
 
-def convert_to_float32(audio_buffer):
-    """Converts audio buffer to float32 format."""
+def to_float32(audio_buffer):
     return np.divide(audio_buffer, np.iinfo(audio_buffer.dtype).max, dtype=np.float32)
 
-def transcribe_audio(audio_path):
-    """Transcribes speech from an audio file."""
-    audio = AudioSegment.from_file(audio_path)
-    audio_buffer = np.array(audio.get_array_of_samples())
-    audio_fp32 = convert_to_float32(audio_buffer)
-    resampled_audio = resample_audio(audio_fp32, audio.frame_rate)
-
-    input_tensor = torch.tensor(resampled_audio).unsqueeze(0)
-    length_tensor = torch.tensor(len(resampled_audio)).unsqueeze(0)
-    processed_signal, _ = preprocessor.forward(input_signal=input_tensor, length=length_tensor)
-    
-    logits = encoder.run(None, {'audio_signal': processed_signal.numpy(), 'length': length_tensor.numpy()})[0][0]
-    blank_id = tokenizer.vocab_size()
-    decoded_ids = [p for p in logits.argmax(axis=1).tolist() if p != blank_id]
-    transcript = tokenizer.decode_ids(decoded_ids)
+def transcribe(audio_path):
+    audio_file = AudioSegment.from_file(audio_path)
+    sr = audio_file.frame_rate
+    audio_buffer = np.array(audio_file.get_array_of_samples())
 
-    return transcript
+    audio_fp32 = to_float32(audio_buffer)
+    audio_16k = resample(audio_fp32, sr)
 
-def generate_response(text, perform_web_search):
-    """Generates a response using the language model, optionally including web search results."""
-    if perform_web_search:
-        web_results = perform_search(text)
-        web_summary = ' '.join([f"Link: {res['link']}\nText: {res['text']}\n\n" for res in web_results])
-        full_prompt = system_prompt + text + "[WEB]" + str(web_summary) + "[OpenGPT 4o]"
-    else:
-        full_prompt = system_prompt + text + "[OpenGPT 4o]"
-    
-    response_stream = client.text_generation(full_prompt, max_new_tokens=300, stream=True, details=True, return_full_text=False)
-    response_text = "".join([resp.token.text for resp in response_stream if resp.token.text != "</s>"])
-    return response_text
-
-async def handle_interaction(audio_input, perform_web_search):
-    """Handles user interaction by transcribing audio, generating a response, and converting it to speech."""
-    user_query = transcribe_audio(audio_input)
-    ai_reply = generate_response(user_query, perform_web_search)
-    tts_engine = edge_tts.Communicate(ai_reply)
+    input_signal = torch.tensor(audio_16k).unsqueeze(0)
+    length = torch.tensor(len(audio_16k)).unsqueeze(0)
+    processed_signal, _ = preprocessor.forward(input_signal=input_signal, length=length)
     
-    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as temp_file:
-        temp_filepath = temp_file.name
-        await tts_engine.save(temp_filepath)
-    return temp_filepath
+    logits = encoder.run(None, {'audio_signal': processed_signal.numpy(), 'length': length.numpy()})[0][0]
 
-with gr.Blocks() as interface:
-    gr.Markdown("# Emotional Support Assistant\nI'm here to offer emotional support and answer your questions. How are you today?")
+    blank_id = tokenizer.vocab_size()
+    decoded_prediction = [p for p in logits.argmax(axis=1).tolist() if p != blank_id]
+    text = tokenizer.decode_ids(decoded_prediction)
+
+    return text
+
+def model(text, web_search):
+    if web_search is True:
+        """Performs a web search, feeds the results to a language model, and returns the answer."""
+        web_results = search(text)
+        web2 = ' '.join([f"Link: {res['link']}\nText: {res['text']}\n\n" for res in web_results])
+        formatted_prompt = system_instructions1 + text + "[WEB]" + str(web2) + "[OpenGPT 4o]"
+        stream = client1.text_generation(formatted_prompt, max_new_tokens=300, stream=True, details=True, return_full_text=False)
+        return "".join([response.token.text for response in stream if response.token.text != "</s>"])
+    else:
+        formatted_prompt = system_instructions1 + text + "[OpenGPT 4o]"
+        stream = client1.text_generation(formatted_prompt, max_new_tokens=300, stream=True, details=True, return_full_text=False)
+        return "".join([response.token.text for response in stream if response.token.text != "</s>"])
+
+async def respond(audio, web_search):
+    user = transcribe(audio)
+    reply = model(user, web_search)
+    communicate = edge_tts.Communicate(reply)
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp_file:
+        tmp_path = tmp_file.name
+        await communicate.save(tmp_path)
+    return tmp_path
+
+with gr.Blocks() as demo:
+    gr.Markdown("# Emotional Support\nHello! I'm here to support you emotionally and answer any questions. How are you feeling today?")
     gr.Markdown("<p style='color:green;'>Developed by Hashir Ehtisham</p>")
     
     with gr.Row():
-        web_search_toggle = gr.Checkbox(label="Perform Web Search", value=False)
-        audio_input = gr.Audio(label="Speak Your Query", source="microphone", type="filepath")
-        audio_output = gr.Audio(label="Response", autoplay=True)
-        gr.Interface(fn=handle_interaction, inputs=[audio_input, web_search_toggle], outputs=[audio_output], live=True)
+        web_search = gr.Checkbox(label="Web Search", value=False)
+        audio_input = gr.Audio(label="Speak Your Query", type="filepath")
+        output = gr.Audio(label="AI Response", autoplay=True)
+        gr.Interface(fn=respond, inputs=[audio_input, web_search], outputs=[output], live=True)
 
 if __name__ == "__main__":
-    interface.queue(max_size=200).launch()
+    demo.queue(max_size=200).launch()