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() 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 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 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