import gradio as gr from phi.agent import Agent from phi.model.groq import Groq import os import logging from sentence_transformers import CrossEncoder from backend.semantic_search import table, retriever import numpy as np from time import perf_counter import requests from jinja2 import Environment, FileSystemLoader from pathlib import Path import time # Set up logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # API Key setup api_key = os.getenv("GROQ_API_KEY") if not api_key: gr.Warning("GROQ_API_KEY not found. Set it in 'Repository secrets'.") logger.error("GROQ_API_KEY not found.") api_key = "" else: os.environ["GROQ_API_KEY"] = api_key # Bhashini API setup bhashini_api_key = os.getenv("API_KEY", "").strip() bhashini_user_id = os.getenv("USER_ID", "").strip() # 🎯 MODEL LIST: Ordered by parameter size (largest β†’ smallest) # Fallback chain: if rate limit (429) or model error, try next model MODEL_CHAIN = [ {"id": "llama-3.3-70b-versatile", "name": "Llama 3.3 70B", "priority": 1}, {"id": "qwen/qwen3-32b", "name": "Qwen3 32B", "priority": 2}, {"id": "meta-llama/llama-4-scout-17b-16e-instruct", "name": "Llama-4 Scout 17B", "priority": 3}, {"id": "groq/compound", "name": "Groq Compound", "priority": 4}, {"id": "gemma2-9b-it", "name": "Gemma2 9B", "priority": 5}, {"id": "llama-3.1-8b-instant", "name": "Llama 3.1 8B Instant", "priority": 6}, ] def bhashini_translate(text: str, from_code: str = "en", to_code: str = "hi") -> dict: """Translates text using Bhashini API""" if not text.strip(): return {"status_code": 400, "message": "Input text is empty", "translated_content": None} print(f'Translating: "{text[:50]}..." from {from_code} to {to_code}') url = 'https://meity-auth.ulcacontrib.org/ulca/apis/v0/model/getModelsPipeline' headers = { "Content-Type": "application/json", "userID": bhashini_user_id, "ulcaApiKey": bhashini_api_key } payload = { "pipelineTasks": [{"taskType": "translation", "config": {"language": {"sourceLanguage": from_code, "targetLanguage": to_code}}}], "pipelineRequestConfig": {"pipelineId": "64392f96daac500b55c543cd"} } try: response = requests.post(url, json=payload, headers=headers, timeout=30) if response.status_code != 200: logger.error(f"Bhashini initial request failed: {response.status_code}") return {"status_code": response.status_code, "message": "Translation request failed", "translated_content": None} response_data = response.json() if "pipelineInferenceAPIEndPoint" not in response_data: return {"status_code": 400, "message": "Unexpected API response", "translated_content": None} service_id = response_data["pipelineResponseConfig"][0]["config"][0]["serviceId"] callback_url = response_data["pipelineInferenceAPIEndPoint"]["callbackUrl"] headers2 = { "Content-Type": "application/json", response_data["pipelineInferenceAPIEndPoint"]["inferenceApiKey"]["name"]: response_data["pipelineInferenceAPIEndPoint"]["inferenceApiKey"]["value"] } compute_payload = { "pipelineTasks": [{"taskType": "translation", "config": {"language": {"sourceLanguage": from_code, "targetLanguage": to_code}, "serviceId": service_id}}], "inputData": {"input": [{"source": text}], "audio": [{"audioContent": None}]} } compute_response = requests.post(callback_url, json=compute_payload, headers=headers2, timeout=60) if compute_response.status_code != 200: return {"status_code": compute_response.status_code, "message": "Translation failed", "translated_content": None} compute_response_data = compute_response.json() translated_content = compute_response_data["pipelineResponse"][0]["output"][0]["target"] return {"status_code": 200, "message": "Success", "translated_content": translated_content} except Exception as e: logger.error(f"Bhashini translation error: {e}") return {"status_code": 500, "message": f"Translation error: {str(e)}", "translated_content": None} def create_agent_with_model(model_id: str) -> Agent: """Create a Phi Agent with specified Groq model - SOCIAL SCIENCE FOCUS""" return Agent( name="Social Science Education Assistant", role="You are a helpful social science tutor for 10th-grade students", instructions=[ "You are an expert social science teacher specializing in 10th-grade curriculum.", "Provide clear, accurate, and age-appropriate explanations on History, Geography, Civics, and Economics.", "Use simple language and real-world examples that students can relate to.", "Focus on key topics: Indian History, World History, Physical & Human Geography, Democratic Politics, Economic Development, and Contemporary Issues.", "Structure responses with headings, bullet points, and timelines when helpful.", "Encourage critical thinking about society, governance, and sustainable development.", "If uncertain, acknowledge limitations and suggest reliable sources like NCERT textbooks.", "Promote values of democracy, secularism, equality, and environmental responsibility." ], model=Groq(id=model_id, api_key=api_key), markdown=True, timeout=60 ) def generate_with_fallback(prompt: str) -> tuple[str, str]: """ Try models in descending size order. Returns: (response_text, model_used) """ last_error = None for model_config in MODEL_CHAIN: model_id = model_config["id"] model_name = model_config["name"] priority = model_config["priority"] logger.info(f"πŸ”„ Trying model #{priority}: {model_name} ({model_id})") try: agent = create_agent_with_model(model_id) response = agent.run(prompt) response_text = response.content if hasattr(response, 'content') else str(response) logger.info(f"βœ… Success with {model_name}") return response_text.strip(), model_name except Exception as e: error_msg = str(e).lower() last_error = e # Check for errors that warrant fallback to next model should_fallback = any([ "429" in error_msg, "rate limit" in error_msg, "too many requests" in error_msg, "model_decommissioned" in error_msg, "model_not_found" in error_msg, "invalid_request_error" in error_msg and "model" in error_msg, "timeout" in error_msg and priority < len(MODEL_CHAIN) ]) if should_fallback: logger.warning(f"⚠️ {model_name} failed ({type(e).__name__}), falling back to next model...") time.sleep(0.5 * priority) continue else: logger.error(f"❌ {model_name} failed with non-fallback error: {e}") break # All models failed error_details = f"{type(last_error).__name__}: {str(last_error)}" if last_error else "Unknown error" logger.error(f"πŸ’₯ All models failed. Last error: {error_details}") fallback_response = ( "⚠️ I'm experiencing high demand right now. Please try again in a moment.\n\n" f"*Technical note: Unable to connect to available models ({error_details})*" ) return fallback_response, "fallback" def retrieve_and_generate_response(query, cross_encoder_choice, history=None): """Generate response using semantic search + multi-model LLM fallback""" top_rerank = 25 top_k_rank = 20 if not query.strip(): return "Please provide a valid question.", [], "N/A" try: start_time = perf_counter() # Encode query and search documents query_vec = retriever.encode(query) documents = table.search(query_vec, vector_column_name="vector").limit(top_rerank).to_list() documents = [doc["text"] for doc in documents] # Re-rank documents using cross-encoder model_name = 'BAAI/bge-reranker-base' if cross_encoder_choice == '(ACCURATE) BGE reranker' else 'cross-encoder/ms-marco-MiniLM-L-6-v2' cross_encoder_model = CrossEncoder(model_name) query_doc_pair = [[query, doc] for doc in documents] cross_scores = cross_encoder_model.predict(query_doc_pair) sim_scores_argsort = list(reversed(np.argsort(cross_scores))) documents = [documents[idx] for idx in sim_scores_argsort[:top_k_rank]] # Create context from top documents context = "\n\n".join(documents[:10]) if documents else "" context = f"Context information from social science educational materials:\n{context}\n\n" # Add conversation history for context history_context = "" if history and len(history) > 0: for user_msg, bot_msg in history[-2:]: if user_msg and bot_msg: history_context += f"Previous Q: {user_msg}\nPrevious A: {bot_msg}\n" # Create full prompt - SOCIAL SCIENCE FOCUS full_prompt = f"{history_context}{context}Question: {query}\n\nPlease answer the question using the context provided above. If the context doesn't contain relevant information, use your general knowledge about 10th-grade social science topics including History, Geography, Civics, and Economics." # Generate response with model fallback chain response_text, model_used = generate_with_fallback(full_prompt) # Add model attribution to response if model_used != "fallback": response_text = f"*Powered by {model_used}*\n\n{response_text}" elapsed = perf_counter() - start_time logger.info(f"⏱️ Response generated in {elapsed:.2f}s using {model_used}") return response_text, documents, model_used except Exception as e: logger.error(f"❌ Error in response generation: {e}") return f"Error: {str(e)}", [], "error" def translate_text(selected_language, history): """Translate the last response in history to the selected language.""" iso_language_codes = { "Hindi": "hi", "Gom": "gom", "Kannada": "kn", "Dogri": "doi", "Bodo": "brx", "Urdu": "ur", "Tamil": "ta", "Kashmiri": "ks", "Assamese": "as", "Bengali": "bn", "Marathi": "mr", "Sindhi": "sd", "Maithili": "mai", "Punjabi": "pa", "Malayalam": "ml", "Manipuri": "mni", "Telugu": "te", "Sanskrit": "sa", "Nepali": "ne", "Santali": "sat", "Gujarati": "gu", "Odia": "or" } to_code = iso_language_codes.get(selected_language, "hi") response_text = history[-1][1] if history and len(history) > 0 and history[-1][1] else '' if not response_text or response_text.startswith("Error:"): return "Translation not available for error messages." # Remove model attribution line for cleaner translation clean_text = response_text.split("\n\n", 1)[-1] if "*Powered by" in response_text else response_text translation = bhashini_translate(clean_text, to_code=to_code) return translation.get('translated_content', 'Translation failed.') # Set up Jinja2 environment proj_dir = Path(__file__).parent env = Environment(loader=FileSystemLoader(proj_dir / 'templates')) template = env.get_template('template.j2') template_html = env.get_template('template_html.j2') # Gradio Interface - SOCIAL SCIENCE THEME with gr.Blocks(title="Social Science Chatbot", theme='gradio/soft', css=""" .model-badge { background: #e0f2fe; color: #0369a1; padding: 2px 8px; border-radius: 12px; font-size: 0.8em; font-weight: 500; display: inline-block; margin: 4px 0; } .gradio-container { max-width: 1200px !important; margin: auto; } """) as demo: # Header section - SOCIAL SCIENCE FOCUS with gr.Row(): with gr.Column(scale=10): gr.HTML(value="""

🌍 Welcome! I am your Social Science Friend!

Ask me anything about History, Geography, Civics & Economics!

✨ Multi-Model AI with 22-Language Translation

""") gr.HTML(value="

Developed by K.M. Ramyasri, TGT, GHS Suthukeny β€’ Open Source β€’ Free for 10th Grade Students

") gr.HTML(value="

Suggestions: ramyasriraman2019@gmail.com

") with gr.Column(scale=3): try: gr.Image(value='logo.png', height=180, width=180, show_label=False) except: gr.HTML("
πŸ—ΊοΈ
") # Model status indicator (Markdown component) model_status = gr.Markdown(value="πŸ”„ *Ready to answer with best available model*") # Chatbot chatbot = gr.Chatbot( [], elem_id="chatbot", avatar_images=( 'https://aui.atlassian.com/aui/8.8/docs/images/avatar-person.svg', 'https://huggingface.co/datasets/huggingface/brand-assets/resolve/main/hf-logo.svg' ), bubble_full_width=False, show_copy_button=True, show_share_button=True, height=400 ) # Input row with gr.Row(): msg = gr.Textbox( scale=4, show_label=False, placeholder="Ask a social science question (e.g., 'What caused the French Revolution?')", container=False, autofocus=True ) submit_btn = gr.Button(value="πŸš€ Ask", scale=1, variant="primary") # Controls with gr.Row(): cross_encoder = gr.Radio( choices=['(FAST) MiniLM-L6v2', '(ACCURATE) BGE reranker'], value='(ACCURATE) BGE reranker', label="πŸ” Document Ranking", info="Balance speed vs accuracy for source retrieval" ) language_dropdown = gr.Dropdown( choices=["Hindi", "Gom", "Kannada", "Dogri", "Bodo", "Urdu", "Tamil", "Kashmiri", "Assamese", "Bengali", "Marathi", "Sindhi", "Maithili", "Punjabi", "Malayalam", "Manipuri", "Telugu", "Sanskrit", "Nepali", "Santali", "Gujarati", "Odia"], value="Hindi", label="🌐 Translate Response To", scale=2 ) # Outputs with gr.Row(): translated_textbox = gr.Textbox(label="🌍 Translated Response", lines=3, interactive=False) prompt_html = gr.HTML(label="πŸ“š Source Documents") # Event handler function - NO .update() calls, return values instead def update_chat_and_translate(message, history, cross_encoder_choice, selected_language): """Handler that returns all outputs in correct order""" # Handle empty message if not message.strip(): return "", history, "", "", "πŸ”„ *Ready to answer with best available model*" # Generate response with fallback chain response, documents, model_used = retrieve_and_generate_response( message, cross_encoder_choice, history ) # Prepare markdown status as STRING (return value, not .update()) status_emoji = "βœ…" if model_used != "fallback" else "⚠️" model_status_text = f"{status_emoji} *Responded using: **{model_used}***" # Add to chat history history.append([message, response]) # Translate response translated_text = translate_text(selected_language, history) # Render HTML template prompt_html_content = template_html.render( documents=documents, query=message, model=model_used, timestamp=time.strftime("%H:%M:%S") ) # RETURN ALL VALUES IN EXACT ORDER MATCHING outputs=[...] # outputs=[msg, chatbot, translated_textbox, prompt_html, model_status] return ( "", # 1. msg: clear input textbox history, # 2. chatbot: updated conversation history translated_text, # 3. translated_textbox: translation result prompt_html_content, # 4. prompt_html: rendered source documents model_status_text # 5. model_status: markdown status (STRING) ) # Bind events - outputs order must match return tuple order msg.submit( update_chat_and_translate, inputs=[msg, chatbot, cross_encoder, language_dropdown], outputs=[msg, chatbot, translated_textbox, prompt_html, model_status] ) submit_btn.click( update_chat_and_translate, inputs=[msg, chatbot, cross_encoder, language_dropdown], outputs=[msg, chatbot, translated_textbox, prompt_html, model_status] ) # Clear button - return tuple matching its outputs order: [chatbot, msg, translated_textbox, prompt_html, model_status] clear = gr.Button("πŸ—‘οΈ Clear Conversation", variant="secondary") clear.click( lambda: ([], "", "", "", "πŸ”„ *Ready to answer with best available model*"), outputs=[chatbot, msg, translated_textbox, prompt_html, model_status] ) # Example questions - SOCIAL SCIENCE FOCUS gr.Examples( examples=[ "What were the main causes of the Indian National Movement?", "Explain the features of the Indian Constitution", "What is sustainable development and why is it important?", "Describe the impact of globalization on Indian economy", "What are the major physical divisions of India?", "Explain the functioning of local government in India", "What led to the rise of nationalism in Europe?", "How does the electoral process work in a democracy?" ], inputs=msg, label="πŸ’‘ Try these social science questions:" ) # Footer with technical details (collapsible) gr.Markdown("""
πŸ”§ Technical: Model Fallback Chain (Largest β†’ Smallest) | Priority | Model | Parameters | Use Case | |----------|-------|-----------|----------| | 1 | Llama 3.3 70B | ~70B | Complex reasoning, highest accuracy | | 2 | Qwen3 32B | ~32B | Strong multilingual & social science performance | | 3 | Llama-4 Scout 17B | ~17B | Balanced speed/quality | | 4 | Groq Compound | ~12B | Specialized mixture model | | 5 | Gemma2 9B | ~9B | Fast responses, good quality | | 6 | Llama 3.1 8B Instant | ~8B | ⚑ Fastest fallback, lowest cost | *If a model hits rate limits or errors, the system automatically tries the next model.*
""") if __name__ == "__main__": logger.info("πŸš€ Starting Social Science Chatbot with multi-model fallback...") logger.info(f"Model chain: {[m['name'] for m in MODEL_CHAIN]}") logger.info("πŸ“š Focus: History, Geography, Civics, Economics (10th Grade)") demo.launch(server_name="0.0.0.0", server_port=7860, show_error=True) # from phi.agent import Agent # from phi.model.groq import Groq # import os # import logging # from sentence_transformers import CrossEncoder # from backend.semantic_search import table, retriever # import numpy as np # from time import perf_counter # import requests # from jinja2 import Environment, FileSystemLoader # from pathlib import Path # # Set up logging # logging.basicConfig(level=logging.INFO) # logger = logging.getLogger(__name__) # # API Key setup # api_key = os.getenv("GROQ_API_KEY") # if not api_key: # gr.Warning("GROQ_API_KEY not found. Set it in 'Repository secrets'.") # logger.error("GROQ_API_KEY not found.") # api_key = "" # Fallback to empty string, but this will fail without a key # else: # os.environ["GROQ_API_KEY"] = api_key # # Bhashini API setup # bhashini_api_key = os.getenv("API_KEY", "").strip() # bhashini_user_id = os.getenv("USER_ID", "").strip() # def bhashini_translate(text: str, from_code: str = "en", to_code: str = "hi") -> dict: # """Translates text from source language to target language using the Bhashini API.""" # if not text.strip(): # print('Input text is empty. Please provide valid text for translation.') # return {"status_code": 400, "message": "Input text is empty", "translated_content": None} # else: # print('Input text - ', text) # print(f'Starting translation process from {from_code} to {to_code}...') # gr.Warning(f'Translating to {to_code}...') # url = 'https://meity-auth.ulcacontrib.org/ulca/apis/v0/model/getModelsPipeline' # headers = { # "Content-Type": "application/json", # "userID": bhashini_user_id, # "ulcaApiKey": bhashini_api_key # } # for key, value in headers.items(): # if not isinstance(value, str) or '\n' in value or '\r' in value: # print(f"Invalid header value for {key}: {value}") # return {"status_code": 400, "message": f"Invalid header value for {key}", "translated_content": None} # payload = { # "pipelineTasks": [{"taskType": "translation", "config": {"language": {"sourceLanguage": from_code, "targetLanguage": to_code}}}], # "pipelineRequestConfig": {"pipelineId": "64392f96daac500b55c543cd"} # } # print('Sending initial request to get the pipeline...') # response = requests.post(url, json=payload, headers=headers) # if response.status_code != 200: # print(f'Error in initial request: {response.status_code}, Response: {response.text}') # return {"status_code": response.status_code, "message": "Error in translation request", "translated_content": None} # print('Initial request successful, processing response...') # response_data = response.json() # print('Full response data:', response_data) # if "pipelineInferenceAPIEndPoint" not in response_data or "callbackUrl" not in response_data["pipelineInferenceAPIEndPoint"]: # print('Unexpected response structure:', response_data) # return {"status_code": 400, "message": "Unexpected API response structure", "translated_content": None} # service_id = response_data["pipelineResponseConfig"][0]["config"][0]["serviceId"] # callback_url = response_data["pipelineInferenceAPIEndPoint"]["callbackUrl"] # print(f'Service ID: {service_id}, Callback URL: {callback_url}') # headers2 = { # "Content-Type": "application/json", # response_data["pipelineInferenceAPIEndPoint"]["inferenceApiKey"]["name"]: response_data["pipelineInferenceAPIEndPoint"]["inferenceApiKey"]["value"] # } # compute_payload = { # "pipelineTasks": [{"taskType": "translation", "config": {"language": {"sourceLanguage": from_code, "targetLanguage": to_code}, "serviceId": service_id}}], # "inputData": {"input": [{"source": text}], "audio": [{"audioContent": None}]} # } # print(f'Sending translation request with text: "{text}"') # compute_response = requests.post(callback_url, json=compute_payload, headers=headers2) # if compute_response.status_code != 200: # print(f'Error in translation request: {compute_response.status_code}, Response: {compute_response.text}') # return {"status_code": compute_response.status_code, "message": "Error in translation", "translated_content": None} # print('Translation request successful, processing translation...') # compute_response_data = compute_response.json() # translated_content = compute_response_data["pipelineResponse"][0]["output"][0]["target"] # print(f'Translation successful. Translated content: "{translated_content}"') # return {"status_code": 200, "message": "Translation successful", "translated_content": translated_content} # # Initialize PhiData Agent # agent = Agent( # name="Science Education Assistant", # role="You are a helpful science tutor for 10th-grade students", # instructions=[ # "You are an expert science teacher specializing in 10th-grade curriculum.", # "Provide clear, accurate, and age-appropriate explanations.", # "Use simple language and examples that students can understand.", # "Focus on concepts from physics, chemistry, and biology.", # "Structure responses with headings and bullet points when helpful.", # "Encourage learning and curiosity." # ], # model=Groq(id="llama3-70b-8192", api_key=api_key), # markdown=True # ) # # Set up Jinja2 environment # proj_dir = Path(__file__).parent # env = Environment(loader=FileSystemLoader(proj_dir / 'templates')) # template = env.get_template('template.j2') # For document context # template_html = env.get_template('template_html.j2') # For HTML output # # Response Generation Function # def retrieve_and_generate_response(query, cross_encoder_choice, history=None): # """Generate response using semantic search and LLM""" # top_rerank = 25 # top_k_rank = 20 # if not query.strip(): # return "Please provide a valid question.", [] # try: # start_time = perf_counter() # # Encode query and search documents # query_vec = retriever.encode(query) # documents = table.search(query_vec, vector_column_name="vector").limit(top_rerank).to_list() # documents = [doc["text"] for doc in documents] # # Re-rank documents using cross-encoder # cross_encoder_model = CrossEncoder('BAAI/bge-reranker-base') if cross_encoder_choice == '(ACCURATE) BGE reranker' else CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2') # query_doc_pair = [[query, doc] for doc in documents] # cross_scores = cross_encoder_model.predict(query_doc_pair) # sim_scores_argsort = list(reversed(np.argsort(cross_scores))) # documents = [documents[idx] for idx in sim_scores_argsort[:top_k_rank]] # # Create context from top documents # context = "\n\n".join(documents[:10]) if documents else "" # context = f"Context information from educational materials:\n{context}\n\n" # # Add conversation history for context # history_context = "" # if history and len(history) > 0: # for user_msg, bot_msg in history[-2:]: # Last 2 exchanges # if user_msg and bot_msg: # history_context += f"Previous Q: {user_msg}\nPrevious A: {bot_msg}\n" # # Create full prompt # full_prompt = f"{history_context}{context}Question: {query}\n\nPlease answer the question using the context provided above. If the context doesn't contain relevant information, use your general knowledge about 10th-grade science topics." # # Generate response # response = agent.run(full_prompt) # response_text = response.content if hasattr(response, 'content') else str(response) # logger.info(f"Response generation took {perf_counter() - start_time:.2f} seconds") # return response_text, documents # Return documents for template # except Exception as e: # logger.error(f"Error in response generation: {e}") # return f"Error generating response: {str(e)}", [] # def simple_chat_function(message, history, cross_encoder_choice): # """Chat function with semantic search and retriever integration""" # if not message.strip(): # return "", history, "" # # Generate response and get documents # response, documents = retrieve_and_generate_response(message, cross_encoder_choice, history) # # Add to history # history.append([message, response]) # # Render template with documents and query # prompt_html = template_html.render(documents=documents, query=message) # return "", history, prompt_html # def translate_text(selected_language, history): # """Translate the last response in history to the selected language.""" # iso_language_codes = { # "Hindi": "hi", "Gom": "gom", "Kannada": "kn", "Dogri": "doi", "Bodo": "brx", "Urdu": "ur", # "Tamil": "ta", "Kashmiri": "ks", "Assamese": "as", "Bengali": "bn", "Marathi": "mr", # "Sindhi": "sd", "Maithili": "mai", "Punjabi": "pa", "Malayalam": "ml", "Manipuri": "mni", # "Telugu": "te", "Sanskrit": "sa", "Nepali": "ne", "Santali": "sat", "Gujarati": "gu", "Odia": "or" # } # to_code = iso_language_codes[selected_language] # response_text = history[-1][1] if history and history[-1][1] else '' # print('response_text for translation', response_text) # translation = bhashini_translate(response_text, to_code=to_code) # return translation.get('translated_content', 'Translation failed.') # # Gradio Interface with layout template # with gr.Blocks(title="Science Chatbot", theme='gradio/soft') as demo: # # Header section # with gr.Row(): # with gr.Column(scale=10): # gr.HTML(value="""

Welcome! I am your friend!

Ask me !I will help you

I AM A CHATBOT FOR 10TH SCIENCE WITH TRANSLATION IN 22 LANGUAGES

""") # gr.HTML(value=f"""

A free chat bot developed by K.M.RAMYASRI,TGT,GHS.SUTHUKENY using Open source LLMs for 10 std students

""") # gr.HTML(value=f"""

Suggestions may be sent to ramyadevi1607@yahoo.com.

""") # with gr.Column(scale=3): # try: # gr.Image(value='logo.png', height=200, width=200) # except: # gr.HTML("
Logo
") # # Chat and input components # chatbot = gr.Chatbot( # [], # elem_id="chatbot", # avatar_images=('https://aui.atlassian.com/aui/8.8/docs/images/avatar-person.svg', # 'https://huggingface.co/datasets/huggingface/brand-assets/resolve/main/hf-logo.svg'), # bubble_full_width=False, # show_copy_button=True, # show_share_button=True, # ) # with gr.Row(): # msg = gr.Textbox( # scale=3, # show_label=False, # placeholder="Enter text and press enter", # container=False, # ) # submit_btn = gr.Button(value="Submit text", scale=1, variant="primary") # # Additional controls # cross_encoder = gr.Radio( # choices=['(FAST) MiniLM-L6v2', '(ACCURATE) BGE reranker'], # value='(ACCURATE) BGE reranker', # label="Embeddings Model", # info="Select the model for document ranking" # ) # language_dropdown = gr.Dropdown( # choices=[ # "Hindi", "Gom", "Kannada", "Dogri", "Bodo", "Urdu", "Tamil", "Kashmiri", "Assamese", "Bengali", "Marathi", # "Sindhi", "Maithili", "Punjabi", "Malayalam", "Manipuri", "Telugu", "Sanskrit", "Nepali", "Santali", # "Gujarati", "Odia" # ], # value="Hindi", # label="Select Language for Translation" # ) # translated_textbox = gr.Textbox(label="Translated Response") # prompt_html = gr.HTML() # Add HTML component for the template # # Event handlers # def update_chat_and_translate(message, history, cross_encoder_choice, selected_language): # if not message.strip(): # return "", history, "", "" # # Generate response and get documents # response, documents = retrieve_and_generate_response(message, cross_encoder_choice, history) # history.append([message, response]) # # Translate response # translated_text = translate_text(selected_language, history) # # Render template with documents and query # prompt_html_content = template_html.render(documents=documents, query=message) # return "", history, translated_text, prompt_html_content # msg.submit(update_chat_and_translate, [msg, chatbot, cross_encoder, language_dropdown], [msg, chatbot, translated_textbox, prompt_html]) # submit_btn.click(update_chat_and_translate, [msg, chatbot, cross_encoder, language_dropdown], [msg, chatbot, translated_textbox, prompt_html]) # clear = gr.Button("Clear Conversation") # clear.click(lambda: ([], "", "", ""), outputs=[chatbot, msg, translated_textbox, prompt_html]) # # Example questions # gr.Examples( # examples=[ # 'What is the difference between metals and non-metals?', # 'What is an ionic bond?', # 'Explain asexual reproduction', # 'What is photosynthesis?', # 'Explain Newton\'s laws of motion' # ], # inputs=msg, # label="Try these example questions:" # ) # if __name__ == "__main__": # demo.launch(server_name="0.0.0.0", server_port=7860)# import gradio as gr # # from phi.agent import Agent # # from phi.model.groq import Groq # # import os # # import logging # # from sentence_transformers import CrossEncoder # # from backend.semantic_search import table, retriever # # import numpy as np # # from time import perf_counter # # import requests # # from jinja2 import Environment, FileSystemLoader # # # Set up logging # # logging.basicConfig(level=logging.INFO) # # logger = logging.getLogger(__name__) # # # API Key setup # # api_key = os.getenv("GROQ_API_KEY") # # if not api_key: # # gr.Warning("GROQ_API_KEY not found. Set it in 'Repository secrets'.") # # logger.error("GROQ_API_KEY not found.") # # api_key = "" # Fallback to empty string, but this will fail without a key # # else: # # os.environ["GROQ_API_KEY"] = api_key # # # Bhashini API setup # # bhashini_api_key = os.getenv("API_KEY") # # bhashini_user_id = os.getenv("USER_ID") # # def bhashini_translate(text: str, from_code: str = "en", to_code: str = "hi") -> dict: # # """Translates text from source language to target language using the Bhashini API.""" # # if not text.strip(): # # print('Input text is empty. Please provide valid text for translation.') # # return {"status_code": 400, "message": "Input text is empty", "translated_content": None} # # else: # # print('Input text - ', text) # # print(f'Starting translation process from {from_code} to {to_code}...') # # gr.Warning(f'Translating to {to_code}...') # # url = 'https://meity-auth.ulcacontrib.org/ulca/apis/v0/model/getModelsPipeline' # # headers = { # # "Content-Type": "application/json", # # "userID": bhashini_user_id, # # "ulcaApiKey": bhashini_api_key # # } # # payload = { # # "pipelineTasks": [{"taskType": "translation", "config": {"language": {"sourceLanguage": from_code, "targetLanguage": to_code}}}], # # "pipelineRequestConfig": {"pipelineId": "64392f96daac500b55c543cd"} # # } # # print('Sending initial request to get the pipeline...') # # response = requests.post(url, json=payload, headers=headers) # # if response.status_code != 200: # # print(f'Error in initial request: {response.status_code}, Response: {response.text}') # # return {"status_code": response.status_code, "message": "Error in translation request", "translated_content": None} # # print('Initial request successful, processing response...') # # response_data = response.json() # # print('Full response data:', response_data) # Debug the full response # # if "pipelineInferenceAPIEndPoint" not in response_data or "callbackUrl" not in response_data["pipelineInferenceAPIEndPoint"]: # # print('Unexpected response structure:', response_data) # # return {"status_code": 400, "message": "Unexpected API response structure", "translated_content": None} # # service_id = response_data["pipelineResponseConfig"][0]["config"][0]["serviceId"] # # callback_url = response_data["pipelineInferenceAPIEndPoint"]["callbackUrl"] # # print(f'Service ID: {service_id}, Callback URL: {callback_url}') # # headers2 = { # # "Content-Type": "application/json", # # response_data["pipelineInferenceAPIEndPoint"]["inferenceApiKey"]["name"]: response_data["pipelineInferenceAPIEndPoint"]["inferenceApiKey"]["value"] # # } # # compute_payload = { # # "pipelineTasks": [{"taskType": "translation", "config": {"language": {"sourceLanguage": from_code, "targetLanguage": to_code}, "serviceId": service_id}}], # # "inputData": {"input": [{"source": text}], "audio": [{"audioContent": None}]} # # } # # print(f'Sending translation request with text: "{text}"') # # compute_response = requests.post(callback_url, json=compute_payload, headers=headers2) # # if compute_response.status_code != 200: # # print(f'Error in translation request: {compute_response.status_code}, Response: {compute_response.text}') # # return {"status_code": compute_response.status_code, "message": "Error in translation", "translated_content": None} # # print('Translation request successful, processing translation...') # # compute_response_data = compute_response.json() # # translated_content = compute_response_data["pipelineResponse"][0]["output"][0]["target"] # # print(f'Translation successful. Translated content: "{translated_content}"') # # return {"status_code": 200, "message": "Translation successful", "translated_content": translated_content} # # # Initialize PhiData Agent # # agent = Agent( # # name="Science Education Assistant", # # role="You are a helpful science tutor for 10th-grade students", # # instructions=[ # # "You are an expert science teacher specializing in 10th-grade curriculum.", # # "Provide clear, accurate, and age-appropriate explanations.", # # "Use simple language and examples that students can understand.", # # "Focus on concepts from physics, chemistry, and biology.", # # "Structure responses with headings and bullet points when helpful.", # # "Encourage learning and curiosity." # # ], # # model=Groq(id="llama3-70b-8192", api_key=api_key), # # markdown=True # # ) # # # Set up Jinja2 environment # # proj_dir = Path(__file__).parent # # env = Environment(loader=FileSystemLoader(proj_dir / 'templates')) # # template_html = env.get_template('template_html.j2') # # # Response Generation Function # # def retrieve_and_generate_response(query, cross_encoder_choice, history=None): # # """Generate response using semantic search and LLM""" # # top_rerank = 25 # # top_k_rank = 20 # # if not query.strip(): # # return "Please provide a valid question." # # try: # # start_time = perf_counter() # # # Encode query and search documents # # query_vec = retriever.encode(query) # # documents = table.search(query_vec, vector_column_name="vector").limit(top_rerank).to_list() # # documents = [doc["text"] for doc in documents] # # # Re-rank documents using cross-encoder # # cross_encoder_model = CrossEncoder('BAAI/bge-reranker-base') if cross_encoder_choice == '(ACCURATE) BGE reranker' else CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2') # # query_doc_pair = [[query, doc] for doc in documents] # # cross_scores = cross_encoder_model.predict(query_doc_pair) # # sim_scores_argsort = list(reversed(np.argsort(cross_scores))) # # documents = [documents[idx] for idx in sim_scores_argsort[:top_k_rank]] # # # Create context from top documents # # context = "\n\n".join(documents[:10]) if documents else "" # # context = f"Context information from educational materials:\n{context}\n\n" # # # Add conversation history for context # # history_context = "" # # if history and len(history) > 0: # # for user_msg, bot_msg in history[-2:]: # Last 2 exchanges # # if user_msg and bot_msg: # # history_context += f"Previous Q: {user_msg}\nPrevious A: {bot_msg}\n" # # # Create full prompt # # full_prompt = f"{history_context}{context}Question: {query}\n\nPlease answer the question using the context provided above. If the context doesn't contain relevant information, use your general knowledge about 10th-grade science topics." # # # Generate response # # response = agent.run(full_prompt) # # response_text = response.content if hasattr(response, 'content') else str(response) # # logger.info(f"Response generation took {perf_counter() - start_time:.2f} seconds") # # return response_text # # except Exception as e: # # logger.error(f"Error in response generation: {e}") # # return f"Error generating response: {str(e)}" # # def simple_chat_function(message, history, cross_encoder_choice): # # """Chat function with semantic search and retriever integration""" # # if not message.strip(): # # return "", history # # # Generate response using the semantic search function # # response = retrieve_and_generate_response(message, cross_encoder_choice, history) # # # Add to history # # history.append([message, response]) # # return "", history # # def translate_text(selected_language, history): # # """Translate the last response in history to the selected language.""" # # iso_language_codes = { # # "Hindi": "hi", "Gom": "gom", "Kannada": "kn", "Dogri": "doi", "Bodo": "brx", "Urdu": "ur", # # "Tamil": "ta", "Kashmiri": "ks", "Assamese": "as", "Bengali": "bn", "Marathi": "mr", # # "Sindhi": "sd", "Maithili": "mai", "Punjabi": "pa", "Malayalam": "ml", "Manipuri": "mni", # # "Telugu": "te", "Sanskrit": "sa", "Nepali": "ne", "Santali": "sat", "Gujarati": "gu", "Odia": "or" # # } # # to_code = iso_language_codes[selected_language] # # response_text = history[-1][1] if history and history[-1][1] else '' # # print('response_text for translation', response_text) # # translation = bhashini_translate(response_text, to_code=to_code) # # return translation.get('translated_content', 'Translation failed.') # # # Gradio Interface with layout template # # with gr.Blocks(title="Science Chatbot", theme='gradio/soft') as demo: # # # Header section # # with gr.Row(): # # with gr.Column(scale=10): # # gr.HTML(value="""

Welcome! I am your friend!

Ask me !I will help you

I AM A CHATBOT FOR 10TH SCIENCE WITH TRANSLATION IN 22 LANGUAGES

""") # # gr.HTML(value=f"""

A free chat bot developed by K.M.RAMYASRI,TGT,GHS.SUTHUKENY using Open source LLMs for 10 std students

""") # # gr.HTML(value=f"""

Suggestions may be sent to ramyadevi1607@yahoo.com.

""") # # with gr.Column(scale=3): # # try: # # gr.Image(value='logo.png', height=200, width=200) # # except: # # gr.HTML("
Logo
") # # # Chat and input components # # chatbot = gr.Chatbot( # # [], # # elem_id="chatbot", # # avatar_images=('https://aui.atlassian.com/aui/8.8/docs/images/avatar-person.svg', # # 'https://huggingface.co/datasets/huggingface/brand-assets/resolve/main/hf-logo.svg'), # # bubble_full_width=False, # # show_copy_button=True, # # show_share_button=True, # # ) # # with gr.Row(): # # msg = gr.Textbox( # # scale=3, # # show_label=False, # # placeholder="Enter text and press enter", # # container=False, # # ) # # submit_btn = gr.Button(value="Submit text", scale=1, variant="primary") # # # Additional controls # # cross_encoder = gr.Radio( # # choices=['(FAST) MiniLM-L6v2', '(ACCURATE) BGE reranker'], # # value='(ACCURATE) BGE reranker', # # label="Embeddings Model", # # info="Select the model for document ranking" # # ) # # language_dropdown = gr.Dropdown( # # choices=[ # # "Hindi", "Gom", "Kannada", "Dogri", "Bodo", "Urdu", "Tamil", "Kashmiri", "Assamese", "Bengali", "Marathi", # # "Sindhi", "Maithili", "Punjabi", "Malayalam", "Manipuri", "Telugu", "Sanskrit", "Nepali", "Santali", # # "Gujarati", "Odia" # # ], # # value="Hindi", # # label="Select Language for Translation" # # ) # # translated_textbox = gr.Textbox(label="Translated Response") # # # Event handlers # # def update_chat_and_translate(message, history, cross_encoder_choice, selected_language): # # if not message.strip(): # # return "", history, "" # # # Generate response # # response = retrieve_and_generate_response(message, cross_encoder_choice, history) # # history.append([message, response]) # # # Translate response # # translated_text = translate_text(selected_language, history) # # return "", history, translated_text # # msg.submit(update_chat_and_translate, [msg, chatbot, cross_encoder, language_dropdown], [msg, chatbot, translated_textbox]) # # submit_btn.click(update_chat_and_translate, [msg, chatbot, cross_encoder, language_dropdown], [msg, chatbot, translated_textbox]) # # clear = gr.Button("Clear Conversation") # # clear.click(lambda: ([], "", ""), outputs=[chatbot, msg, translated_textbox]) # # # Example questions # # gr.Examples( # # examples=[ # # 'What is the difference between metals and non-metals?', # # 'What is an ionic bond?', # # 'Explain asexual reproduction', # # 'What is photosynthesis?', # # 'Explain Newton\'s laws of motion' # # ], # # inputs=msg, # # label="Try these example questions:" # # ) # # if __name__ == "__main__": # # demo.launch(server_name="0.0.0.0", server_port=7860)# import gradio as gr