import gradio as gr from edge_tts import list_voices 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 import re import time # 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) 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: # 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 fetching or processing {link}: {e}") all_results.append({"link": link, "text": None}) else: all_results.append({"link": None, "text": None}) print(all_results) return all_results # Speech Recognition Model Configuration model_name = "neongeckocom/stt_pt_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/Mistral-7B-Instruct-v0.2") system_instructions1 = "[SYSTEM] Responda como OpenGPT 4o, Feito por 'KingNish', Mantenha a conversa muito curta, clara, amigável e concisa. O texto fornecido é uma solicitação de um tipo específico de resposta sua, assistente virtual. A solicitação pede que você forneça respostas amigáveis. A expectativa é que eu evite apresentações e passe a responder a dúvida diretamente, elaborando todos os aspectos da solicitação. Como um assistente com tecnologia de IA, sua tarefa é gerar respostas que pareçam ter sido criadas pelo usuário e não por um assistente com tecnologia de IA. [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=300, 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=300, stream=True, details=True, return_full_text=False) return "".join([response.token.text for response in stream if response.token.text != ""]) async def get_voices(): voices = await edge_tts.list_voices() return list(voices) # Executar a função assíncrona para obter as vozes voices = asyncio.run(get_voices()) # Filtrar as vozes em português do Brasil pt_br_voices = [voice for voice in voices if voice["Locale"] == "pt-BR"] # Escolher uma voz (por exemplo, a primeira da lista) chosen_voice = pt_br_voices[0]["Name"] if pt_br_voices else None async def respond(audio, web_search): if audio is None: return None user = transcribe(audio) reply = model(user, web_search) if chosen_voice: communicate = edge_tts.Communicate(reply, voice=chosen_voice) else: communicate = edge_tts.Communicate(reply) # Usa a voz padrão se nenhuma voz pt-BR for encontrada with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp_file: tmp_path = tmp_file.name await communicate.save(tmp_path) return tmp_path def transcribe_and_respond(audio, web_search): return asyncio.run(respond(audio, web_search)) with gr.Blocks() as demo: with gr.Row(): web_search = gr.Checkbox(label="Web Search", value=False) Adjusted Gradio Audio Component with Silence Threshold input_audio = gr.Audio( sources=["microphone"], type="filepath", streaming=True, min_value=-0.1, # Adjust this value to set the silence threshold max_value=0.1 # Adjust this value to set the silence threshold ) output_audio = gr.Audio(label="AI Response", autoplay=True) is_recording = gr.State(False) last_interaction_time = gr.State(time.time()) def toggle_recording(): return not is_recording.value def process_audio(audio, web_search, is_rec): current_time = time.time() if is_rec and (current_time - last_interaction_time.value > 2): last_interaction_time.value = current_time return transcribe_and_respond(audio, web_search), False return None, is_rec input_audio.stream(process_audio, inputs=[input_audio, web_search, is_recording], outputs=[output_audio, is_recording]) demo.load(toggle_recording, outputs=[is_recording]) if __name__ == "__main__": demo.queue(max_size=200).launch()