""" Personal AI Coach with CrewAI and Mistral Multilingual support with advanced conversational AI """ import gradio as gr import asyncio import os import sys from datetime import datetime from typing import Dict, List, Tuple, Optional import numpy as np sys.path.append(os.path.dirname(os.path.abspath(__file__))) from crew_config import PersonalCoachCrew from agents.tools.voice_tools import MultilingualVoiceProcessor from utils.config import Config from dotenv import load_dotenv import certifi load_dotenv() class PersonalCoachApp: """Main application using CrewAI orchestration""" def __init__(self): print("Initializing Personal Coach AI with CrewAI...") self.config = Config() # Initialize CrewAI self.crew = PersonalCoachCrew(self.config) # Initialize voice processor self.voice_processor = MultilingualVoiceProcessor() # Session management self.conversation_history = [] self.session_data = { "start_time": datetime.now(), "language": "en", "user_profile": {} } print("Personal Coach AI initialized successfully!") async def process_input( self, text_input: str, voice_input: Optional[np.ndarray], language: str, history: List ) -> Tuple: """Process user input through CrewAI""" try: # Prepare input data if voice_input is not None: # Process voice input with language detection transcribed_text, detected_lang = await self.voice_processor.transcribe( voice_input, language ) text_input = transcribed_text self.session_data["language"] = detected_lang else: self.session_data["language"] = language if not text_input: return history, None, "", None # Prepare crew input crew_input = { "user_message": text_input, "language": self.session_data["language"], "conversation_history": history[-5:], # Last 5 exchanges "user_profile": self.session_data.get("user_profile", {}) } # Execute crew print(f"Processing input in {self.session_data['language']}...") print("PersonalCoachCrew methods:", dir(PersonalCoachCrew)) print("Has process?", hasattr(PersonalCoachCrew, "process")) result = self.crew.process(inputs=crew_input) # Extract response response_text = result.get("final_response", "I'm here to help. Please tell me more.") # Generate audio response audio_response = await self.voice_processor.synthesize( response_text, self.session_data["language"], voice_type="meditation" ) # Update history history.append([text_input, response_text]) # Update user profile if "user_profile_update" in result: self.session_data["user_profile"].update(result["user_profile_update"]) return history, audio_response, "", None except Exception as e: print(f"Error in process_input: {str(e)}") error_message = f"I apologize, but I encountered an error: {str(e)}" history.append(["Error", error_message]) return history, None, "", None def clear_conversation(self): """Clear conversation and reset session""" self.conversation_history = [] self.session_data["user_profile"] = {} return [], None def create_interface(): """Create Gradio interface with multilingual support""" app = PersonalCoachApp() with gr.Blocks(theme=gr.themes.Soft(), title="Personal AI Coach") as interface: gr.Markdown(""" # 🧘 Personal AI Coach - Multilingual CrewAI System Powered by Mistral AI and CrewAI's multi-agent framework. Supports multiple languages! **Features:** - 🌍 Multilingual voice and text support - 🤖 4 specialized AI agents working together - 🧠 Advanced Mistral AI for deep understanding - 📚 Wisdom from 13 spiritual and self-help texts - 🎙️ Natural voice interactions in your language """) with gr.Row(): # Main chat interface with gr.Column(scale=3): chatbot = gr.Chatbot( height=500, bubble_full_width=False #avatar_images=(None, "🧘") ) with gr.Row(): language = gr.Dropdown( choices=[ ("English", "en"), ("Spanish", "es"), ("French", "fr"), ("German", "de"), ("Italian", "it"), ("Portuguese", "pt"), ("Hindi", "hi"), ("Chinese", "zh"), ("Japanese", "ja"), ("Korean", "ko"), ("Arabic", "ar"), ("Russian", "ru") ], value="en", label="Language", scale=1 ) text_input = gr.Textbox( placeholder="Type your message or click the microphone...", show_label=False, scale=3 ) voice_input = gr.Audio( type="numpy", label="Speak here", scale=1 ) with gr.Row(): send_btn = gr.Button("Send 📤", variant="primary") clear_btn = gr.Button("Clear 🗑️") audio_output = gr.Audio( label="🔊 Coach Response", autoplay=True ) # Sidebar with gr.Column(scale=1): gr.Markdown(""" ### 🤖 CrewAI Agent Team **Agent 1: Empathetic Listener** - Multilingual voice processing - Emotional understanding - Context analysis **Agent 2: Wisdom Keeper** - RAG with Mistral AI - Spiritual text knowledge - Personalized guidance **Agent 3: Guardian** - Response validation - Safety checks - Tone refinement **Agent 4: Conversation Guide** - Natural dialogue flow - Voice synthesis - Feedback integration ### 🌍 Supported Languages Voice input/output in 12+ languages ### 📚 Knowledge Sources - Bhagavad Gita - Power of Now - Atomic Habits - Meditations - And 9 more texts... """) # Examples in multiple languages with gr.Accordion("💡 Example Prompts", open=False): gr.Examples( examples=[ ["I'm feeling overwhelmed with work pressure", "en"], ["Je me sens perdu dans ma vie", "fr"], ["Estoy luchando con la ansiedad", "es"], ["Ich möchte bessere Gewohnheiten aufbauen", "de"], ["मुझे अपने जीवन का उद्देश्य नहीं मिल रहा", "hi"], ["我想学习冥想", "zh"] ], inputs=[text_input, language] ) # # Event handlers async def handle_submission(text, voice, lang, history): return await app.process_input(text, voice, lang, history) # # Connect events for trigger in [text_input.submit, send_btn.click]: trigger( fn=lambda *args: asyncio.run(handle_submission(*args)), inputs=[text_input, voice_input, language, chatbot], outputs=[chatbot, audio_output, text_input, voice_input] ) # for trigger in [text_input.submit, send_btn.click]: # trigger( # fn=app.process_input, # <-- Use async method directly! # inputs=[text_input, voice_input, language, chatbot], # outputs=[chatbot, audio_output, text_input, voice_input] # ) clear_btn.click( fn=app.clear_conversation, outputs=[chatbot, audio_output] ) return interface if __name__ == "__main__": print("Starting Personal Coach AI with CrewAI...") interface = create_interface() interface.launch() #app = create_interface()