import gradio as gr # Install necessary libraries os.system('pip install streamlit torch onnxruntime transformers sentencepiece pydub soxr edge-tts requests beautifulsoup4') # Import modules from other files from chatbot import chatbot, model_inference, BOT_AVATAR, EXAMPLES, model_selector, decoding_strategy, temperature, max_new_tokens, repetition_penalty, top_p from live_chat import videochat # Define Gradio theme theme = gr.themes.Soft( primary_hue="blue", secondary_hue="orange", neutral_hue="gray", font=[gr.themes.GoogleFont('Libre Franklin'), gr.themes.GoogleFont('Public Sans'), 'system-ui', 'sans-serif'] ).set( body_background_fill_dark="#111111", block_background_fill_dark="#111111", block_border_width="1px", block_title_background_fill_dark="#1e1c26", input_background_fill_dark="#292733", button_secondary_background_fill_dark="#24212b", border_color_primary_dark="#343140", background_fill_secondary_dark="#111111", color_accent_soft_dark="transparent" ) import edge_tts import asyncio import tempfile import numpy as np import soxr from pydub import AudioSegment import torch import sentencepiece as spm import onnxruntime as ort from huggingface_hub import hf_hub_download, InferenceClient import requests from bs4 import BeautifulSoup import urllib import random # 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', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36', 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36 Edg/111.0.1661.62', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/111.0' ] def get_useragent(): """Returns a random user agent from the list.""" return random.choice(_useragent_list) 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.extract() # Get the remaining visible text visible_text = soup.get_text(strip=True) return visible_text def search(term, num_results=1, lang="en", advanced=True, sleep_interval=0, timeout=5, safe="active", ssl_verify=None): """Performs a Google search and returns the results.""" escaped_term = urllib.parse.quote_plus(term) start = 0 all_results = [] # Fetch results in batches while start < num_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 - start, # Number of results to fetch in this batch "hl": lang, "start": start, "safe": safe, }, 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"}) # If no results, continue to the next batch if not result_block: start += 1 continue # Extract link and text from each result for result in result_block: link = result.find("a", href=True) if link: link = link["href"] try: # 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: # Handle errors fetching or processing webpage print(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) # Update starting index for next batch return all_results # Speech Recognition Model Configuration model_name = "neongeckocom/stt_en_citrinet_512_gamma_0_25" sample_rate = 16000 # 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 Configuration client1 = InferenceClient("mistralai/Mixtral-8x7B-Instruct-v0.1") system_instructions1 = "[SYSTEM] Answer as Real 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_fp32, sr): return soxr.resample(audio_fp32, sr, sample_rate) def to_float32(audio_buffer): return np.divide(audio_buffer, np.iinfo(audio_buffer.dtype).max, dtype=np.float32) 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()) audio_fp32 = to_float32(audio_buffer) audio_16k = resample(audio_fp32, sr) 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) logits = encoder.run(None, {'audio_signal': processed_signal.numpy(), 'length': length.numpy()})[0][0] 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=512, stream=True, details=True, return_full_text=False) return "".join([response.token.text for response in stream if response.token.text != ""]) else: formatted_prompt = system_instructions1 + text + "[OpenGPT 4o]" stream = client1.text_generation(formatted_prompt, max_new_tokens=512, stream=True, details=True, return_full_text=False) return "".join([response.token.text for response in stream if response.token.text != ""]) 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 voice: gr.Markdown("## Temproraly Not Working (Update in Progress)") with gr.Row(): web_search = gr.Checkbox(label="Web Search", value=False) input = gr.Audio(label="User Input", sources="microphone", type="filepath") output = gr.Audio(label="AI", autoplay=True) gr.Interface(fn=respond, inputs=[input, web_search], outputs=[output], live=True) # Create Gradio blocks for different functionalities # Chat interface block with gr.Blocks( fill_height=True, css=""".gradio-container .avatar-container {height: 40px width: 40px !important;} #duplicate-button {margin: auto; color: white; background: #f1a139; border-radius: 100vh; margin-top: 2px; margin-bottom: 2px;}""", ) as chat: gr.Markdown("### Image Chat, Image Generation and Normal Chat") with gr.Row(elem_id="model_selector_row"): # model_selector defined in chatbot.py pass # decoding_strategy, temperature, top_p defined in chatbot.py decoding_strategy.change( fn=lambda selection: gr.Slider( visible=( selection in [ "contrastive_sampling", "beam_sampling", "Top P Sampling", "sampling_top_k", ] ) ), inputs=decoding_strategy, outputs=temperature, ) decoding_strategy.change( fn=lambda selection: gr.Slider(visible=(selection in ["Top P Sampling"])), inputs=decoding_strategy, outputs=top_p, ) gr.ChatInterface( fn=model_inference, chatbot=chatbot, examples=EXAMPLES, multimodal=True, cache_examples=False, additional_inputs=[ model_selector, decoding_strategy, temperature, max_new_tokens, repetition_penalty, top_p, gr.Checkbox(label="Web Search", value=True), ], ) # Live chat block with gr.Blocks() as livechat: gr.Interface( fn=videochat, inputs=[gr.Image(type="pil",sources="webcam", label="Upload Image"), gr.Textbox(label="Prompt", value="what he is doing")], outputs=gr.Textbox(label="Answer") ) # Other blocks (instant, dalle, playground, image, instant2, video) with gr.Blocks() as instant: gr.HTML("") with gr.Blocks() as dalle: gr.HTML("") with gr.Blocks() as playground: gr.HTML("") with gr.Blocks() as image: gr.Markdown("""### More models are coming""") gr.TabbedInterface([ instant, dalle, playground], ['Instant🖼️','Powerful🖼️', 'Playground🖼']) with gr.Blocks() as instant2: gr.HTML("") with gr.Blocks() as video: gr.Markdown("""More Models are coming""") gr.TabbedInterface([ instant2], ['Instant🎥']) # Main application block with gr.Blocks(theme=theme, title="OpenGPT 4o DEMO") as demo: gr.Markdown("# OpenGPT 4o") gr.TabbedInterface([chat, voice, livechat, image, video], ['💬 SuperChat']) demo.queue(max_size=300) demo.launch()