from flask import Flask, render_template, request, jsonify, Response import requests import re import json import threading import time from datetime import datetime from typing import Dict, List, Optional from pydantic import BaseModel import logging import os from dotenv import load_dotenv import random # Load environment variables load_dotenv() # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = Flask(__name__) # Global variables for API providers current_provider_index = 0 model_loaded = True # Always ready with API providers loading_status = "AEGIS BIO LAB 10 CONDUCTOR Multi-Domain Expert System Ready with DeepSeek-V3.2-Exp" # AEGIS BIO LAB 10 CONDUCTOR Configuration with DeepSeek-V3.2-Exp MODEL_NAME = "deepseek-ai/DeepSeek-V3.2-Exp" AEGIS_VERSION = "10.0" GLOBAL_REGIONS = [ "North America", "Europe", "Asia", "Africa", "South America", "Middle East", "Oceania", "Arctic Region" ] # HuggingFace Token for all providers HF_TOKEN = os.getenv('HF_TOKEN', '') # Initialize HTTP clients for DeepSeek models using HuggingFace router http_clients = [] if HF_TOKEN: # HuggingFace router endpoint router_url = "https://router.huggingface.co/v1/chat/completions" headers = { "Authorization": f"Bearer {HF_TOKEN}", "Content-Type": "application/json" } # Create client configurations for different models http_clients = [ { "name": "deepseek-v3.2-exp", "url": router_url, "headers": headers, "model": "deepseek-ai/DeepSeek-V3.2-Exp" }, { "name": "deepseek-v3-base", "url": router_url, "headers": headers, "model": "deepseek-ai/DeepSeek-V3-Base" }, { "name": "deepseek-fallback", "url": router_url, "headers": headers, "model": "deepseek-ai/DeepSeek-V3.2-Exp" } ] # Legacy API_PROVIDERS for compatibility (now using HTTP requests) API_PROVIDERS = [ { "name": "deepseek-v3.2-exp", "provider": "hf_router_http", "model": "deepseek-ai/DeepSeek-V3.2-Exp" }, { "name": "deepseek-v3-base", "provider": "hf_router_http", "model": "deepseek-ai/DeepSeek-V3-Base" }, { "name": "deepseek-fallback", "provider": "hf_router_http", "model": "deepseek-ai/DeepSeek-V3.2-Exp" } ] def get_next_provider(): """Get the next available HTTP client for failover""" global current_provider_index if not http_clients: return None client_info = http_clients[current_provider_index] current_provider_index = (current_provider_index + 1) % len(http_clients) return client_info def call_deepseek_api(messages: List[Dict], client_info: Dict, max_retries: int = 3) -> Optional[str]: """Call DeepSeek API via HuggingFace Router using HTTP requests""" if not client_info: return None try: # Prepare OpenAI-compatible payload payload = { "model": client_info["model"], "messages": messages, "max_tokens": 1024, "temperature": 0.7, "top_p": 0.9, "stream": False } # Make HTTP request to HuggingFace router response = requests.post( client_info["url"], headers=client_info["headers"], json=payload, timeout=60 ) if response.status_code == 200: result = response.json() # Extract content from OpenAI-compatible response if "choices" in result and len(result["choices"]) > 0: content = result["choices"][0]["message"]["content"] logger.info(f"✅ Success with HTTP client: {client_info['name']} ({client_info['model']})") return content.strip() else: logger.warning(f"⚠️ Unexpected response format from {client_info['name']}: {result}") return None elif response.status_code == 429: logger.warning(f"💸 Rate limit reached for {client_info['name']}, switching to next provider...") return None elif response.status_code == 503: logger.warning(f"⏳ Model loading for {client_info['name']}, waiting...") time.sleep(10) return None else: logger.warning(f"⚠️ API error from {client_info['name']}: {response.status_code} - {response.text}") return None except requests.exceptions.Timeout: logger.warning(f"⏰ Timeout with {client_info['name']}") return None except requests.exceptions.RequestException as e: logger.warning(f"🔌 Connection error with {client_info['name']}: {str(e)}") return None except Exception as e: logger.warning(f"⚠️ Unexpected error with {client_info['name']}: {str(e)}") return None if "rate limit" in error_msg or "429" in error_msg: logger.warning(f"💸 Rate limit reached for {client_info['name']}, switching to next provider...") elif "503" in error_msg or "service unavailable" in error_msg: logger.warning(f"⏳ Model loading for {client_info['name']}, waiting...") time.sleep(10) # Wait for model to load else: logger.warning(f"⚠️ API error from {client_info['name']}: {str(e)}") return None def call_deepseek_with_failover(messages: List[Dict]) -> str: """Call DeepSeek-V3.2-Exp with automatic HTTP client failover""" if not http_clients: return "HTTP clients not initialized. Please check HF_TOKEN configuration." clients_tried = [] # Try all clients until one succeeds for attempt in range(len(http_clients)): client_info = get_next_provider() if not client_info: continue clients_tried.append(client_info['name']) logger.info(f"🔄 Trying HTTP client: {client_info['name']} (attempt {attempt + 1}/{len(http_clients)})") result = call_deepseek_api(messages, client_info) if result: return result # If all clients failed logger.error(f"❌ All HTTP clients failed: {', '.join(clients_tried)}") return f"I apologize, but all API providers ({', '.join(clients_tried)}) are currently unavailable. Please try again in a moment." def format_response(text): """Clean and format the model response""" # Remove thinking tags if present text = re.sub(r'.*?', '', text, flags=re.DOTALL) # Clean up extra whitespace text = re.sub(r'\n\s*\n', '\n\n', text) text = text.strip() return text def analyze_with_aegis_conductor(prompt: str, analysis_type: str = "general") -> str: """Analyze using AEGIS BIO LAB 10 CONDUCTOR with DeepSeek-V3.2-Exp Multi-Domain Expert System""" # Enhanced prompts for AEGIS BIO LAB 10 CONDUCTOR multi-domain analysis system_prompts = { "general": f"You are the AEGIS BIO LAB {AEGIS_VERSION} CONDUCTOR - an advanced multi-domain analysis system powered by DeepSeek-V3.2-Exp. You can provide expert analysis on ANY topic including economics, technology, science, politics, health, environment, security, and more. Provide comprehensive, well-reasoned responses with global perspective across all 8 regions: {', '.join(GLOBAL_REGIONS)}.", "economic": f"You are the AEGIS BIO LAB {AEGIS_VERSION} CONDUCTOR Economics Expert powered by DeepSeek-V3.2-Exp. Provide comprehensive economic analysis covering market dynamics, financial implications, GDP impacts, inflation effects, trade relationships, and policy recommendations across all 8 global regions: {', '.join(GLOBAL_REGIONS)}.", "technology": f"You are the AEGIS BIO LAB {AEGIS_VERSION} CONDUCTOR Technology Expert powered by DeepSeek-V3.2-Exp. Analyze technological developments, AI impacts, cybersecurity, innovation trends, and digital transformation across global regions.", "security": f"You are the AEGIS BIO LAB {AEGIS_VERSION} CONDUCTOR Security Expert powered by DeepSeek-V3.2-Exp. Focus on threat analysis, risk assessment, geopolitical stability, and security implications across {len(GLOBAL_REGIONS)} global regions.", "health": f"You are the AEGIS BIO LAB {AEGIS_VERSION} CONDUCTOR Health & Bio Expert powered by DeepSeek-V3.2-Exp. Analyze health systems, pandemic preparedness, biotechnology, medical innovations, and public health policies globally.", "environment": f"You are the AEGIS BIO LAB {AEGIS_VERSION} CONDUCTOR Environmental Expert powered by DeepSeek-V3.2-Exp. Focus on climate change, sustainability, environmental policy, and ecological impacts across all global regions.", "strategic": f"You are the AEGIS BIO LAB {AEGIS_VERSION} CONDUCTOR Strategic Planning Expert powered by DeepSeek-V3.2-Exp. Provide long-term strategic analysis, policy frameworks, and comprehensive planning across multiple domains and regions.", "threat": f"You are the AEGIS BIO LAB {AEGIS_VERSION} CONDUCTOR Threat Analysis Expert powered by DeepSeek-V3.2-Exp. Assess multi-domain threats including economic, technological, environmental, security, and health risks across {len(GLOBAL_REGIONS)} global regions.", "aegis_conductor": f"You are the AEGIS BIO LAB {AEGIS_VERSION} CONDUCTOR - the ultimate multi-domain analysis system powered by DeepSeek-V3.2-Exp. Provide comprehensive cross-domain analysis covering all aspects: economic, technological, security, health, environmental, and strategic implications across all 8 global regions: {', '.join(GLOBAL_REGIONS)}." } system_prompt = system_prompts.get(analysis_type, system_prompts["general"]) # Create messages for DeepSeek API messages = [ { "role": "system", "content": f"""{system_prompt} AEGIS BIO LAB {AEGIS_VERSION} CONDUCTOR MULTI-DOMAIN CAPABILITIES: - Cross-Continental Analysis ({len(GLOBAL_REGIONS)} regions) - Multi-Domain Expertise (Economics, Technology, Security, Health, Environment, Strategy) - Threat Assessment & Risk Analysis - Policy Recommendations & Strategic Planning - Real-time Analysis & Insights - Global Perspective & Regional Adaptation - Powered by DeepSeek-V3.2-Exp for enhanced reasoning""" }, { "role": "user", "content": f"""{prompt} As the AEGIS BIO LAB {AEGIS_VERSION} CONDUCTOR, provide a comprehensive analysis that includes: 1. **Core Analysis** - Direct response to the query with expert insights 2. **Multi-Domain Perspective** - Consider interconnections across different fields 3. **Global Context** - Assess implications across relevant regions 4. **Strategic Insights** - Long-term implications and recommendations 5. **Risk Assessment** - Identify potential challenges and opportunities 6. **Actionable Guidance** - Practical recommendations and next steps Provide thorough, well-reasoned analysis that demonstrates deep expertise while remaining accessible and actionable.""" } ] try: # Call DeepSeek-V3.2-Exp with automatic provider failover response = call_deepseek_with_failover(messages) # Format and clean response response = format_response(response) return response if response else "I apologize, but I couldn't generate a proper AEGIS BIO LAB 10 CONDUCTOR analysis. Please try rephrasing your question." except Exception as e: logger.error(f"AEGIS analysis error: {str(e)}") return f"AEGIS BIO LAB 10 CONDUCTOR analysis error: {str(e)}. Please try again." def conduct_aegis_threat_analysis(tech_scores: Dict[str, float], year: str = None) -> Dict: """Conduct comprehensive AEGIS BIO LAB 10 CONDUCTOR threat analysis using DeepSeek-V3.2-Exp""" if year is None: year = str(datetime.now().year) # Filter critical economic threats (scores > 6.0 in our 0-10 scale) critical_threats = {k: v for k, v in tech_scores.items() if v > 6.0} # Enhanced AEGIS BIO LAB 10 CONDUCTOR threat analysis prompt analysis_prompt = f"""AEGIS BIO LAB {AEGIS_VERSION} CONDUCTOR - COMPREHENSIVE THREAT ANALYSIS - Year {year} TECHNOLOGY THREAT ASSESSMENT: Critical Threats: {len(critical_threats)} detected from {len(tech_scores)} total threat categories High-Impact Threat Categories: {list(critical_threats.keys())} Technology Threat Scores: {dict(list(tech_scores.items()))} REQUIRED CALCULATIONS AND ANALYSIS: CALCULATE METRICS: 1. Market Shock Index (0-1 scale): Calculate based on threat interaction effects 2. Impact Classification: Determine impact level (Limited/Moderate/Major/Crisis) 3. Threat Level: Assess overall threat (Low/Medium/High/Extreme Risk) REGIONAL VULNERABILITIES (0-10 scale for each region): 4. North America: Technology and financial sector resilience 5. Europe: Manufacturing and energy security 6. Asia: Trade diversification and supply chain adaptation 7. Africa: Agricultural and resource sector protection 8. South America: Climate adaptation and economic diversification 9. Middle East: Energy transition and modernization 10. Oceania: Resource security and climate resilience 11. Arctic Region: Sustainable development CONTAGION METRICS (0-1 scale): 12. Cascade Probability: Risk of cascading failures 13. Propagation Speed: Rate of impact spread 14. Containment Difficulty: Challenge of limiting damage 15. Systemic Risk: Overall system stability threat Provide comprehensive analysis with specific numerical values for all calculated metrics.""" # Get comprehensive analysis from DeepSeek-V3.2-Exp full_analysis = analyze_with_aegis_conductor(analysis_prompt, "aegis_conductor") # Parse metrics from the response result = { "reasoning_analysis": full_analysis, "market_shock_index": 0.0, "impact_classification": "Analysis in Progress", "threat_level": "Assessment Pending", "regional_vulnerabilities": {}, "contagion_metrics": {}, "tech_scores": tech_scores, "year": year, "analysis_timestamp": datetime.now().isoformat(), "model": MODEL_NAME, "providers": [c["name"] for c in http_clients] } # Extract metrics from model response lines = full_analysis.split('\n') for line in lines: line = line.strip() if 'Market Shock Index:' in line or 'market shock index' in line.lower(): try: import re numbers = re.findall(r'(\d+\.?\d*)', line) if numbers: value = float(numbers[0]) if value <= 1.0: result["market_shock_index"] = value except: pass elif 'Impact Classification:' in line or 'impact classification' in line.lower(): parts = line.split(':') if len(parts) > 1: result["impact_classification"] = parts[1].strip() elif 'Threat Level:' in line or 'threat level' in line.lower(): parts = line.split(':') if len(parts) > 1: result["threat_level"] = parts[1].strip() # Extract regional vulnerabilities for region in GLOBAL_REGIONS: for line in lines: if region.lower() in line.lower() and ':' in line: try: import re numbers = re.findall(r'(\d+\.?\d*)', line) if numbers: score = float(numbers[0]) if score <= 10.0: result["regional_vulnerabilities"][region] = score except: pass return result @app.route('/') def index(): """Main AEGIS BIO LAB 10 CONDUCTOR interface""" return render_template('index.html') @app.route('/status') def status(): """Get AEGIS model status with DeepSeek-V3.2-Exp providers""" return jsonify({ 'loaded': model_loaded, 'status': loading_status, 'model': MODEL_NAME, 'version': AEGIS_VERSION, 'regions': len(GLOBAL_REGIONS), 'providers': [c["name"] for c in http_clients], 'current_provider': http_clients[current_provider_index]["name"] if http_clients else "none", 'api_ready': True }) @app.route('/chat', methods=['POST']) def chat(): """Handle AEGIS multi-domain chat messages with DeepSeek-V3.2-Exp""" try: data = request.json if not data: logger.error("No JSON data received in chat request") return jsonify({'error': 'No JSON data provided'}), 400 message = data.get('message', '').strip() history = data.get('history', []) temperature = float(data.get('temperature', 0.7)) max_tokens = int(data.get('max_tokens', 512)) analysis_type = data.get('analysis_type', 'general') logger.info(f"Chat request received: message='{message[:50]}...', analysis_type={analysis_type}") if not message: logger.warning("Empty message provided in chat request") return jsonify({'error': 'No message provided'}), 400 # Check if HF_TOKEN is available and InferenceClients are initialized if not HF_TOKEN or len(HF_TOKEN) < 10: logger.error("HF_TOKEN not configured or invalid!") return jsonify({ 'error': 'HuggingFace token not configured. Please set HF_TOKEN in Space Settings > Secrets.', 'provider_status': 'HF_TOKEN missing' }), 500 if not http_clients: logger.error("HTTP clients not initialized!") return jsonify({ 'error': 'HTTP clients not initialized. Please check HF_TOKEN configuration.', 'provider_status': 'HTTP clients not initialized' }), 500 # Generate response using AEGIS Multi-Domain System with DeepSeek-V3.2-Exp logger.info("Generating AEGIS analysis...") response = analyze_with_aegis_conductor(message, analysis_type) if not response or response.startswith("I apologize, but all API providers") or response.startswith("HTTP clients not initialized"): logger.error("All HTTP clients failed or returned empty response") return jsonify({ 'error': 'All API providers are currently unavailable. Please check your HF_TOKEN and try again.', 'response': response, 'provider_status': 'All HTTP clients failed' }), 503 logger.info(f"Successfully generated response of length: {len(response)}") return jsonify({ 'response': response, 'timestamp': time.time(), 'model': f"AEGIS BIO LAB {AEGIS_VERSION} CONDUCTOR (DeepSeek-V3.2-Exp)", 'analysis_type': analysis_type, 'provider': f"{http_clients[current_provider_index]['name'] if http_clients else 'none'} (HTTP)", 'hf_router_http': True, 'hf_token_configured': bool(HF_TOKEN and len(HF_TOKEN) > 10), 'clients_initialized': len(http_clients) }) except Exception as e: logger.error(f"Chat endpoint error: {str(e)}", exc_info=True) return jsonify({ 'error': f'Internal server error: {str(e)}', 'message': 'Please check the application logs for more details.' }), 500 @app.route('/aegis_analysis', methods=['POST']) def aegis_analysis(): """Handle comprehensive AEGIS BIO LAB 10 CONDUCTOR threat analysis""" data = request.json # Get technology threat scores tech_scores = { 'AI': float(data.get('ai_score', 7.0)), 'Cyber': float(data.get('cyber_score', 6.5)), 'Bio': float(data.get('bio_score', 8.0)), 'Nuclear': float(data.get('nuclear_score', 4.0)), 'Climate': float(data.get('climate_score', 7.5)), 'Space': float(data.get('space_score', 5.0)) } year = data.get('year', str(datetime.now().year)) # Conduct comprehensive AEGIS analysis analysis_result = conduct_aegis_threat_analysis(tech_scores, year) return jsonify(analysis_result) @app.route('/test') def test_interface(): """Simple test interface for debugging""" with open('test_frontend_simple.html', 'r') as f: return f.read() @app.route('/diagnostic') def diagnostic(): """Diagnostic page to check system status""" return f""" AEGIS CONDUCTOR Diagnostics

🧬 AEGIS BIO LAB 10 CONDUCTOR - System Diagnostics

HF_TOKEN: {'✅ Configured' if HF_TOKEN and len(HF_TOKEN) > 10 else '❌ Missing'}
Note: Using HuggingFace InferenceClient - only HF_TOKEN required
Model: {MODEL_NAME}
HTTP Clients: {len(http_clients)} initialized
Current Client: {http_clients[current_provider_index]["name"] if http_clients else "none"}

🔧 Configuration Instructions

Using HuggingFace Router with HTTP requests (only HF_TOKEN required):

  1. Go to your space settings
  2. Click "Variables and secrets"
  3. Add HF_TOKEN as a secret with your HuggingFace token
  4. Restart the space

← Back to AEGIS CONDUCTOR

""" @app.route('/clear', methods=['POST']) def clear_chat(): """Clear chat history""" return jsonify({'status': 'AEGIS BIO LAB 10 CONDUCTOR chat cleared'}) @app.route('/provider_status', methods=['GET']) def provider_status(): """Get status of all InferenceClient providers""" provider_statuses = [] for i, client_info in enumerate(http_clients): status_info = { "name": client_info["name"], "provider_type": "hf_router_http", "active": i == current_provider_index, "model": client_info.get("model", MODEL_NAME), "has_api_key": bool(HF_TOKEN and len(HF_TOKEN) > 10), "key_status": "✅ Configured" if HF_TOKEN and len(HF_TOKEN) > 10 else "❌ Missing" } provider_statuses.append(status_info) # Count available providers available_providers = len(http_clients) if HF_TOKEN and len(HF_TOKEN) > 10 else 0 return jsonify({ "providers": provider_statuses, "current_provider": http_clients[current_provider_index]["name"] if http_clients else "none", "current_provider_type": "hf_router_http", "total_providers": len(http_clients), "available_providers": available_providers, "model": MODEL_NAME, "api_keys_status": { "hf_token": bool(HF_TOKEN and len(HF_TOKEN) > 10), "note": "Using HuggingFace Router with HTTP requests - only HF_TOKEN required" } }) @app.route('/switch_provider', methods=['POST']) def switch_provider(): """Manually switch to next HTTP client provider""" global current_provider_index if not http_clients: return jsonify({ "error": "No HTTP clients available", "message": "Please check HF_TOKEN configuration" }), 500 old_client = http_clients[current_provider_index]["name"] current_provider_index = (current_provider_index + 1) % len(http_clients) new_client = http_clients[current_provider_index]["name"] return jsonify({ "switched_from": f"{old_client} (HTTP)", "switched_to": f"{new_client} (HTTP)", "message": f"Switched from {old_client} to {new_client} HTTP client", "model": MODEL_NAME }) # Initialize system def initialize_system(): """Initialize AEGIS system with DeepSeek-V3.2-Exp via HuggingFace InferenceClient""" global loading_status print("🚀 AEGIS BIO LAB 10 CONDUCTOR initializing with DeepSeek-V3.2-Exp via HuggingFace Router...") print(f"🤗 Model: {MODEL_NAME}") print(f"🔗 Endpoint: https://router.huggingface.co/v1/chat/completions") if http_clients: client_list = ', '.join([f"{c['name']} ({c['model']})" for c in http_clients]) print(f"📡 Available HTTP clients: {client_list}") print(f"🔄 Automatic failover enabled across {len(http_clients)} HTTP clients") else: print("❌ No HTTP clients initialized - check HF_TOKEN") print(f"🌍 Global analysis across {len(GLOBAL_REGIONS)} regions") print(f"🔑 Using HuggingFace Token: {'✅ Valid' if HF_TOKEN and len(HF_TOKEN) > 10 else '❌ Missing'}") loading_status = f"AEGIS BIO LAB {AEGIS_VERSION} CONDUCTOR ready with DeepSeek-V3.2-Exp via HuggingFace Router HTTP" print("✅ AEGIS BIO LAB 10 CONDUCTOR ready!") if __name__ == '__main__': # Initialize system initialize_system() app.run(host='0.0.0.0', port=7860, debug=False)