Update app.py

app.py

@@ -1,7 +1,6 @@
 import os
 import urllib
 import requests
-import feedparser
 from bs4 import BeautifulSoup
 import torch
 import gradio as gr
@@ -38,7 +37,68 @@ def fetch_news(term, num_results=2):
         results.append({"link": entry.link, "text": entry.title})
     logger.debug(f"Fetched news results: {results}")
     return results
-    
+
+# Function to perform a Google search and return the results
+def search(term, num_results=2, lang="en", timeout=5, safe="active", ssl_verify=None):
+    logger.debug(f"Starting search for term: {term}")
+    escaped_term = urllib.parse.quote_plus(term)
+    start = 0
+    all_results = []
+    max_chars_per_page = 8000
+
+    with requests.Session() as session:
+        while start < num_results:
+            try:
+                resp = session.get(
+                    url="https://www.google.com/search",
+                    headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/111.0"},
+                    params={
+                        "q": term,
+                        "num": num_results - start,
+                        "hl": lang,
+                        "start": start,
+                        "safe": safe,
+                    },
+                    timeout=timeout,
+                    verify=ssl_verify,
+                )
+                resp.raise_for_status()
+                soup = BeautifulSoup(resp.text, "html.parser")
+                result_block = soup.find_all("div", attrs={"class": "g"})
+                if not result_block:
+                    start += 1
+                    continue
+                for result in result_block:
+                    link = result.find("a", href=True)
+                    if link:
+                        link = link["href"]
+                        try:
+                            webpage = session.get(link, headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/111.0"})
+                            webpage.raise_for_status()
+                            visible_text = extract_text_from_webpage(webpage.text)
+                            if len(visible_text) > max_chars_per_page:
+                                visible_text = visible_text[:max_chars_per_page] + "..."
+                            all_results.append({"link": link, "text": visible_text})
+                        except requests.exceptions.RequestException as e:
+                            logger.error(f"Error fetching or processing {link}: {e}")
+                            all_results.append({"link": link, "text": None})
+                    else:
+                        all_results.append({"link": None, "text": None})
+                start += len(result_block)
+            except Exception as e:
+                logger.error(f"Error during search: {e}")
+                break
+    logger.debug(f"Search results: {all_results}")
+    return all_results
+
+# Function to extract visible text from HTML content
+def extract_text_from_webpage(html_content):
+    soup = BeautifulSoup(html_content, "html.parser")
+    for tag in soup(["script", "style", "header", "footer", "nav"]):
+        tag.extract()
+    visible_text = soup.get_text(strip=True)
+    return visible_text
+
 # Function to format the prompt for the language model
 def format_prompt(user_prompt, chat_history):
     logger.debug(f"Formatting prompt with user prompt: {user_prompt} and chat history: {chat_history}")
@@ -72,26 +132,40 @@ def model_inference(
         if web_search:
             logger.debug("Performing news search")
             news_results = fetch_news(user_prompt["text"])
-            news_text = ' '.join([f"Link: {res['link']}\nText: {res['text']}\n\n" for res in news_results])
-            formatted_prompt = format_prompt(f"{user_prompt['text']} [NEWS] {news_text}", chat_history)
+            news2 = ' '.join([f"Link: {res['link']}\nText: {res['text']}\n\n" for res in news_results])
+            formatted_prompt = format_prompt(f"{user_prompt['text']} [NEWS] {news2}", chat_history)
+            inputs = tokenizer(formatted_prompt, return_tensors="pt").to(DEVICE)
+            if model:
+                outputs = model.generate(
+                    **inputs,
+                    max_new_tokens=max_new_tokens,
+                    repetition_penalty=repetition_penalty,
+                    do_sample=True,
+                    temperature=temperature,
+                    top_p=top_p
+                )
+                response = tokenizer.decode(outputs[0], skip_special_tokens=True)
+            else:
+                response = "Model is not available. Please try again later."
+            logger.debug(f"Model response: {response}")
+            return response
         else:
             formatted_prompt = format_prompt(user_prompt["text"], chat_history)
-
-        inputs = tokenizer(formatted_prompt, return_tensors="pt").to(DEVICE)
-        if model:
-            outputs = model.generate(
-                **inputs,
-                max_new_tokens=max_new_tokens,
-                repetition_penalty=repetition_penalty,
-                do_sample=True,
-                temperature=temperature,
-                top_p=top_p
-            )
-            response = tokenizer.decode(outputs[0], skip_special_tokens=True)
-        else:
-            response = "Model is not available. Please try again later."
-        logger.debug(f"Model response: {response}")
-        return response
+            inputs = tokenizer(formatted_prompt, return_tensors="pt").to(DEVICE)
+            if model:
+                outputs = model.generate(
+                    **inputs,
+                    max_new_tokens=max_new_tokens,
+                    repetition_penalty=repetition_penalty,
+                    do_sample=True,
+                    temperature=temperature,
+                    top_p=top_p
+                )
+                response = tokenizer.decode(outputs[0], skip_special_tokens=True)
+            else:
+                response = "Model is not available. Please try again later."
+            logger.debug(f"Model response: {response}")
+            return response
     else:
         return "Image input not supported in this implementation."
 
@@ -154,44 +228,42 @@ chatbot = gr.Chatbot(
 
 # Define Gradio interface
 def chat_interface(user_input, history, web_search, decoding_strategy, temperature, max_new_tokens, repetition_penalty, top_p):
-    logger.debug(f"Chat interface called with user_input: {user_input}")
-    if isinstance(user_input, str):
-        user_input = {"text": user_input, "files": []}
+    # Ensure the tokenizer is accessible within the function scope
+    global tokenizer
+    
+    # Perform model inference
     response = model_inference(
-        user_input,
-        history,
-        web_search,
-        temperature,
-        max_new_tokens,
-        repetition_penalty,
-        top_p,
-        tokenizer=tokenizer  # Pass tokenizer to model_inference
+        user_prompt=user_input,
+        chat_history=history,
+        web_search=web_search,
+        temperature=temperature,
+        max_new_tokens=max_new_tokens,
+        repetition_penalty=repetition_penalty,
+        top_p=top_p,
+        tokenizer=tokenizer  # Pass tokenizer to the model_inference function
     )
-    history.append((user_input["text"], response))
-    logger.debug(f"Updated chat history: {history}")
-    return history, history
+    
+    # Return the response
+    return response
 
-# Create Gradio interface
+# Define the Gradio interface components
 interface = gr.Interface(
     fn=chat_interface,
     inputs=[
-        gr.Textbox(label="User Input"),
-        gr.State([]),
-        gr.Checkbox(label="Fetch News", value=True),
+        gr.Textbox(label="User Input", placeholder="Type your message here..."),
+        gr.Textbox(label="Chat History", placeholder="Chat history will appear here..."),
+        gr.Checkbox(label="Perform Web Search", default=False),
         decoding_strategy,
         temperature,
         max_new_tokens,
         repetition_penalty,
         top_p
     ],
-    outputs=[
-        chatbot,
-        gr.State([])
-    ],
-    title="OpenGPT-4o-Chatty",
-    description="An AI assistant capable of insightful conversations and news fetching."
+    outputs=gr.Textbox(label="Assistant Response"),
+    live=True,
+    layout="vertical",
+    theme="compact"
 )
 
-if __name__ == "__main__":
-    logger.debug("Launching Gradio interface")
-    interface.launch()
+# Launch the Gradio interface
+interface.launch()