Spaces:

ShadowTEM
/

Chatbot

Sleeping

Created a variable to change the number of vectordb results easier

e484140 verified over 1 year ago

9.46 kB

import os
import time
import json
import asyncio
import numpy as np
import soundfile as sf
from datetime import datetime
# from pydub import AudioSegment

# External dependencies.
from flask import Flask, request, jsonify, send_file,render_template

# Import your modules.
from app.stt import AudioProcessor
from app.llm import LLMProcessor
from app.vectorstore import vectorstore
from app.tts import tts
from app.utils import remove_stars


# Set a custom user agent.
os.environ["USER_AGENT"] = "my-app/1.0"

# Rename your class to avoid confusion with Flask's app instance.
class CancerApp:
    def __init__(self, vectorstore_index_path):
        self.audio_processor = AudioProcessor()
        self.llm_processor = LLMProcessor()
        self.vectorstore = vectorstore(vectorstore_index_path, Initlize_with=3)
        self.vectorstore_index = self.vectorstore.load_vectorstore(vectorstore_index_path)
        # Run an initial search to load vectorstore contents.
        self.vectorstore.search_vectorstore("cancer", 1)
        self.tts = tts()
        self.Number_of_search_result_from_vectordb = 10

    def text_to_speech(self, message, output_filename, Saved_response=""):
        search_results = self.vectorstore.search_vectorstore(message, self.Number_of_search_result_from_vectordb)
        query = Saved_response
        for i, result in enumerate(search_results, 1):
            query += f"\n{i}. {result}"
        
        try:
            gemini_response = self.llm_processor.call_gemini_llm("gemini-2.0-flash", message, query)
        except Exception as e:
            print("Gemini error:", e)
            gemini_response = f"Error: {str(e)}"
        gemini_response = remove_stars(gemini_response)
        Saved_response += "message: " + message + "\n" + "gemini_response: " + gemini_response + "\n"
        self.audio_processor.log_conversation(message, bot_text=gemini_response)
        audio_data, sample_rate = asyncio.run(self.tts.cpu_stream_to_audio(gemini_response))
        sf.write(output_filename, audio_data, sample_rate)
        return Saved_response

    def speech_to_speech(self, audiofile_path, output_filename, Saved_response=""):
        transcription = self.audio_processor.transcribe_audio(audiofile_path, language="en")
        search_results = self.vectorstore.search_vectorstore(transcription, self.Number_of_search_result_from_vectordb)
        query = Saved_response  
        for i, result in enumerate(search_results, 1):
            query += f"\n{i}. {result}"
        try:
            gemini_response = self.llm_processor.call_gemini_llm("gemini-2.0-flash",  transcription, query)
        except Exception as e:
            print("Gemini error:", e)
            gemini_response = f"Error: {str(e)}"
        gemini_response = remove_stars(gemini_response)
        Saved_response += "transcription: " + transcription + "\n" + "gemini_response: " + gemini_response + "\n"
        self.audio_processor.log_conversation(transcription, bot_text=gemini_response)
        audio_data, sample_rate = asyncio.run(self.tts.cpu_stream_to_audio(gemini_response))
        sf.write(output_filename, audio_data, sample_rate)
        return Saved_response

    def speech_to_text(self, audiofile_path, Saved_response=""):
        try:
            transcription = self.audio_processor.transcribe_audio(audiofile_path, language="en")
        except Exception as ex:
            print(f"Transcription error: {ex}")
            transcription = ""
        
        if not transcription:
            print("No transcription available; aborting further processing.")
            return "", Saved_response

        try:
            search_results = self.vectorstore.search_vectorstore(transcription, self.Number_of_search_result_from_vectordb)
        except Exception as ex:
            print(f"Vectorstore search error: {ex}")
            search_results = []
        
        query = f"{Saved_response}"
        for i, result in enumerate(search_results, 1):
            query += f"\n{i}. {result}"

        
        try:
            message = f"Old_conversation:{Saved_response} , message:{transcription}" 
            gemini_response = self.llm_processor.call_gemini_llm("gemini-2.0-flash", message,query)
            gemini_response = remove_stars(gemini_response)
        except Exception as e:
            print("Gemini error:", e)
            gemini_response = f"Error: {str(e)}"
        gemini_response = remove_stars(gemini_response)
        Saved_response = f"transcription: {transcription}\n" + f"gemini_response: {gemini_response}\n" + f"Old_conversation:{Saved_response}"
        self.audio_processor.log_conversation(transcription, bot_text=gemini_response)
        return gemini_response, Saved_response

    def text_to_text(self, message, Saved_response=""):
        search_results = self.vectorstore.search_vectorstore(message, self.Number_of_search_result_from_vectordb)
        query = Saved_response
        for i, result in enumerate(search_results, 1):
            query += f"\n{i}. {result}"
        
        try:
            gemini_response = self.llm_processor.call_gemini_llm("gemini-2.0-flash",message,query)
            gemini_response = remove_stars(gemini_response)
        except Exception as e:
            print("Gemini error:", e)
            gemini_response = f"Error: {str(e)}"
        Saved_response += "message: " + message + "\n" + "gemini_response: " + gemini_response + "\n"
        self.audio_processor.log_conversation(message, bot_text=gemini_response)
        return gemini_response, Saved_response

# Create Flask API instance.
flask_app = Flask(__name__)

vectordb = vectorstore(path="data/PDF/Cancer The Evolutionary Legacy .pdf", Initlize_with=1)
print("done creating vectorstore")
vectordb.add_to_vectorstore_from_pdf("data/PDF/cancer_dictionary.pdf")
print("done adding cancer dictionary")

vectordb.add_to_vectorstore_from_pdf("data/PDF/Colon and Other GastrointestinalCancers.pdf")
print("done adding colon and other gastrointestinal cancers")

vectordb.add_to_vectorstore_from_pdf("data/PDF/Medical Dictionary.pdf")
print("done adding medical dictionary")

vectordb.add_to_vectorstore_from_pdf("data/PDF/Molecular biology of cancer.pdf")
print("done adding molecular biology of cancer")

vectordb.add_to_vectorstore_from_pdf("data/PDF/The biology of cancer.pdf")
print("done adding the biology of cancer")

vectordb.add_to_vectorstore_from_pdf("data/PDF/Being mortal _ medicine and what matters in the end .pdf")
print("done adding the biology of cancer")

vectordb.add_to_vectorstore_from_pdf("data/PDF/The Emperor of All Maladies_ A Biography of Cancer.pdf")
print("done adding the biology of cancer")

print("finished vectorization")

vectordb_path = "VectorDB/vectorstore_mainV2"
cancer_app_instance = CancerApp(vectordb_path)

#Endpoint for text-to-text processing.
@flask_app.route('/text_to_text', methods=['POST'])
def api_text_to_text():
    data = request.get_json()
    message = data.get("message", "")
    saved_response = data.get("Saved_response", "")
    response, saved_response = cancer_app_instance.text_to_text(message, Saved_response=saved_response)
    return jsonify({"gemini_response": response, "Saved_response": saved_response})



# Endpoint for text-to-speech processing.
@flask_app.route('/text_to_speech', methods=['POST'])
def api_text_to_speech():
    data = request.get_json()
    message = data.get("message", "")
    output_filename = data.get("output_filename", "output.wav")
    saved_response = data.get("Saved_response", "")
    print("got the data")
    # Generate the WAV file from the text
    saved_response = cancer_app_instance.text_to_speech(message, output_filename, Saved_response=saved_response)
    # Return the file instead of JSON
    return send_file(output_filename, mimetype="audio/wav", as_attachment=True)

# Endpoint for speech-to-text processing.
@flask_app.route('/speech_to_text', methods=['POST'])
def api_speech_to_text():
    if 'audiofile' not in request.files:
        return jsonify({"error": "No audio file provided"}), 400
    audio_file = request.files['audiofile']
    # Save uploaded file temporarily.
    audio_ext = audio_file.filename.split('.')[-1]
    audio_path = f"temp_audio_input.{audio_ext}"
    audio_file.save(audio_path)
    saved_response = request.form.get("Saved_response", "")
    gemini_response, saved_response = cancer_app_instance.speech_to_text(audio_path, Saved_response=saved_response)
    os.remove(audio_path)
    return jsonify({"gemini_response": gemini_response, "Saved_response": saved_response})

# Endpoint for speech-to-speech processing.
@flask_app.route('/speech_to_speech', methods=['POST'])
def api_speech_to_speech():
    if 'audiofile' not in request.files:
        return jsonify({"error": "No audio file provided"}), 400
    audio_file = request.files['audiofile']
    audio_ext = audio_file.filename.split('.')[-1]
    audio_path = f"temp_audio_input.{audio_ext}"
    audio_file.save(audio_path)
    output_filename = request.form.get("output_filename", "speech_output.wav")
    saved_response = request.form.get("Saved_response", "")
    saved_response = cancer_app_instance.speech_to_speech(audio_path, output_filename, Saved_response=saved_response)
    os.remove(audio_path)
    # Return the output file for download.
    return send_file(output_filename, as_attachment=True)

if __name__ == '__main__':
    flask_app.run(host='0.0.0.0', port=7860, debug=True)

	import os
	import time
	import json
	import asyncio
	import numpy as np
	import soundfile as sf
	from datetime import datetime
	# from pydub import AudioSegment

	# External dependencies.
	from flask import Flask, request, jsonify, send_file,render_template

	# Import your modules.
	from app.stt import AudioProcessor
	from app.llm import LLMProcessor
	from app.vectorstore import vectorstore
	from app.tts import tts
	from app.utils import remove_stars


	# Set a custom user agent.
	os.environ["USER_AGENT"] = "my-app/1.0"

	# Rename your class to avoid confusion with Flask's app instance.
	class CancerApp:
	def __init__(self, vectorstore_index_path):
	self.audio_processor = AudioProcessor()
	self.llm_processor = LLMProcessor()
	self.vectorstore = vectorstore(vectorstore_index_path, Initlize_with=3)
	self.vectorstore_index = self.vectorstore.load_vectorstore(vectorstore_index_path)
	# Run an initial search to load vectorstore contents.
	self.vectorstore.search_vectorstore("cancer", 1)
	self.tts = tts()
	self.Number_of_search_result_from_vectordb = 10

	def text_to_speech(self, message, output_filename, Saved_response=""):
	search_results = self.vectorstore.search_vectorstore(message, self.Number_of_search_result_from_vectordb)
	query = Saved_response
	for i, result in enumerate(search_results, 1):
	query += f"\n{i}. {result}"

	try:
	gemini_response = self.llm_processor.call_gemini_llm("gemini-2.0-flash", message, query)
	except Exception as e:
	print("Gemini error:", e)
	gemini_response = f"Error: {str(e)}"
	gemini_response = remove_stars(gemini_response)
	Saved_response += "message: " + message + "\n" + "gemini_response: " + gemini_response + "\n"
	self.audio_processor.log_conversation(message, bot_text=gemini_response)
	audio_data, sample_rate = asyncio.run(self.tts.cpu_stream_to_audio(gemini_response))
	sf.write(output_filename, audio_data, sample_rate)
	return Saved_response

	def speech_to_speech(self, audiofile_path, output_filename, Saved_response=""):
	transcription = self.audio_processor.transcribe_audio(audiofile_path, language="en")
	search_results = self.vectorstore.search_vectorstore(transcription, self.Number_of_search_result_from_vectordb)
	query = Saved_response
	for i, result in enumerate(search_results, 1):
	query += f"\n{i}. {result}"
	try:
	gemini_response = self.llm_processor.call_gemini_llm("gemini-2.0-flash", transcription, query)
	except Exception as e:
	print("Gemini error:", e)
	gemini_response = f"Error: {str(e)}"
	gemini_response = remove_stars(gemini_response)
	Saved_response += "transcription: " + transcription + "\n" + "gemini_response: " + gemini_response + "\n"
	self.audio_processor.log_conversation(transcription, bot_text=gemini_response)
	audio_data, sample_rate = asyncio.run(self.tts.cpu_stream_to_audio(gemini_response))
	sf.write(output_filename, audio_data, sample_rate)
	return Saved_response

	def speech_to_text(self, audiofile_path, Saved_response=""):
	try:
	transcription = self.audio_processor.transcribe_audio(audiofile_path, language="en")
	except Exception as ex:
	print(f"Transcription error: {ex}")
	transcription = ""

	if not transcription:
	print("No transcription available; aborting further processing.")
	return "", Saved_response

	try:
	search_results = self.vectorstore.search_vectorstore(transcription, self.Number_of_search_result_from_vectordb)
	except Exception as ex:
	print(f"Vectorstore search error: {ex}")
	search_results = []

	query = f"{Saved_response}"
	for i, result in enumerate(search_results, 1):
	query += f"\n{i}. {result}"


	try:
	message = f"Old_conversation:{Saved_response} , message:{transcription}"
	gemini_response = self.llm_processor.call_gemini_llm("gemini-2.0-flash", message,query)
	gemini_response = remove_stars(gemini_response)
	except Exception as e:
	print("Gemini error:", e)
	gemini_response = f"Error: {str(e)}"
	gemini_response = remove_stars(gemini_response)
	Saved_response = f"transcription: {transcription}\n" + f"gemini_response: {gemini_response}\n" + f"Old_conversation:{Saved_response}"
	self.audio_processor.log_conversation(transcription, bot_text=gemini_response)
	return gemini_response, Saved_response

	def text_to_text(self, message, Saved_response=""):
	search_results = self.vectorstore.search_vectorstore(message, self.Number_of_search_result_from_vectordb)
	query = Saved_response
	for i, result in enumerate(search_results, 1):
	query += f"\n{i}. {result}"

	try:
	gemini_response = self.llm_processor.call_gemini_llm("gemini-2.0-flash",message,query)
	gemini_response = remove_stars(gemini_response)
	except Exception as e:
	print("Gemini error:", e)
	gemini_response = f"Error: {str(e)}"
	Saved_response += "message: " + message + "\n" + "gemini_response: " + gemini_response + "\n"
	self.audio_processor.log_conversation(message, bot_text=gemini_response)
	return gemini_response, Saved_response

	# Create Flask API instance.
	flask_app = Flask(__name__)

	vectordb = vectorstore(path="data/PDF/Cancer The Evolutionary Legacy .pdf", Initlize_with=1)
	print("done creating vectorstore")
	vectordb.add_to_vectorstore_from_pdf("data/PDF/cancer_dictionary.pdf")
	print("done adding cancer dictionary")

	vectordb.add_to_vectorstore_from_pdf("data/PDF/Colon and Other GastrointestinalCancers.pdf")
	print("done adding colon and other gastrointestinal cancers")

	vectordb.add_to_vectorstore_from_pdf("data/PDF/Medical Dictionary.pdf")
	print("done adding medical dictionary")

	vectordb.add_to_vectorstore_from_pdf("data/PDF/Molecular biology of cancer.pdf")
	print("done adding molecular biology of cancer")

	vectordb.add_to_vectorstore_from_pdf("data/PDF/The biology of cancer.pdf")
	print("done adding the biology of cancer")

	vectordb.add_to_vectorstore_from_pdf("data/PDF/Being mortal _ medicine and what matters in the end .pdf")
	print("done adding the biology of cancer")

	vectordb.add_to_vectorstore_from_pdf("data/PDF/The Emperor of All Maladies_ A Biography of Cancer.pdf")
	print("done adding the biology of cancer")

	print("finished vectorization")

	vectordb_path = "VectorDB/vectorstore_mainV2"
	cancer_app_instance = CancerApp(vectordb_path)

	#Endpoint for text-to-text processing.
	@flask_app.route('/text_to_text', methods=['POST'])
	def api_text_to_text():
	data = request.get_json()
	message = data.get("message", "")
	saved_response = data.get("Saved_response", "")
	response, saved_response = cancer_app_instance.text_to_text(message, Saved_response=saved_response)
	return jsonify({"gemini_response": response, "Saved_response": saved_response})



	# Endpoint for text-to-speech processing.
	@flask_app.route('/text_to_speech', methods=['POST'])
	def api_text_to_speech():
	data = request.get_json()
	message = data.get("message", "")
	output_filename = data.get("output_filename", "output.wav")
	saved_response = data.get("Saved_response", "")
	print("got the data")
	# Generate the WAV file from the text
	saved_response = cancer_app_instance.text_to_speech(message, output_filename, Saved_response=saved_response)
	# Return the file instead of JSON
	return send_file(output_filename, mimetype="audio/wav", as_attachment=True)

	# Endpoint for speech-to-text processing.
	@flask_app.route('/speech_to_text', methods=['POST'])
	def api_speech_to_text():
	if 'audiofile' not in request.files:
	return jsonify({"error": "No audio file provided"}), 400
	audio_file = request.files['audiofile']
	# Save uploaded file temporarily.
	audio_ext = audio_file.filename.split('.')[-1]
	audio_path = f"temp_audio_input.{audio_ext}"
	audio_file.save(audio_path)
	saved_response = request.form.get("Saved_response", "")
	gemini_response, saved_response = cancer_app_instance.speech_to_text(audio_path, Saved_response=saved_response)
	os.remove(audio_path)
	return jsonify({"gemini_response": gemini_response, "Saved_response": saved_response})

	# Endpoint for speech-to-speech processing.
	@flask_app.route('/speech_to_speech', methods=['POST'])
	def api_speech_to_speech():
	if 'audiofile' not in request.files:
	return jsonify({"error": "No audio file provided"}), 400
	audio_file = request.files['audiofile']
	audio_ext = audio_file.filename.split('.')[-1]
	audio_path = f"temp_audio_input.{audio_ext}"
	audio_file.save(audio_path)
	output_filename = request.form.get("output_filename", "speech_output.wav")
	saved_response = request.form.get("Saved_response", "")
	saved_response = cancer_app_instance.speech_to_speech(audio_path, output_filename, Saved_response=saved_response)
	os.remove(audio_path)
	# Return the output file for download.
	return send_file(output_filename, as_attachment=True)

	if __name__ == '__main__':
	flask_app.run(host='0.0.0.0', port=7860, debug=True)