import os import json import base64 import httpx from typing import Dict, List, Any, Optional from groq import AsyncGroq from config.prompts import ( SCENARIO_GENERATION_PROMPT, CHARACTER_ROLEPLAY_PROMPT, CHARACTER_CONCLUSION_PROMPT, CHARACTER_GUIDED_ROLEPLAY_PROMPT, SKILL_ANALYSIS_PROMPT, get_skill_analysis_template ) class GroqService: def __init__(self): from dotenv import load_dotenv load_dotenv() api_key = os.getenv("GROQ_API_KEY") if not api_key: print("⚠️ WARNING: GROQ_API_KEY not found in environment variables") print(" ℹ️ Groq service will not be available. System will use Ollama only.") print(" ℹ️ To use Groq, set GROQ_API_KEY in your .env file or environment") self.client = None self.available = False else: self.client = AsyncGroq(api_key=api_key) self.available = True self.model = "openai/gpt-oss-20b" # You can change this to mixtral-8x7b-32768 or llama3-8b-8192 # Token counters for different operations self.token_counts = { 'preview': {'input': 0, 'output': 0, 'total': 0}, 'conversation': {'input': 0, 'output': 0, 'total': 0}, 'assessment': {'input': 0, 'output': 0, 'total': 0} } def _update_token_count(self, operation_type: str, usage_stats): """Update token counts for specific operation""" if usage_stats and hasattr(usage_stats, 'prompt_tokens') and hasattr(usage_stats, 'completion_tokens'): input_tokens = usage_stats.prompt_tokens output_tokens = usage_stats.completion_tokens total_tokens = usage_stats.total_tokens self.token_counts[operation_type]['input'] += input_tokens self.token_counts[operation_type]['output'] += output_tokens self.token_counts[operation_type]['total'] += total_tokens print(f"GROQ Token count for {operation_type}: Input={input_tokens}, Output={output_tokens}, Total={total_tokens}") def get_token_counts(self) -> Dict[str, Any]: """Get current token counts for all operations""" return self.token_counts.copy() def reset_token_counts(self): """Reset all token counters""" for operation in self.token_counts: self.token_counts[operation] = {'input': 0, 'output': 0, 'total': 0} async def generate_scenario(self, admin_input: Dict[str, Any]) -> Optional[Dict[str, Any]]: """Generate structured scenario from admin input""" try: # Format details dictionary into a readable string details_text = "" for key, value in admin_input['details'].items(): details_text += f"{key}: {value}\n" # Format roleplay questions if present questions_text = "" roleplay_questions = admin_input.get('roleplay_questions') if roleplay_questions: questions_text = "\nGUIDED CONVERSATION QUESTIONS (The scenario should accommodate THESE topics):\n" for idx, q in enumerate(roleplay_questions, 1): questions_text += f"{idx}. {q.get('question', '')}\n" prompt = SCENARIO_GENERATION_PROMPT.format( category=admin_input['category'], objective=admin_input['objective'], learner_role=admin_input['learner_role'], ai_role=admin_input['ai_role'], details=details_text + questions_text, skills=', '.join(admin_input['skills_to_assess']) ) response = await self.client.chat.completions.create( messages=[ { "role": "system", "content": "You are a training scenario expert. Always respond with valid JSON only." }, { "role": "user", "content": prompt } ], model=self.model, temperature=0.7, max_tokens=2048 ) # Track tokens for preview operation self._update_token_count('preview', response.usage) response_text = response.choices[0].message.content.strip() # Try to extract JSON from response try: # Remove any markdown formatting if response_text.startswith('```json'): response_text = response_text[7:] if response_text.endswith('```'): response_text = response_text[:-3] return json.loads(response_text) except json.JSONDecodeError as e: print(f"JSON parsing error: {e}") print(f"Response text: {response_text}") return None except Exception as e: print(f"Error generating scenario: {e}") return None def _analyze_question_progression(self, roleplay_questions: List[Dict[str, str]], conversation_history: List[Dict[str, str]]) -> str: """ Analyze conversation history to determine which guided questions have been addressed. Returns a formatted string explaining the progression status. """ if not roleplay_questions: return "No guided questions for this roleplay." # Combine all conversation text for analysis conversation_text = " ".join([turn.get('message', '') for turn in conversation_history]).lower() progression_lines = [] for idx, q in enumerate(roleplay_questions, 1): question_text = q.get('question', '').lower() # Simple keyword matching - check if question topic appears in conversation keywords = question_text.split()[:3] # Use first 3 words as keywords addressed = any(keyword in conversation_text for keyword in keywords if len(keyword) > 3) status = "✓ ADDRESSED" if addressed else "⚠ NOT YET ADDRESSED" progression_lines.append(f"Question {idx}: {q.get('question', '')} - {status}") # Determine next question to focus on for idx, q in enumerate(roleplay_questions): question_text = q.get('question', '').lower() keywords = question_text.split()[:3] addressed = any(keyword in conversation_text for keyword in keywords if len(keyword) > 3) if not addressed: progression_lines.append(f"\n**NEXT FOCUS**: Question {idx + 1} - {q.get('question', '')}") break return "\n".join(progression_lines) async def play_character(self, scenario: Dict[str, Any], conversation_history: List[Dict[str, str]], user_message: str, is_conclusion: bool = False, roleplay_questions: List[Dict[str, str]] = None) -> Optional[str]: """AI character responses during roleplay""" try: # Format conversation history history_text = "" for turn in conversation_history: speaker = "Learner" if turn['speaker'] == 'learner' else scenario['ai_character']['name'] history_text += f"{speaker}: {turn['message']}\n" # Calculate current turn number current_turn = len([t for t in conversation_history if t.get('speaker') == 'learner']) + 1 # Extract roleplay metadata from scenario roleplay_name = scenario.get('roleplay_name', 'Roleplay Session') character_role = scenario.get('ai_role', 'Character') learner_role = scenario.get('learner_role', 'Learner') difficulty = scenario.get('difficulty', 'medium') duration = scenario.get('duration', 300) if is_conclusion: prompt_template = CHARACTER_CONCLUSION_PROMPT prompt_kwargs = { "character_name": scenario['ai_character']['name'], "personality": scenario['ai_character']['personality'], "goals": scenario['ai_character']['goals'], "background": scenario['ai_character']['background'], "emotional_state": scenario['ai_character'].get('emotional_state', 'neutral'), "context": scenario['scenario_setup']['context'], "environment": scenario['scenario_setup']['environment'], "constraints": scenario['scenario_setup']['constraints'], "conversation_history": history_text, "user_message": user_message } elif roleplay_questions: prompt_template = CHARACTER_GUIDED_ROLEPLAY_PROMPT # Format questions for the prompt questions_text = "" for idx, q in enumerate(roleplay_questions, 1): questions_text += f"{idx}. {q.get('question', '')}\n" # Analyze question progression question_progress = self._analyze_question_progression(roleplay_questions, conversation_history) prompt_kwargs = { "character_name": scenario['ai_character']['name'], "personality": scenario['ai_character']['personality'], "goals": scenario['ai_character']['goals'], "background": scenario['ai_character']['background'], "emotional_state": scenario['ai_character'].get('emotional_state', 'neutral'), "context": scenario['scenario_setup']['context'], "environment": scenario['scenario_setup']['environment'], "objective": scenario.get('objective', ''), "additional_info": scenario.get('details', {}).get('background', ''), "constraints": scenario['scenario_setup']['constraints'], "guided_questions": questions_text, "question_progress_analysis": question_progress, "conversation_history": history_text, "user_message": user_message, "roleplay_name": roleplay_name, "character_role": character_role, "learner_role": learner_role, "difficulty": difficulty, "duration": duration, "current_turn": current_turn } else: prompt_template = CHARACTER_ROLEPLAY_PROMPT prompt_kwargs = { "character_name": scenario['ai_character']['name'], "personality": scenario['ai_character']['personality'], "goals": scenario['ai_character']['goals'], "background": scenario['ai_character']['background'], "emotional_state": scenario['ai_character'].get('emotional_state', 'neutral'), "context": scenario['scenario_setup']['context'], "environment": scenario['scenario_setup']['environment'], "objective": scenario.get('objective', ''), "additional_info": scenario.get('details', {}).get('background', ''), "constraints": scenario['scenario_setup']['constraints'], "conversation_history": history_text, "user_message": user_message, "roleplay_name": roleplay_name, "character_role": character_role, "learner_role": learner_role, "difficulty": difficulty, "duration": duration, "current_turn": current_turn } prompt = prompt_template.format(**prompt_kwargs) response = await self.client.chat.completions.create( messages=[ { "role": "system", "content": "You are a world-class roleplay AI. You MUST stay strictly in character as a human-like counterpart (client, customer, etc.). You NEVER break character. You MUST follow the provided 'MANDATORY GUIDED QUESTIONS' if they exist. Respond professionally, naturally, and concisely (1-3 sentences)." }, { "role": "user", "content": prompt } ], model=self.model, temperature=0.8, max_tokens=512 ) # Track tokens for conversation operation self._update_token_count('conversation', response.usage) return response.choices[0].message.content.strip() except Exception as e: print(f"Error in character roleplay: {e}") return "I'm having trouble responding right now. Please try again." async def analyze_skills(self, scenario: Dict[str, Any], conversation_turns: List[Dict[str, str]], engagement_metadata: str = "") -> Optional[Dict[str, Any]]: """Comprehensive skill analysis with engagement-aware scoring""" try: # Format conversation for analysis conversation_text = "" for turn in conversation_turns: speaker = "Learner" if turn['speaker'] == 'learner' else "AI Character" conversation_text += f"{speaker}: {turn['message']}\n" # Build scenario context scenario_context = f""" Category: {scenario['category']} Objective: {scenario['objective']} Context: {scenario['scenario_setup']['context']} Success Criteria: {scenario['success_criteria']} AI Character: {scenario['ai_character']['name']} - {scenario['ai_character']['background']} """ skills = scenario['skills_to_assess'] skill_template = get_skill_analysis_template(skills) prompt = SKILL_ANALYSIS_PROMPT.format( skills=', '.join(skills), scenario_context=scenario_context, engagement_metadata=engagement_metadata, conversation=conversation_text, skill_analysis_template=skill_template ) response = await self.client.chat.completions.create( messages=[ { "role": "system", "content": "You are an expert skill assessor. Provide STRICT, ACCURATE analysis in valid JSON format only. Do NOT inflate scores. Follow the engagement-based score ceilings exactly." }, { "role": "user", "content": prompt } ], model=self.model, temperature=0.3, # Lower temperature for consistent analysis max_tokens=3000 ) # Track tokens for assessment operation self._update_token_count('assessment', response.usage) response_text = response.choices[0].message.content.strip() # Try to extract JSON from response try: # Remove any markdown formatting if response_text.startswith('```json'): response_text = response_text[7:] elif response_text.startswith('```'): response_text = response_text[3:] if response_text.endswith('```'): response_text = response_text[:-3] response_text = response_text.strip() analysis = json.loads(response_text) # Validate the analysis has required structure if not all(key in analysis for key in ['skill_analysis', 'overall_performance', 'conversation_analysis', 'recommendations']): print("Analysis missing required keys") return None return analysis except json.JSONDecodeError as e: print(f"JSON parsing error in analysis: {e}") print(f"Response text: {response_text}") # Try to extract JSON from within the text using regex import re json_match = re.search(r'\{.*\}', response_text, re.DOTALL) if json_match: try: analysis = json.loads(json_match.group()) print("Successfully extracted JSON using regex fallback") return analysis except json.JSONDecodeError: print("Regex fallback also failed") return None except Exception as e: print(f"Error analyzing skills: {e}") return None async def get_completion(self, prompt: str) -> Optional[str]: """Get a simple text completion from the LLM""" try: response = await self.client.chat.completions.create( messages=[ { "role": "system", "content": "You are a helpful assistant. Provide clear, natural responses without any formatting symbols." }, { "role": "user", "content": prompt } ], model=self.model, temperature=0.7, max_tokens=1500 ) # Track tokens for preview operation (get_completion is used for scenario formatting) self._update_token_count('preview', response.usage) return response.choices[0].message.content.strip() except Exception as e: print(f"Error getting completion: {e}") return None async def generate_tts(self, text: str) -> Optional[str]: """ Generate human-like voice using Groq's TTS model. Uses direct HTTP request to avoid SDK version conflicts. Returns base64 encoded audio string. """ if not self.available: return None # Clean text slightly to ensure no non-TTS characters interfere if not text or len(text.strip()) == 0: return None # Limit text length as per Groq limits (usually 4096 chars) text = text[:4000] try: api_key = os.getenv("GROQ_API_KEY") url = "https://api.groq.com/openai/v1/audio/speech" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } payload = { "model": "canopylabs/orpheus-v1-english", "voice": "tara", "input": text, "response_format": "mp3" } print(f"INFO: Requesting TTS from Groq for text: '{text[:30]}...'") async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post(url, headers=headers, json=payload) if response.status_code == 200: audio_data = response.content if audio_data: b64_audio = base64.b64encode(audio_data).decode('utf-8') print(f"SUCCESS: Generated TTS base64 (length: {len(b64_audio)})") return b64_audio else: print(f"ERROR: Groq TTS API returned {response.status_code}: {response.text}") return None except Exception as e: print(f"EXCEPTION in generate_tts: {str(e)}") return None