Upload app.py

app.py

@@ -0,0 +1,442 @@
+
+
+# Commented out IPython magic to ensure Python compatibility.
+# %%shell
+# 
+# pip install -q langchain_community langchain_huggingface faiss-cpu gradio openai google-generativeai langchain-google-genai torch torchvision torchaudio youtokentome pypdf accelerate
+#
+
+
+# Commented out IPython magic to ensure Python compatibility.
+# %%capture
+# %%shell
+# 
+# # Install the custom version of NeMo by AI4Bharat
+# wget https://indic-asr-public.objectstore.e2enetworks.net/ai4b_nemo.zip
+# 
+# unzip -q /content/ai4b_nemo.zip && cd NeMo
+# bash reinstall.sh
+# 
+# cd ..
+#
+
+# Commented out IPython magic to ensure Python compatibility.
+# %%capture
+# %%shell
+# 
+# git clone -q https://github.com/VarunGumma/IndicTransTokenizer
+# cd IndicTransTokenizer
+# pip install -q --editable ./
+# cd ..
+#
+
+
+# Commented out IPython magic to ensure Python compatibility.
+# %%capture
+# %%shell
+# 
+# apt-get install libsndfile1-dev ffmpeg
+# 
+# git clone https://github.com/gokulkarthik/TTS
+# cd TTS
+# 
+# pip3 install -e .[all]
+# pip3 install -r requirements.txt
+# 
+# cd ..
+#
+
+"""## **Restart session**
+"""
+
+# Commented out IPython magic to ensure Python compatibility.
+# %%capture
+# !pip install gradio
+
+# Commented out IPython magic to ensure Python compatibility.
+# %%capture
+# 
+# # INFO: If you're unable to import these libraries, just rerun this cell again.
+# 
+import gradio as gr
+from torch import cuda, inference_mode
+import nemo.collections.asr as nemo_asr
+from IndicTransTokenizer import IndicProcessor
+from transformers import AutoModelForSeq2SeqLM, AutoTokenizer
+
+
+DEVICE = "cuda" if cuda.is_available() else "cpu"
+
+print(f"Using device: {DEVICE}")
+
+
+import os
+from langchain_community.vectorstores import FAISS
+from langchain_huggingface import HuggingFaceEmbeddings
+from langchain_community.document_loaders import PyPDFLoader
+
+"""### Load and convert PDF data into vectorDB"""
+
+pm_kisan_doc = "/content/PM-KISANOperationalGuidelines(English).pdf"
+
+from langchain_community.document_loaders import PyPDFLoader
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+
+text_splitter = RecursiveCharacterTextSplitter(
+    chunk_size=600,
+    chunk_overlap=100
+)
+
+loader = PyPDFLoader(pm_kisan_doc)
+pages = loader.load_and_split(text_splitter=text_splitter)
+
+pages_chunks = [page.page_content for page in pages]
+print(f"Generated {len(pages_chunks)} chunks of {pm_kisan_doc}")
+
+pages_chunks[8]
+
+embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")
+
+faiss = FAISS.from_texts(pages_chunks, embeddings)
+
+"""### Querying the vectorDB"""
+
+# Test query
+result = faiss.similarity_search("what are the benefits of PM kisan yojna", k=3)
+
+# This returns the most relevant doc similar to the query
+
+print(result[0].page_content)
+
+Result_with_score = faiss.similarity_search_with_score("what are the benefits of PM kisan yojna", k=3)
+Result_with_score[0]
+
+os.environ['GEMINI_API_KEY'] = userdata.get('GEMINI_API_KEY')
+
+import google.generativeai as genai
+
+def get_gemini_output(prompt, temperature=0.6):
+
+    genai.configure(api_key= os.environ['GEMINI_API_KEY'])
+    model =  genai.GenerativeModel(model_name='gemini-pro')
+    answer = model.generate_content(prompt,
+                                    generation_config=genai.types.GenerationConfig(
+        temperature=0.6))
+
+    return answer.text
+
+"""## Build an end-to-end RAG powered Voice Assistant
+"""
+
+ip = IndicProcessor(inference=True)
+
+# Commented out IPython magic to ensure Python compatibility.
+# # %%capture
+ 
+en2indic_tokenizer = AutoTokenizer.from_pretrained("ai4bharat/indictrans2-en-indic-dist-200M", trust_remote_code=True)
+en2indic_model = AutoModelForSeq2SeqLM.from_pretrained("ai4bharat/indictrans2-en-indic-dist-200M", trust_remote_code=True)
+
+
+# Commented out IPython magic to ensure Python compatibility.
+# # %%capture
+ 
+indic2en_tokenizer = AutoTokenizer.from_pretrained("ai4bharat/indictrans2-indic-en-dist-200M", trust_remote_code=True)
+indic2en_model = AutoModelForSeq2SeqLM.from_pretrained("ai4bharat/indictrans2-indic-en-dist-200M", trust_remote_code=True)
+
+
+model_tokenizer_config = {
+    "en2indic": {
+        "tokenizer": en2indic_tokenizer,
+        "model": en2indic_model,
+    },
+    "indic2en": {
+        "tokenizer": indic2en_tokenizer,
+        "model": indic2en_model,
+    }
+}
+
+def indic_translate(src_lang: str, tgt_lang: str, sents_to_translate: list):
+
+  lang_map = {
+    "punjabi": "pan_Guru",
+    "bengali": "ben_Beng",
+    "malayalam": "mal_Mlym",
+    "marathi": "mar_Deva",
+    "tamil": "tam_Taml",
+    "gujarati": "guj_Gujr",
+    "telugu": "tel_Telu",
+    "hindi": "hin_Deva",
+    "kannada": "kan_Knda",
+    "odia": "ory_Orya",
+    "english": "eng_Latn"
+    }
+
+  src_lang = lang_map[src_lang]
+  tgt_lang = lang_map[tgt_lang]
+
+  if src_lang == "eng_Latn":
+    tokenizer = model_tokenizer_config["en2indic"]["tokenizer"]
+    model = model_tokenizer_config["en2indic"]["model"]
+
+    print(f"Using en2indic, src_lang: {src_lang}, tgt_lang: {tgt_lang}")
+
+  else:
+    tokenizer = model_tokenizer_config["indic2en"]["tokenizer"]
+    model = model_tokenizer_config["indic2en"]["model"]
+
+    print(f"Using indic2en, src_lang: {src_lang}, tgt_lang: {tgt_lang}")
+
+
+  batch = ip.preprocess_batch(sents_to_translate, src_lang=src_lang, tgt_lang=tgt_lang, show_progress_bar=False)
+  batch = tokenizer(batch, padding="longest", truncation=True, max_length=256, return_tensors="pt")
+
+  with inference_mode():
+      print("Generating...")
+      outputs = model.generate(**batch, num_beams=5, num_return_sequences=1, max_length=256)
+
+  with tokenizer.as_target_tokenizer():
+      outputs = tokenizer.batch_decode(outputs, skip_special_tokens=True, clean_up_tokenization_spaces=True)
+
+  if tgt_lang != "en_Latn":
+    print(f"Postprocessing for {tgt_lang}")
+    outputs = ip.postprocess_batch(outputs, lang=tgt_lang)
+
+
+  return outputs
+
+def download_ai4b_tts_model(lang: str):
+
+  lang_map = {
+      "odia": "or",
+      "hindi": "hi",
+      "tamil": "ta",
+      "telugu": "te",
+      "punjabi": "pa",
+      "kannada": "kn",
+      "bengali": "bn",
+      "marathi": "mr",
+      "gujarati": "gu",
+      "malayalam": "ml",
+  }
+
+  selected_lang = lang_map[lang]
+
+  download_path = f"/{selected_lang}.zip"
+
+  if os.path.exists(download_path):
+    print(f"IndicTTS Model for {lang} already exists.")
+
+def run_tts(text, tts_lang):
+
+  lang_map = {
+      "odia": "or",
+      "hindi": "hi",
+      "tamil": "ta",
+      "telugu": "te",
+      "punjabi": "pa",
+      "kannada": "kn",
+      "bengali": "bn",
+      "marathi": "mr",
+      "gujarati": "gu",
+      "malayalam": "ml",
+  }
+
+  download_ai4b_tts_model(lang=tts_lang)
+
+  tts_lang = lang_map[tts_lang]
+  print(f"Lang code: {tts_lang}")
+
+
+  tts_command = f'python3 -m TTS.bin.synthesize --text "{text}" \
+    --model_path /models/v1/{tts_lang}/fastpitch/best_model.pth \
+    --config_path /models/v1/{tts_lang}/fastpitch/config.json \
+    --vocoder_path /models/v1/{tts_lang}/hifigan/best_model.pth \
+    --vocoder_config_path /models/v1/{tts_lang}/hifigan/config.json \
+    --speakers_file_path /models/v1/{tts_lang}/fastpitch/speakers.pth \
+    --out_path /tts_output.wav \
+    --speaker_idx male'
+
+  if DEVICE == "cuda":
+    tts_command += " --use_cuda True"
+    print(f"Running IndicTTS on GPU")
+
+  else:
+    print(f"Running IndicTTS on CPU")
+
+  os.system(tts_command)
+
+os.makedirs('/asr_models')
+
+def download_ai4b_asr_model(lang: str):
+
+  available_langs = {
+      "odia": "or",
+      "hindi": "hi",
+      "tamil": "ta",
+      "telugu": "te",
+      "punjabi": "pa",
+      "kannada": "kn",
+      "bengali": "bn",
+      "marathi": "mr",
+      "gujarati": "gu",
+      "malayalam": "ml",
+  }
+
+  download_path = f"/asr_models/ai4b_indicConformer_{available_langs[lang]}.nemo"
+  print(f"Downloaded ASR model path: {download_path}")
+
+  if os.path.exists(download_path):
+      print(f"Model for {lang} already exists.")
+
+  elif lang not in available_langs:
+      raise ValueError(f"Invalid language code: {lang}")
+
+  return download_path
+
+import librosa
+
+def preprocess_audio(audio_path):
+    audio, sr = librosa.load(audio_path, sr=None, mono=True)
+    return audio, sr
+
+def transcribe(audio: str, lang: str):
+    audio, sr = preprocess_audio(audio)
+
+    lang_map = {
+      "odia": "or",
+      "hindi": "hi",
+      "tamil": "ta",
+      "telugu": "te",
+      "punjabi": "pa",
+      "kannada": "kn",
+      "bengali": "bn",
+      "marathi": "mr",
+      "gujarati": "gu",
+      "malayalam": "ml",
+    }
+
+    download_path = download_ai4b_asr_model(lang=lang)
+
+    asr_model = nemo_asr.models.ASRModel.restore_from(
+          download_path, map_location=DEVICE
+    )
+
+    transcription = asr_model.transcribe(audio, batch_size=1, language_id=lang_map[lang])[0][0]
+    print(f"Transcription: {transcription}")
+
+    return transcription
+
+def query_vector_db(query):
+  # Combine the top-3 similar documents from the vectorDB
+  result = " ".join([result.page_content for result in faiss.similarity_search(query, k=3)])
+
+  return result
+
+from langchain_core.prompts import PromptTemplate
+
+def process_user_query(user_query, retrieved_doc):
+
+  prompt_template = PromptTemplate.from_template(
+    "You are a chatbot , which provides information to user based on their queries, \
+    the user asks: {user_query}, The information from the related query is: {retrieved_doc}. \
+    Now give the output based on the query and relevant information that i provided, written in a structured, well-formatted and concise way. \
+    The length of the output should be no more than 70 words, must be in 5 lines."
+  )
+
+  prompt = prompt_template.format(user_query=user_query, retrieved_doc=retrieved_doc)
+
+  processed_doc = get_gemini_output(prompt)
+  print(processed_doc)
+
+  return processed_doc
+
+#Context awareness
+from collections import deque
+
+class ContextManger:
+  def __init__(self,max_history=7):
+    self.history = deque(maxlen=max_history)
+
+  def add_interaction(self,query,response):
+    self.history.append((query,response))
+
+  def get_context(self):
+    return list(self.history)
+
+context_manager = ContextManger()
+
+# context = context_manager.get_context()
+#     contexulized_query = f"Previous context: {context} \n\nCurrent query: {indic_to_en}"
+
+import traceback
+
+def process_gradio_input(audio, user_lang):
+  try:
+    # Use IndicASR to transcribe the input audio
+    print(f"Transcribing...")
+    query_transcription = transcribe(audio, lang=user_lang)
+
+    # Convert the Indic text from transcription to English, so that GPT-3.5 can process it
+    print(f"Translating indic to en..")
+    indic_to_en = indic_translate(src_lang=user_lang, tgt_lang="english", sents_to_translate=[query_transcription])[0]
+
+    # context_manager = ContextManager()
+
+    context = context_manager.get_context()
+    contexulized_query = f"Previous context: {context} \n\nCurrent query: {indic_to_en}"
+
+    # Query the Vector DB to get the relevant document from the query
+    print(f"Querying vector db")
+    retrieved_doc = query_vector_db(contexulized_query)
+
+    # Extract relevant information from the retrieved document
+    print(f"Processing user query")
+    processed_doc = process_user_query(user_query=contexulized_query, retrieved_doc=retrieved_doc)
+
+    context_manager.add_interaction(indic_to_en, processed_doc)
+
+    # Break the document into chunks for faster batch processing
+    print(f"Breaking document into chunks..")
+    processed_doc_chunks = processed_doc.strip().split(". ")
+    processed_doc_chunks = [f"{chunk}." for chunk in processed_doc_chunks if chunk != ""]
+
+    # Translate the the extracted information back to Indic language
+    print(f"Translating en to indic..")
+    en_to_indic_chunks = indic_translate(src_lang="english", tgt_lang=user_lang, sents_to_translate=processed_doc_chunks)
+    en_to_indic_doc = " ".join(en_to_indic_chunks)
+    print(f"en_to_indic_doc: {en_to_indic_doc}")
+
+    # Run IndicTTS to generate audio
+    print(f"Running TTS to generate audio..")
+    run_tts(text=en_to_indic_doc, tts_lang=user_lang)
+    print("Finished running TTS")
+
+    audio_outfile_path = "/content/tts_output.wav"
+
+
+    return en_to_indic_doc, audio_outfile_path
+
+  except Exception as e:
+        error_message = f"An error occurred: {str(e)}\n\nTraceback:\n{traceback.format_exc()}"
+        print(error_message)
+        return error_message, None
+
+def launch_gradio_app(show_log=False):
+
+  languages = ["hindi", "odia", "tamil", "telugu", "punjabi", "kannada", "bengali", "marathi", "gujarati", "malayalam"]
+
+  iface = gr.Interface(
+      fn=process_gradio_input,
+      inputs=[
+          gr.Audio(sources=['upload', 'microphone'], type="filepath", show_download_button=True),  # Input audio
+          gr.Dropdown(languages, label="Language", value="hindi"),  # Language selection
+      ],
+      outputs=["text", "audio"],
+      allow_flagging="never",
+      title="Farmer's Voice Assistant 🧑‍🌾 Powered by AI4Bharat Tech",
+      description="Know about latest farming schemes, this system is powered by tools from AI4Bharat, like IndicASR, IndicTTS and IndicTrans",
+  )
+
+  iface.launch(debug=show_log)
+
+launch_gradio_app(show_log=True)