import json import requests from typing import Dict, List, Any, Optional from config.prompts import ( SCENARIO_GENERATION_PROMPT, CHARACTER_ROLEPLAY_PROMPT, SKILL_ANALYSIS_PROMPT, get_skill_analysis_template ) class OllamaService: def __init__(self, base_url: str = "http://127.0.0.1:11434"): self.base_url = base_url self.model = "gemma3:12b" # Using available gemma3:12b model # 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 _estimate_tokens(self, text: str) -> int: """Estimate token count for text (rough approximation: 1 token ≈ 4 characters)""" return max(1, len(text) // 4) def _update_token_count(self, operation_type: str, usage_stats: Dict[str, Any]): """Update token counts for specific operation using actual Ollama stats""" if usage_stats: input_tokens = usage_stats.get('prompt_eval_count', 0) output_tokens = usage_stats.get('eval_count', 0) total_tokens = usage_stats.get('total_duration', input_tokens + output_tokens) # If total_duration is time-based, use sum of input + output if total_tokens > 100000: # Likely a duration in nanoseconds, not token count total_tokens = input_tokens + output_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"OLLAMA Token count for {operation_type}: Input={input_tokens}, Output={output_tokens}, Total={total_tokens}") else: print(f"OLLAMA: No usage stats available for {operation_type}, skipping token count") 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} def _make_request(self, prompt: str, system_prompt: str = "", temperature: float = 0.7, max_tokens: int = 2048) -> tuple[Optional[str], Optional[Dict[str, Any]]]: """Make a request to Ollama API and return response text and usage stats""" try: url = f"{self.base_url}/api/generate" print(f"DEBUG: Making Ollama request to: {url}") print(f"DEBUG: Using model: {self.model}") # Combine system and user prompts for Ollama full_prompt = prompt if system_prompt: full_prompt = f"System: {system_prompt}\n\nUser: {prompt}" payload = { "model": self.model, "prompt": full_prompt, "stream": False, "options": { "temperature": temperature, "num_predict": max_tokens } } print(f"DEBUG: Payload model: {payload['model']}") response = requests.post(url, json=payload, timeout=120) print(f"DEBUG: Response status: {response.status_code}") if response.status_code == 404: print(f"ERROR: 404 - Model '{self.model}' not found. Available models can be checked with: ollama list") return None, None response.raise_for_status() result = response.json() response_text = result.get('response', '').strip() usage_stats = { 'prompt_eval_count': result.get('prompt_eval_count', 0), 'eval_count': result.get('eval_count', 0), 'total_duration': result.get('total_duration', 0), 'prompt_eval_duration': result.get('prompt_eval_duration', 0), 'eval_duration': result.get('eval_duration', 0) } print(f"DEBUG: Ollama response length: {len(response_text)}") print(f"DEBUG: Ollama usage stats: {usage_stats}") return response_text, usage_stats except requests.exceptions.ConnectionError as e: print(f"Error connecting to Ollama server: {e}") print(f"Make sure Ollama is running on {self.base_url}") return None, None except requests.exceptions.RequestException as e: print(f"Error making Ollama request: {e}") if hasattr(e, 'response') and e.response is not None: print(f"Response content: {e.response.text}") return None, None except Exception as e: print(f"Unexpected error in Ollama request: {e}") return None, None def generate_scenario(self, admin_input: Dict[str, Any]) -> Optional[Dict[str, Any]]: """Generate structured scenario from admin input""" try: prompt = SCENARIO_GENERATION_PROMPT.format( category=admin_input['category'], objective=admin_input['objective'], details=admin_input['details'], skills=', '.join(admin_input['skills_to_assess']) ) system_prompt = "You are a training scenario expert. Always respond with valid JSON only." response_text, usage_stats = self._make_request( prompt=prompt, system_prompt=system_prompt, temperature=0.7, max_tokens=2048 ) if not response_text: return None # Track tokens for preview operation using actual Ollama stats self._update_token_count('preview', usage_stats) # 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] return json.loads(response_text.strip()) 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 with Ollama: {e}") return None def play_character(self, scenario: Dict[str, Any], conversation_history: List[Dict[str, str]], user_message: str) -> 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" prompt = CHARACTER_ROLEPLAY_PROMPT.format( 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 ) system_prompt = "You are a roleplay character. Stay in character at all times. Respond naturally and realistically." response, usage_stats = self._make_request( prompt=prompt, system_prompt=system_prompt, temperature=0.8, max_tokens=512 ) if response: # Track tokens for conversation operation using actual Ollama stats self._update_token_count('conversation', usage_stats) return response if response else "I'm having trouble responding right now. Please try again." except Exception as e: print(f"Error in character roleplay with Ollama: {e}") return "I'm having trouble responding right now. Please try again." def analyze_skills(self, scenario: Dict[str, Any], conversation_turns: List[Dict[str, str]]) -> Optional[Dict[str, Any]]: """Comprehensive skill analysis""" 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, conversation=conversation_text, skill_analysis_template=skill_template ) system_prompt = "You are an expert skill assessor. Provide detailed, accurate analysis in valid JSON format only." response_text, usage_stats = self._make_request( prompt=prompt, system_prompt=system_prompt, temperature=0.3, # Lower temperature for consistent analysis max_tokens=3000 ) if not response_text: return None # Track tokens for assessment operation using actual Ollama stats self._update_token_count('assessment', usage_stats) # 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() # Sanitize the JSON text before parsing response_text = self._sanitize_json_text(response_text) 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 multiple fallback methods return self._try_json_fallbacks(response_text) except Exception as e: print(f"Error analyzing skills with Ollama: {e}") return None def get_completion(self, prompt: str) -> Optional[str]: """Get a simple text completion from Ollama""" try: system_prompt = "You are a helpful assistant. Provide clear, natural responses without any formatting symbols." response, usage_stats = self._make_request( prompt=prompt, system_prompt=system_prompt, temperature=0.7, max_tokens=1500 ) if response: # Track tokens for preview operation using actual Ollama stats (get_completion is used for scenario formatting) self._update_token_count('preview', usage_stats) return response except Exception as e: print(f"Error getting completion from Ollama: {e}") return None def _sanitize_json_text(self, text: str) -> str: """Sanitize JSON text to fix common issues before parsing""" try: # Fix common issues that cause JSON parsing to fail sanitized = text # Fix malformed contractions and apostrophes import re # Fix "doesn's" -> "doesn't", "can's" -> "can't", etc. sanitized = re.sub(r"(\w+)n's\b", r"\1n't", sanitized) sanitized = re.sub(r"(\w+)'s\b", r"\1's", sanitized) # Fix possessives # Fix common Unicode issues sanitized = sanitized.replace('"', '"').replace('"', '"') # Smart quotes to regular quotes sanitized = sanitized.replace(''', "'").replace(''', "'") # Smart apostrophes # Fix trailing commas in JSON objects/arrays sanitized = re.sub(r',(\s*[}\]])', r'\1', sanitized) # Fix missing commas between objects/arrays sanitized = re.sub(r'}\s*{', r'},{', sanitized) sanitized = re.sub(r']\s*\[', r'],[', sanitized) # Fix unescaped quotes inside strings (basic attempt) # This is a simple heuristic and may not catch all cases lines = sanitized.split('\n') for i, line in enumerate(lines): # Look for lines that seem to have unescaped quotes in string values if ':' in line and line.strip().endswith('"'): # Count quotes in the value part (after the colon) key_val = line.split(':', 1) if len(key_val) == 2: value_part = key_val[1].strip() # If there are odd number of quotes, there might be an issue if value_part.count('"') % 2 != 0: # Simple fix: escape quotes that aren't at string boundaries value_part = re.sub(r'(? Optional[Dict[str, Any]]: """Try multiple fallback methods to extract valid JSON""" import re # Fallback 1: Regex extraction with sanitization try: json_match = re.search(r'\{.*\}', text, re.DOTALL) if json_match: extracted_json = json_match.group() sanitized_json = self._sanitize_json_text(extracted_json) analysis = json.loads(sanitized_json) print("Successfully extracted JSON using regex fallback with sanitization") return analysis except Exception as e: print(f"Regex fallback with sanitization failed: {e}") # Fallback 2: Try to fix and parse line by line try: # Build a basic valid JSON structure print("Attempting to reconstruct JSON structure...") # Look for key sections in the response skill_analysis_match = re.search(r'"skill_analysis":\s*\{.*?\}(?=,\s*")', text, re.DOTALL) overall_performance_match = re.search(r'"overall_performance":\s*\{.*?\}(?=,\s*")', text, re.DOTALL) conversation_analysis_match = re.search(r'"conversation_analysis":\s*\{.*?\}(?=,\s*")', text, re.DOTALL) recommendations_match = re.search(r'"recommendations":\s*\{.*?\}', text, re.DOTALL) if all([skill_analysis_match, overall_performance_match, conversation_analysis_match, recommendations_match]): reconstructed = "{" reconstructed += skill_analysis_match.group() + "," reconstructed += overall_performance_match.group() + "," reconstructed += conversation_analysis_match.group() + "," reconstructed += recommendations_match.group() reconstructed += "}" reconstructed = self._sanitize_json_text(reconstructed) analysis = json.loads(reconstructed) print("Successfully reconstructed JSON from sections") return analysis except Exception as e: print(f"JSON reconstruction failed: {e}") # Fallback 3: Create a minimal valid response print("All JSON parsing attempts failed, creating minimal fallback response") return self._create_fallback_assessment() def _create_fallback_assessment(self) -> Dict[str, Any]: """Create a basic assessment when JSON parsing completely fails""" return { "skill_analysis": { "Overall Performance": { "score": 5, "evidence": ["Assessment could not be completed due to technical issues"], "strengths": ["Session completed successfully"], "improvement_areas": ["Unable to provide detailed feedback due to technical issues"] } }, "overall_performance": { "weighted_score": 5, "performance_level": "Assessment Unavailable", "total_turns": 1, "conversation_quality": "Assessment could not be completed due to technical issues" }, "conversation_analysis": { "strengths": ["Session completed"], "critical_moments": ["Assessment unavailable"], "missed_opportunities": ["Unable to analyze due to technical issues"] }, "recommendations": { "immediate_focus": ["Please retry the session for detailed feedback"], "practice_suggestions": ["Technical issues prevented detailed analysis"], "advanced_skills": ["Retry recommended for proper assessment"] } } def health_check(self) -> bool: """Check if Ollama service is available""" try: url = f"{self.base_url}/api/tags" print(f"DEBUG: Checking Ollama health at: {url}") response = requests.get(url, timeout=5) print(f"DEBUG: Health check response status: {response.status_code}") if response.status_code == 200: # Check if our model is available models = response.json().get('models', []) model_names = [model.get('name', '') for model in models] print(f"DEBUG: Available models: {model_names}") if self.model not in model_names: print(f"WARNING: Model '{self.model}' not found in available models") print(f"Available models: {model_names}") return False return True return False except Exception as e: print(f"DEBUG: Health check failed: {e}") return False