Spaces:
Build error
Build error
| import streamlit as st | |
| from datetime import datetime | |
| import anthropic | |
| import os | |
| import tempfile | |
| class AITutorService: | |
| def __init__(self): | |
| self._client = None | |
| self.initialize_client() | |
| self.voice_enabled = False | |
| self.tts_mode = None | |
| self.openai_enabled = False | |
| def client(self): | |
| """Property to safely access the Anthropic client""" | |
| if self._client is None: | |
| raise ValueError("Anthropic client not properly initialized") | |
| return self._client | |
| def initialize_client(self): | |
| """Initialize Anthropic client with better error handling""" | |
| try: | |
| api_key = st.secrets.get("ANTHROPIC_API_KEY") or os.getenv('ANTHROPIC_API_KEY') | |
| if not api_key: | |
| st.error("ANTHROPIC_API_KEY not found in secrets or environment variables") | |
| return | |
| self._client = anthropic.Anthropic(api_key=api_key) | |
| self.voice_enabled = True | |
| self.tts_mode = 'openai' | |
| self.openai_enabled = True | |
| except Exception as e: | |
| st.error(f"Failed to initialize Anthropic client: {str(e)}") | |
| self._client = None | |
| self.voice_enabled = False | |
| self.tts_mode = None | |
| self.openai_enabled = False | |
| def generate_response(self, user_input: str, current_topic: str = None) -> str: | |
| """Generate response using Claude with better error handling""" | |
| if not self.client: | |
| return "I apologize, but I'm not properly connected right now. Please check the API configuration." | |
| try: | |
| # Build system prompt | |
| system_prompt = """You are an expert tutor helping students learn. | |
| Give clear, step-by-step explanations with examples.""" | |
| if current_topic and current_topic != "All Topics": | |
| system_prompt += f"\nYou are currently tutoring about {current_topic}." | |
| # Generate response | |
| response = self.client.messages.create( | |
| model="claude-3-opus-20240229", | |
| temperature=0.7, | |
| max_tokens=1024, | |
| system=system_prompt, | |
| messages=[{ | |
| "role": "user", | |
| "content": user_input | |
| }] | |
| ) | |
| # Get response text | |
| response_text = response.content[0].text | |
| # Update metrics | |
| self.update_engagement_metrics(user_input, response_text) | |
| return response_text | |
| except anthropic.APIError as e: | |
| st.error(f"API Error: {str(e)}") | |
| return "I encountered an API error. Please try again in a moment." | |
| except anthropic.APIConnectionError as e: | |
| st.error(f"Connection Error: {str(e)}") | |
| return "I'm having trouble connecting to my API. Please check your internet connection." | |
| except Exception as e: | |
| st.error(f"Unexpected error: {str(e)}") | |
| return "I encountered an unexpected error. Please try again." | |
| def speak(self, text: str): | |
| """Make the AI tutor speak using available TTS method""" | |
| if not self.voice_enabled: | |
| return | |
| if self.tts_mode == 'openai': | |
| if self.openai_enabled: | |
| self.speak_with_openai(text) | |
| else: | |
| st.warning("OpenAI TTS is not available. Voice output is disabled.") | |
| self.tts_mode = None | |
| self.voice_enabled = False | |
| else: | |
| st.warning("Local TTS is not implemented yet.") | |
| def speak_with_openai(self, text: str): | |
| """Generate speech using OpenAI's TTS""" | |
| try: | |
| # Generate speech using OpenAI | |
| response = self.client.audio.speech.create( | |
| model="tts-1", | |
| voice="alloy", # or "nova", "shimmer", "echo", "fable" | |
| input=text | |
| ) | |
| # Create a temporary file to store the audio | |
| with tempfile.NamedTemporaryFile(delete=False, suffix='.mp3') as tmp_file: | |
| response.stream_to_file(tmp_file.name) | |
| # Create audio player in Streamlit | |
| audio_file = open(tmp_file.name, 'rb') | |
| audio_bytes = audio_file.read() | |
| st.audio(audio_bytes, format='audio/mp3') | |
| # Clean up | |
| audio_file.close() | |
| os.unlink(tmp_file.name) | |
| except Exception as e: | |
| st.error(f"Error during OpenAI speech synthesis: {e}") | |
| def update_engagement_metrics(self, user_input: str, response: str): | |
| """Update engagement metrics""" | |
| if 'tutor_context' not in st.session_state: | |
| st.session_state.tutor_context = { | |
| 'engagement_metrics': [] | |
| } | |
| metrics = st.session_state.tutor_context['engagement_metrics'] | |
| metrics.append({ | |
| 'timestamp': datetime.now().isoformat(), | |
| 'sentiment_score': min(1.0, len(user_input) / 100.0), | |
| 'interaction_length': len(user_input) | |
| }) |