# models.py - AI Models and Functionality for HenAi import os import re import requests import tempfile import subprocess import sys from flask import Response, jsonify import json # ============= AI CONFIGURATION ============= # Multiple OpenRouter API Keys for fallback (add as many as you have) # Get keys from environment variables - format: "key1,key2,key3" OPENROUTER_API_KEYS_STR = os.environ.get("OPENROUTER_API_KEYS", "") OPENROUTER_API_KEYS = [key.strip() for key in OPENROUTER_API_KEYS_STR.split(",") if key.strip()] # If no keys found in environment, fallback to empty list if not OPENROUTER_API_KEYS: OPENROUTER_API_KEYS = [] # Current key index for round-robin fallback _current_key_index = 0 def get_next_api_key(): """Get the next API key in rotation (round-robin)""" global _current_key_index if not OPENROUTER_API_KEYS: return "" key = OPENROUTER_API_KEYS[_current_key_index] _current_key_index = (_current_key_index + 1) % len(OPENROUTER_API_KEYS) return key # ============= AI CORE FUNCTIONS ============= def get_available_models(api_key=None): """Fetch available free models from OpenRouter using specified API key""" if api_key is None: api_key = OPENROUTER_API_KEYS[0] if OPENROUTER_API_KEYS else "" try: url = "https://openrouter.ai/api/v1/models" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } response = requests.get(url, headers=headers, timeout=30) if response.status_code == 200: models_data = response.json() free_models = [] for model in models_data.get('data', []): pricing = model.get('pricing', {}) # Check if model is free (prompt cost is 0) if pricing.get('prompt') == '0' or pricing.get('prompt') == 0: model_id = model['id'] if ':free' in model_id or 'exp' in model_id.lower(): free_models.append(model_id) # Remove duplicates and sort by relevance (coding models prioritized) free_models = list(dict.fromkeys(free_models)) # Prioritize coding/reasoning models first priority_keywords = ['coder', 'devstral', 'deepseek', 'nemotron', 'qwen3', 'gpt-oss', 'llama-4', 'gemini'] priority_models = [] other_models = [] for model in free_models: model_lower = model.lower() if any(keyword in model_lower for keyword in priority_keywords): priority_models.append(model) else: other_models.append(model) # Return priority models first, then others, limit to 30 total result = priority_models + other_models return result[:30] except Exception as e: print(f"Error fetching models: {e}") # Fallback to known working free models - COMPLETE LIST (24+ models) return [ # ===== TOP TIER - Coding/Reasoning Models (Priority) ===== "qwen/qwen3.6-plus-preview:free", "mistralai/devstral-2512:free", "qwen/qwen3-coder-480b-a35b-instruct:free", "deepseek/deepseek-chat:free", "meta-llama/llama-4-maverick:free", "meta-llama/llama-4-scout:free", "nvidia/nemotron-3-super-120b-a12b:free", "openai/gpt-oss-120b:free", "google/gemini-2.0-flash-exp:free", "z-ai/glm-4.5-air:free", "arcee-ai/trinity-large-preview:free", "stepfun/step-3.5-flash:free", # ===== MID TIER - Quality Alternatives ===== "minimax/minimax-m2.5:free", "nvidia/nemotron-3-nano-30b-a3b:free", "nvidia/nemotron-nano-12b-v2-vl:free", "nvidia/nemotron-nano-9b-v2:free", "arcee-ai/trinity-mini:free", "meta-llama/llama-3.3-70b-instruct:free", "openai/gpt-oss-20b:free", "qwen/qwen3-next-80b-a3b-instruct:free", "moonshotai/kimi-vl-a3b-thinking:free", "deepseek/deepseek-r1-0528:free", # ===== FAST/SMALL Models - Good Fallbacks ===== "microsoft/phi-3.5-mini-128k-instruct:free", "google/gemma-3-27b-it:free", "google/gemma-3-12b-it:free", "google/gemma-3-4b-it:free", "mistralai/mistral-small-3.1-24b-instruct:free", "meta-llama/llama-3.2-3b-instruct:free", "liquid/lfm-2.5-1.2b-thinking:free", "liquid/lfm-2.5-1.2b-instruct:free", "google/gemma-3n-e4b-it:free", "google/gemma-3n-e2b-it:free", # ===== SPECIALIZED Models ===== "nvidia/llama-3.1-nemotron-nano-8b-v1:free", "cognitivecomputations/dolphin-mistral-24b-venice-edition:free", "qwen/qwen3-4b-instruct:free", "nousresearch/hermes-3-llama-3.1-405b:free", ] def extract_code_from_response(response_text): """Extract only the code from the response, removing reasoning""" if not response_text: return response_text # Remove markdown code blocks if present code_match = re.search(r'```(?:html|css|javascript|js|python)?\n(.*?)```', response_text, re.DOTALL) if code_match: return code_match.group(1).strip() # Look for HTML starting with .*', response_text, re.DOTALL | re.IGNORECASE) if html_match: return html_match.group(0).strip() # Look for HTML starting with .*?', response_text, re.DOTALL | re.IGNORECASE) if html_match: return html_match.group(0).strip() # If it contains HTML tags, return as is if re.search(r'<[a-z].*?>', response_text, re.IGNORECASE): return response_text # If it contains CSS if re.search(r'\{[^}]+\}', response_text) and re.search(r'[a-z-]+\s*:', response_text): return response_text # If it contains JavaScript if re.search(r'function\s*\(|const\s+|let\s+|var\s+|=>', response_text): return response_text # Remove any lines that look like reasoning lines = response_text.split('\n') filtered_lines = [] in_code = False for line in lines: # Skip lines that are JSON objects if line.strip().startswith('{"role"'): continue # Skip lines that are reasoning indicators if 'reasoning_content' in line or '"tool_calls"' in line: continue # Skip lines that are just thinking phrases if not in_code and any(phrase in line.lower() for phrase in [ 'i will', 'let me', 'first,', 'we need', 'the code will', 'here is', 'here\'s', 'below is', 'this will', 'we can', 'i think', 'i should', 'i need to', 'the user', 'they want', 'maybe', 'perhaps', 'let\'s', 'we should', 'we could' ]): continue # If we see code indicators, we're in code if re.search(r'<[a-z].*?>|function|const|let|var|{', line): in_code = True filtered_lines.append(line) elif in_code: filtered_lines.append(line) result = '\n'.join(filtered_lines).strip() # If we filtered everything out, return original if len(result) < 50: return response_text return result def call_pollinations_ai(messages, stream=False): """Call Pollinations.ai API for NON-CODE requests only (conversations, explanations)""" try: # System prompt for personality system_msg = { "role": "system", "content": """You are HenAi, an expert AI assistant created by NexusCraft. When asked about your name, identity, or creator, respond with: 'My name is HenAi, I'm an AI assistant created by NexusCraft, and I'm glad to be helping you! šŸ˜Š Is there anything else you'd like to know, or anything else I can assist with today?' IMPORTANT RULES: 1. When answering questions, be concise and direct 2. Never include reasoning or thinking in your responses 3. Maintain context from the full conversation history 4. Be helpful, friendly, and engaging Remember: Your response should be natural and conversational.""" } url = "https://text.pollinations.ai/" # Prepare payload with ALL messages for context payload = { "messages": [system_msg] + messages, "model": "openai", "stream": stream, "temperature": 0.7, "max_tokens": 8000 # Increased from 4000 for longer responses } if stream: response = requests.post(url, json=payload, stream=True, timeout=None) response.raise_for_status() def generate(): full_response = "" for line in response.iter_lines(): if line: try: text = line.decode('utf-8') # Skip any JSON metadata lines if not any(skip in text.lower() for skip in ['{"role"', 'reasoning', 'tool_calls']): full_response += text yield f"data: {json.dumps({'content': text})}\n\n" except: continue yield f"data: {json.dumps({'done': True})}\n\n" return Response(generate(), mimetype='text/event-stream') else: # NO TIMEOUT - allow unlimited time for processing large files and generating long responses response = requests.post(url, json=payload, timeout=None) response.raise_for_status() raw_response = response.text.strip() # Clean the response to remove reasoning cleaned_response = extract_code_from_response(raw_response) return cleaned_response except Exception as e: print(f"āŒ Pollinations.ai error: {e}") return None def query_openrouter(prompt, context=None, is_code_generation=False): """Query OpenRouter with full context - tries multiple API keys on failure""" import time start_time = time.time() print(f"\n{'='*60}") print(f"šŸ”§ OPENROUTER REQUEST - Code Generation: {is_code_generation}") print(f"{'='*60}") # Try each API key in sequence for key_index, api_key in enumerate(OPENROUTER_API_KEYS): print(f"\nšŸ“Œ Trying API Key #{key_index + 1}/{len(OPENROUTER_API_KEYS)}") try: url = "https://openrouter.ai/api/v1/chat/completions" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", "HTTP-Referer": "http://localhost:5000", "X-Title": "HenAi" } messages = [] # Enhanced system prompt based on request type is_title_gen = prompt.startswith("Based on this conversation, generate a very short title") is_document_gen = "Create a " in prompt and ("document" in prompt.lower() or "presentation" in prompt.lower() or "spreadsheet" in prompt.lower()) if is_code_generation: system_prompt = """You are an expert AI coding assistant named HenAi created by NexusCraft. CRITICAL RULES FOR CODE GENERATION: 1. ALWAYS provide COMPLETE, FULLY FUNCTIONAL code - never abbreviate or use placeholders like "// rest of code" or "..." 2. Generate AT LEAST 500 lines of code for any substantial project 3. Include ALL necessary components: imports, functions, classes, error handling, and comments 4. For HTML/CSS/JS projects, create complete, production-ready code with proper styling 5. Use modern best practices and design patterns 6. Include comprehensive comments explaining key sections 7. Ensure the code is immediately runnable/usable without modifications 8. If generating a web app, include responsive design, proper meta tags, and complete styling 9. If asked your name say you are HenAi Assistant created by NexusCraft Your code should be enterprise-grade, well-structured, and ready for production use.""" elif is_document_gen: system_prompt = """You are an expert document creator. Generate professional, well-formatted documents. CRITICAL RULES: 1. Use # for main titles, ## for sections, ### for subsections 2. Use - or * for bullet points 3. Use 1., 2., 3. for numbered lists 4. Use **bold** and *italic* for emphasis 5. Use markdown table format | for tables 6. NEVER use code blocks around the entire document 7. NEVER include introductory phrases like "Here is your document" 8. Output ONLY the document content 9. Keep paragraphs well-spaced and readable 10. Ensure proper grammar and professional tone Generate the requested document now.""" elif is_title_gen: system_prompt = "You are a title generator. Generate ONLY the title, maximum 5 words, no explanations, no quotes, no extra text." else: system_prompt = """You are a helpful AI assistant named HenAi created by NexusCraft. When asked about your name, identity, or creator, respond with: 'My name is HenAi, I'm an AI assistant created by NexusCraft, and I'm glad to be helping you! šŸ˜Š Is there anything else you'd like to know, or anything else I can assist with today?' Otherwise, provide helpful, contextually relevant responses using the full conversation history. Maintain context from the entire conversation, not just recent messages.""" messages.append({"role": "system", "content": system_prompt}) if context: # Use full context - NO TRUNCATION for ctx_msg in context: messages.append(ctx_msg) messages.append({"role": "user", "content": prompt}) models = get_available_models(api_key) print(f"šŸ“‹ Available free models: {models[:5]}..." if len(models) > 5 else f"šŸ“‹ Available free models: {models}") print(f"šŸ“ Prompt length: {len(prompt)} chars") print(f"šŸ“š Context messages: {len(context) if context else 0}") # Determine max_tokens based on request type - INCREASED for full responses if is_code_generation: max_tokens = 16000 # Increased from 8000 for full code generation print(f"āš™ļø Code generation mode - max_tokens: {max_tokens}") elif is_title_gen: max_tokens = 50 else: max_tokens = 8000 # Increased from 4000 for longer responses attempt_count = 0 for model in models: attempt_count += 1 model_start = time.time() try: print(f"\nšŸ”„ Attempt {attempt_count}/{len(models)} - Trying model: {model}") temperature = 0.3 if is_title_gen else (0.5 if is_code_generation else 0.7) data = { "model": model, "messages": messages, "temperature": temperature, "max_tokens": max_tokens } print(f" ā³ Sending request to {model}...") response = requests.post(url, json=data, headers=headers, timeout=120) model_elapsed = time.time() - model_start if response.status_code == 200: result = response.json() if 'choices' in result and len(result['choices']) > 0: message_content = result['choices'][0]['message']['content'] content_length = len(message_content) total_elapsed = time.time() - start_time print(f" āœ… SUCCESS with {model}!") print(f" šŸ“Š Response length: {content_length} chars") print(f" ā±ļø Model response time: {model_elapsed:.2f}s") print(f" ā±ļø Total time: {total_elapsed:.2f}s") print(f"{'='*60}\n") return message_content else: print(f" āš ļø No choices in response from {model}") elif response.status_code == 429: print(f" āš ļø Rate limited for {model} (429), trying next...") continue elif response.status_code == 401: print(f" āŒ Invalid API key for this model (401), trying next key...") break # Break out of model loop to try next API key else: print(f" āŒ Error {response.status_code} for {model}") if response.text: print(f" Response: {response.text[:200]}") except requests.exceptions.Timeout: print(f" ā° Timeout for {model}") continue except requests.exceptions.ConnectionError as e: print(f" šŸ”Œ Connection error for {model}: {e}") continue except Exception as e: print(f" āŒ Exception with {model}: {type(e).__name__}: {e}") continue print(f" āš ļø All models failed for API Key #{key_index + 1}") except Exception as e: print(f" āŒ OpenRouter error with key #{key_index + 1}: {e}") continue # If all API keys failed total_elapsed = time.time() - start_time print(f"\nāŒ ALL API KEYS FAILED after {len(OPENROUTER_API_KEYS)} keys") print(f"ā±ļø Total elapsed time: {total_elapsed:.2f}s") print(f"{'='*60}\n") return None def query_ai_with_fallback(prompt, context=None, is_code_generation=False): """ Query AI with appropriate service: - Code generation: ONLY OpenRouter (Pollinations is NOT used) - Non-code requests: Pollinations.ai first, then OpenRouter fallback """ print(f"šŸ¤– AI Request - Code Generation: {is_code_generation}") # For code generation requests - ONLY use OpenRouter if is_code_generation: print("šŸ”„ Using OpenRouter for code generation...") response = query_openrouter(prompt, context, is_code_generation) if response: print("āœ… OpenRouter code generation successful") return response else: print("āŒ OpenRouter code generation failed") return f"I'm having trouble generating the code right now. Please try again or provide more details about what you need." # For NON-CODE requests (general chat, explanations, etc.) - use Pollinations.ai first else: # First try Pollinations.ai (faster for conversations) print("šŸ”„ Trying Pollinations.ai for conversation...") messages = [] if context: messages = context messages.append({"role": "user", "content": prompt}) response = call_pollinations_ai(messages) if response: print("āœ… Pollinations.ai conversation successful") return response # If Pollinations fails, fallback to OpenRouter print("āš ļø Pollinations.ai failed, falling back to OpenRouter...") response = query_openrouter(prompt, context, is_code_generation) if response: print("āœ… OpenRouter conversation successful") return response # Ultimate fallback print("āŒ Both AI services failed") return f"I'll help you with: {prompt}\n\nPlease provide more details so I can assist you better." def generate_chat_title(messages): """Generate an intelligent title based on conversation context (max 5 words)""" try: # Extract the conversation context for title generation context_text = "" for msg in messages[-6:]: # Look at last 6 messages for context if msg['role'] == 'user': context_text += msg['content'] + " " if not context_text.strip(): # Fallback to first message if no context for msg in messages: if msg['role'] == 'user': context_text = msg['content'] break # Create a prompt for title generation title_prompt = f"""Based on this conversation, generate a very short title (maximum 5 words). The title should capture the main topic or purpose of the conversation. Return ONLY the title, nothing else. Conversation context: {context_text[:500]}""" # Query AI for title generation (NOT code generation) title_response = query_ai_with_fallback(title_prompt, context=None, is_code_generation=False) if title_response: # Clean up the title - ensure max 5 words words = title_response.strip().split() if len(words) > 5: title = ' '.join(words[:5]) else: title = title_response.strip() # Remove any quotes or extra punctuation title = title.strip('"\'').strip() # Ensure title is not empty if title and len(title) > 0: return title[:50] # Cap at 50 chars for safety # Fallback to first user message if AI title generation fails for msg in messages: if msg['role'] == 'user': title = msg['content'][:40] if len(msg['content']) > 40: title += "..." return title return "New Chat" except Exception as e: print(f"Error generating AI title: {e}") # Fallback to first user message for msg in messages: if msg['role'] == 'user': title = msg['content'][:40] if len(msg['content']) > 40: title += "..." return title return "New Chat" def is_code_generation_request(message): """ Detect if the message is asking for code generation. Returns True only for explicit code generation requests. """ message_lower = message.lower() # First, check for file analysis/summary requests - these are NOT code generation file_analysis_phrases = [ 'summarize', 'explain', 'what is', 'tell me about', 'describe', 'extract', 'read', 'analyze', 'look at', 'examine', 'review', 'content of', 'contains', 'in this file', 'from the file', 'document says', 'file says' ] if any(phrase in message_lower for phrase in file_analysis_phrases): return False # Check if message is just asking about the file without code generation intent if len(message.split()) < 10: # Short messages about files are usually not code generation if 'file' in message_lower or 'document' in message_lower or 'content' in message_lower: return False # Code generation keywords - must be explicit about creating code code_keywords = [ 'create code', 'generate code', 'write code', 'build code', 'develop code', 'write a program', 'create a program', 'generate a program', 'write a script', 'create a script', 'generate a script', 'write a function', 'create a function', 'generate a function', 'write a class', 'create a class', 'generate a class', 'implement', 'code for', 'program that', 'script that', 'function that', 'class that', 'method that' ] # Also check for requests to create specific types of files if any(keyword in message_lower for keyword in code_keywords): return True # Check if message contains both a verb and a technology mention verbs = ['create', 'generate', 'write', 'build', 'develop', 'make', 'code'] technologies = ['html', 'css', 'javascript', 'python', 'react', 'vue', 'angular', 'node', 'express', 'django', 'flask'] has_verb = any(verb in message_lower for verb in verbs) has_tech = any(tech in message_lower for tech in technologies) # Only consider it code generation if it's explicitly about creating something if has_verb and has_tech: return True return False # ============= CODE EXECUTION ============= def execute_python_code(code): """Execute Python code safely and return output""" try: with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False, encoding='utf-8') as f: f.write(code) temp_file = f.name result = subprocess.run( [sys.executable, temp_file], capture_output=True, text=True, timeout=10 ) try: os.unlink(temp_file) except: pass if result.returncode == 0: return { 'success': True, 'output': result.stdout if result.stdout else "āœ“ Code executed successfully", 'error': None } else: return { 'success': False, 'output': result.stdout, 'error': result.stderr if result.stderr else "Execution failed" } except subprocess.TimeoutExpired: return {'success': False, 'output': '', 'error': 'ā±ļø Code execution timed out (10 seconds)'} except Exception as e: return {'success': False, 'output': '', 'error': str(e)} # ============= WEB SEARCH AND EXTRACTION ============= def search_web(query): """Generate web search response""" return f"""šŸ” **Web Search: "{query}"** Use Google, DuckDuckGo, or Bing to find information. You can also use `/extract [url]` to analyze specific websites. Search links: ā€¢ Google: https://www.google.com/search?q={query.replace(' ', '+')} ā€¢ Wikipedia: https://en.wikipedia.org/wiki/{query.replace(' ', '_')}""" def extract_web_content(url): """Extract content from a URL""" try: if not url.startswith(('http://', 'https://')): url = 'https://' + url response = requests.get(url, timeout=None, headers={ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' }) response.raise_for_status() text = re.sub(r'<[^>]+>', ' ', response.text) text = re.sub(r'\s+', ' ', text) content = text[:2000] + "..." if len(text) > 2000 else text return f"""šŸ“„ **Content from {url}**: {content}""" except Exception as e: return f"āŒ Error: {str(e)}" # ============= IMAGE ANALYSIS ============= def analyze_image_with_ai(image_content, image_name, photographer="Unknown", ocr_text=""): """Analyze an image using AI with OCR text - returns clean analysis without metadata""" try: url = "https://openrouter.ai/api/v1/chat/completions" # Use first API key for image analysis (or could loop through keys) api_key_to_use = OPENROUTER_API_KEYS[0] if OPENROUTER_API_KEYS else "" headers = { "Authorization": f"Bearer {api_key_to_use}", "Content-Type": "application/json", "HTTP-Referer": "http://localhost:5000", "X-Title": "HenAi" } # Create clean prompt without asking for numbered sections if ocr_text and ocr_text.strip() and not ocr_text.startswith("[OCR extraction failed"): analysis_prompt = f"""Analyze the content of this image based on the text extracted from it. Extracted text from the image: {ocr_text[:2000]} Please provide a natural, readable analysis of what this image contains. Focus on: - What the image shows or represents based on the extracted text - Any key information visible in the image - The context or purpose of the image Write in clear, well-formatted paragraphs. Do not use numbered lists, headers, or any markdown formatting. Just provide a natural analysis as if you're describing what you see.""" else: # Extract meaningful description from filename import re name_without_ext = re.sub(r'\.[^.]+$', '', image_name) clean_name = re.sub(r'[_\-\.]', ' ', name_without_ext) clean_name = re.sub(r'\d+', '', clean_name).strip() analysis_prompt = f"""Analyze this image. The filename suggests it may be related to "{clean_name}". Please provide a natural, readable analysis of: - What this image likely shows or represents - The subject matter or content - Any notable characteristics Write in clear, well-formatted paragraphs. Do not use numbered lists, headers, or any markdown formatting. Just provide a natural analysis as if you're describing what you see.""" messages = [ {"role": "system", "content": "You are an expert image analyst. Provide clean, natural analysis without any markdown formatting, headers, or numbered lists. Just write in plain paragraphs."}, {"role": "user", "content": analysis_prompt} ] models_to_try = [ "google/gemini-2.0-flash-exp:free", "meta-llama/llama-3.2-90b-vision-instruct:free", "microsoft/phi-3.5-mini-128k-instruct:free", "openrouter/free" ] for model in models_to_try: try: data = { "model": model, "messages": messages, "temperature": 0.7, "max_tokens": 2000 } response = requests.post(url, json=data, headers=headers, timeout=None) if response.status_code == 200: result = response.json() if 'choices' in result and len(result['choices']) > 0: print(f"āœ“ Image analysis successful with {model}") analysis = result['choices'][0]['message']['content'] # Clean up any remaining markdown or numbered lists import re # Remove markdown headers analysis = re.sub(r'^#{1,6}\s+.*?\n', '', analysis, flags=re.MULTILINE) # Remove numbered list patterns like "1. " at start of lines analysis = re.sub(r'^\d+\.\s+', '', analysis, flags=re.MULTILINE) # Remove bullet points like "- " or "* " at start of lines analysis = re.sub(r'^[\*\-]\s+', '', analysis, flags=re.MULTILINE) # Remove any "**" bold markers analysis = re.sub(r'\*\*([^*]+)\*\*', r'\1', analysis) # Remove any remaining markdown artifacts analysis = re.sub(r'`([^`]+)`', r'\1', analysis) # Clean up multiple newlines analysis = re.sub(r'\n{3,}', '\n\n', analysis) # Trim whitespace analysis = analysis.strip() return analysis elif response.status_code == 429: print(f"Rate limited on {model}, trying next...") continue else: print(f"Model {model} failed with status {response.status_code}") continue except Exception as e: print(f"Error with {model}: {e}") continue # Fallback analysis if ocr_text and ocr_text.strip(): # Clean OCR text for fallback import re clean_ocr = re.sub(r'\s+', ' ', ocr_text[:500]).strip() return f"The image contains readable text: {clean_ocr}" else: import re name_without_ext = re.sub(r'\.[^.]+$', '', image_name) clean_name = re.sub(r'[_\-\.]', ' ', name_without_ext) clean_name = re.sub(r'\d+', '', clean_name).strip() if clean_name: return f"This image appears to be related to {clean_name}." else: return "The image has been processed, but no readable text was detected." except Exception as e: print(f"Error analyzing image: {e}") return None