Spaces:
Runtime error
Runtime error
| import os | |
| import dotenv | |
| import gradio as gr | |
| import lancedb | |
| import logging | |
| from langchain.embeddings.cohere import CohereEmbeddings | |
| from langchain.llms import Cohere | |
| from langchain.prompts import PromptTemplate | |
| from langchain.chains import RetrievalQA | |
| from langchain.vectorstores import LanceDB | |
| from langchain.text_splitter import RecursiveCharacterTextSplitter | |
| # Assume these loaders are implemented based on your specific requirements | |
| from custom_document_loaders import TextLoader, PyPDFLoader, DocxLoader, ImageLoader | |
| import argostranslate.package | |
| import argostranslate.translate | |
| import shutil | |
| # Configuration and Logging | |
| dotenv.load_dotenv(".env") | |
| DB_PATH = "/tmp/lancedb" | |
| COHERE_MODEL_NAME = "multilingual-22-12" | |
| LANGUAGE_ISO_CODES = {"English": "en", "Hindi": "hi", "Turkish": "tr", "French": "fr"} | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| # Initialize argostranslate | |
| argostranslate.package.update_package_index() | |
| def initialize_documents_and_embeddings(input_file_path): | |
| logger.info(f"Processing file: {input_file_path}") | |
| file_extension = os.path.splitext(input_file_path)[1].lower() | |
| loader = None | |
| if file_extension in [".txt"]: | |
| loader = TextLoader(input_file_path) | |
| elif file_extension in [".pdf"]: | |
| loader = PyPDFLoader(input_file_path) | |
| elif file_extension in [".doc", ".docx"]: | |
| loader = DocxLoader(input_file_path) | |
| elif file_extension in [".jpg", ".jpeg", ".png"]: | |
| loader = ImageLoader(input_file_path) | |
| else: | |
| raise ValueError("Unsupported file type. Supported files are .txt, .pdf, .docx, and image files.") | |
| documents = loader.load() | |
| text_splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50) | |
| texts = text_splitter.split_documents(documents) | |
| embeddings = CohereEmbeddings(model=COHERE_MODEL_NAME) | |
| return texts, embeddings | |
| def initialize_database(texts, embeddings): | |
| if os.path.exists(DB_PATH): | |
| shutil.rmtree(DB_PATH) # Ensure a fresh start | |
| db = lancedb.connect(DB_PATH) | |
| table = db.create_table("multiling-rag", mode="overwrite") | |
| return LanceDB.from_documents(texts, embeddings, connection=table) | |
| def translate_text(text, from_code, to_code): | |
| installed_languages = argostranslate.translate.get_installed_languages() | |
| from_lang = next((lang for lang in installed_languages if lang.code == from_code), None) | |
| to_lang = next((lang for lang in installed_languages if lang.code == to_code), None) | |
| if not from_lang or not to_lang: | |
| logger.error("Translation languages not installed.") | |
| return "Translation error" | |
| translation = from_lang.get_translation(to_lang) | |
| return translation.translate(text) | |
| def answer_question(question, input_language, output_language, db): | |
| try: | |
| input_lang_code = LANGUAGE_ISO_CODES[input_language] | |
| output_lang_code = LANGUAGE_ISO_CODES[output_language] | |
| question_in_english = translate_text(question, input_lang_code, "en") if input_language != "English" else question | |
| # Simplified retrieval and response logic for demonstration | |
| response = "This is a simulated response based on the question." | |
| result_in_target_language = translate_text(response, "en", output_lang_code) if output_language != "English" else response | |
| return result_in_target_language | |
| except Exception as e: | |
| logger.error(f"Error in answer_question: {str(e)}") | |
| return "An error occurred while processing your question." | |
| def document_analysis_and_feedback(document_path, feedback): | |
| # Placeholder for document analysis logic | |
| response = "Document analysis and feedback processing is not fully implemented." | |
| return response | |
| def setup_gradio_interface(db): | |
| with gr.Blocks() as demo: | |
| gr.Markdown("# Multilingual Health and Wellness Chatbot") | |
| with gr.Tab("Ask a Question"): | |
| with gr.Row(): | |
| input_language = gr.Dropdown(list(LANGUAGE_ISO_CODES.keys()), label="Input Language") | |
| output_language = gr.Dropdown(list(LANGUAGE_ISO_CODES.keys()), label="Output Language") | |
| question = gr.Textbox(label="Your question") | |
| answer = gr.Textbox(label="Answer") | |
| question.submit(lambda q, i, o: answer_question(q, i, o, db), inputs=[question, input_language, output_language], outputs=answer) | |
| with gr.Tab("Upload Document"): | |
| with gr.Row(): | |
| document = gr.File(label="Upload your health document") | |
| feedback_box = gr.Textbox(label="Feedback (optional)") | |
| upload_response = gr.Textbox(label="Analysis Result") | |
| document.submit(document_analysis_and_feedback, inputs=[document, feedback_box], outputs=upload_response) | |
| return demo | |
| def main(): | |
| INPUT_FILE_PATH = "sample-text.txt" # Placeholder file path | |
| texts, embeddings = initialize_documents_and_embeddings(INPUT_FILE_PATH) | |
| db = initialize_database(texts, embeddings) | |
| demo = setup_gradio_interface(db) | |
| demo.launch() | |
| if __name__ == "__main__": | |
| main() | |