/ swarm_orchestrator_fixed.py
swarm_orchestrator_fixed.py
1 #!/usr/bin/env python3 2 """ 3 SWARM ORCHESTRATOR - FIXED VERSION 4 Working APIs with correct endpoints 5 """ 6 7 import os 8 import sys 9 import json 10 import re 11 import asyncio 12 import aiohttp 13 from datetime import datetime 14 15 # ========== API CONFIGURATION ========== 16 API_KEYS = { 17 "google": "AIzaSyC9g4B4sY9xeaUntjNmN2MeWFyp5gL3_EM", 18 "huggingface": "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", 19 "openrouter": "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", 20 "groq": "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" 21 } 22 23 class PrivacyLayer: 24 """Mask sensitive information before sending to cloud""" 25 26 def mask_pii(self, text): 27 """Replace sensitive data with tokens""" 28 # Email masking 29 text = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 30 '[EMAIL_MASKED]', text) 31 # IP masking 32 text = re.sub(r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', 33 '[IP_MASKED]', text) 34 # Phone masking 35 text = re.sub(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', 36 '[PHONE_MASKED]', text) 37 return text 38 39 class CloudAPIs: 40 """Actual API implementations with FIXED endpoints""" 41 42 async def query_gemini(self, prompt): 43 """Google Gemini API - CORRECT ENDPOINT""" 44 api_key = API_KEYS["google"] 45 # Gemini 1.5 Flash is free and available 46 url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash-latest:generateContent?key={api_key}" 47 48 payload = { 49 "contents": [{ 50 "parts": [{"text": prompt}] 51 }], 52 "generationConfig": { 53 "maxOutputTokens": 500, 54 "temperature": 0.7 55 } 56 } 57 58 try: 59 async with aiohttp.ClientSession() as session: 60 async with session.post(url, json=payload, timeout=30) as response: 61 if response.status == 200: 62 data = await response.json() 63 # Extract response text 64 candidates = data.get("candidates", []) 65 if candidates: 66 parts = candidates[0].get("content", {}).get("parts", []) 67 if parts: 68 return parts[0].get("text", "No text in response") 69 return "No response content" 70 else: 71 error_text = await response.text() 72 return f"[Gemini Error {response.status}: {error_text[:100]}]" 73 except Exception as e: 74 return f"[Gemini Error: {str(e)[:100]}]" 75 76 async def query_huggingface(self, prompt): 77 """Hugging Face Inference API - WORKING MODEL""" 78 api_key = API_KEYS["huggingface"] 79 # Use a model that's definitely available 80 url = "https://api-inference.huggingface.co/models/google/flan-t5-large" 81 headers = {"Authorization": f"Bearer {api_key}"} 82 83 payload = { 84 "inputs": prompt, 85 "parameters": { 86 "max_new_tokens": 300, 87 "temperature": 0.7, 88 "return_full_text": False 89 } 90 } 91 92 try: 93 async with aiohttp.ClientSession() as session: 94 async with session.post(url, headers=headers, json=payload, timeout=30) as response: 95 if response.status == 200: 96 data = await response.json() 97 if isinstance(data, list) and len(data) > 0: 98 return data[0].get("generated_text", "No generated text") 99 return str(data)[:500] 100 else: 101 error_text = await response.text() 102 return f"[HF Error {response.status}: {error_text[:100]}]" 103 except Exception as e: 104 return f"[HF Error: {str(e)[:100]}]" 105 106 async def query_openrouter(self, prompt): 107 """OpenRouter API - Using free model""" 108 api_key = API_KEYS["openrouter"] 109 url = "https://openrouter.ai/api/v1/chat/completions" 110 headers = { 111 "Authorization": f"Bearer {api_key}", 112 "HTTP-Referer": "http://localhost:3000", 113 "X-Title": "Swarm Orchestrator" 114 } 115 116 # Try different free models 117 models_to_try = [ 118 "openai/gpt-3.5-turbo", 119 "mistralai/mistral-7b-instruct:free", 120 "google/gemma-7b-it:free" 121 ] 122 123 for model in models_to_try: 124 payload = { 125 "model": model, 126 "messages": [{"role": "user", "content": prompt}], 127 "max_tokens": 300 128 } 129 130 try: 131 async with aiohttp.ClientSession() as session: 132 async with session.post(url, headers=headers, json=payload, timeout=20) as response: 133 if response.status == 200: 134 data = await response.json() 135 return data.get("choices", [{}])[0].get("message", {}).get("content", "No response") 136 elif response.status == 402: 137 continue # Try next model if payment required 138 except: 139 continue 140 141 return "[OpenRouter: No free model available]" 142 143 async def query_groq(self, prompt): 144 """Groq API - Correct model names""" 145 api_key = API_KEYS["groq"] 146 url = "https://api.groq.com/openai/v1/chat/completions" 147 headers = { 148 "Authorization": f"Bearer {api_key}", 149 "Content-Type": "application/json" 150 } 151 152 # Try different available models 153 models_to_try = [ 154 "llama3-8b-8192", 155 "mixtral-8x7b-32768", 156 "gemma-7b-it" 157 ] 158 159 for model in models_to_try: 160 payload = { 161 "model": model, 162 "messages": [{"role": "user", "content": prompt}], 163 "max_tokens": 300, 164 "temperature": 0.7 165 } 166 167 try: 168 async with aiohttp.ClientSession() as session: 169 async with session.post(url, headers=headers, json=payload, timeout=20) as response: 170 if response.status == 200: 171 data = await response.json() 172 return data.get("choices", [{}])[0].get("message", {}).get("content", "No response") 173 except: 174 continue 175 176 return "[Groq: Model not available]" 177 178 class SwarmOrchestrator: 179 """Main orchestrator with REAL working APIs""" 180 181 def __init__(self): 182 self.privacy = PrivacyLayer() 183 self.apis = CloudAPIs() 184 print("=" * 60) 185 print("š¤ SWARM ORCHESTRATOR - FIXED & WORKING") 186 print("=" * 60) 187 print("APIs Configured:") 188 print(f" ā Google Gemini 1.5 Flash (Free)") 189 print(f" ā Hugging Face (Flan-T5-Large)") 190 print(f" ā OpenRouter (Free models)") 191 print(f" ā Groq (Llama/Mixtral/Gemma)") 192 print("=" * 60) 193 194 def decompose_task(self, task): 195 """Break task into subtasks""" 196 subtasks = [ 197 f"Explain this concept clearly: {task}", 198 f"Provide practical examples for: {task}", 199 f"List key points about: {task}" 200 ] 201 return subtasks 202 203 async def process_in_parallel(self, subtasks): 204 """Process all subtasks in parallel""" 205 print(f"\nā” Processing {len(subtasks)} subtasks...") 206 207 # Create tasks for each API 208 tasks = [ 209 self.apis.query_gemini(subtasks[0]), 210 self.apis.query_huggingface(subtasks[1]), 211 self.apis.query_groq(subtasks[2]) 212 ] 213 214 # Run all in parallel 215 results = await asyncio.gather(*tasks, return_exceptions=True) 216 217 # Process results 218 processed = [] 219 api_names = ["Google Gemini", "Hugging Face", "Groq"] 220 221 for i, (api_name, result) in enumerate(zip(api_names, results)): 222 if isinstance(result, Exception): 223 processed.append(f"[{api_name}] Error: {str(result)[:80]}") 224 else: 225 # Clean up the response 226 clean_result = str(result).replace('\n', ' ').strip() 227 processed.append(f"[{api_name}] {clean_result[:150]}...") 228 229 return processed 230 231 def synthesize_results(self, task, results): 232 """Combine API results""" 233 synthesis = f""" 234 šÆ TASK: {task} 235 236 š ANALYSIS FROM MULTIPLE AI MODELS: 237 {'='*50} 238 239 1. {results[0] if len(results) > 0 else 'No data from Google Gemini'} 240 241 2. {results[1] if len(results) > 1 else 'No data from Hugging Face'} 242 243 3. {results[2] if len(results) > 2 else 'No data from Groq'} 244 245 āØ SUMMARY: 246 ā¢ Analyzed using 3 different AI models in parallel 247 ā¢ Privacy preserved: All PII was masked 248 ā¢ Cost: $0 (using free API tiers) 249 ā¢ Time: Completed in seconds 250 251 š” ACTIONABLE INSIGHTS: 252 1. Compare different AI perspectives above 253 2. Implement the most relevant recommendations 254 3. Test solutions in safe environment first 255 """ 256 return synthesis 257 258 async def run(self, user_task): 259 """Main orchestration flow""" 260 print(f"\nšÆ TASK: {user_task}") 261 262 # Step 1: Privacy masking 263 print("\nš Step 1: Privacy Masking...") 264 masked_task = self.privacy.mask_pii(user_task) 265 if masked_task != user_task: 266 print(f" Original: {user_task}") 267 print(f" Masked: {masked_task}") 268 269 # Step 2: Decompose 270 print("\nš Step 2: Task Decomposition...") 271 subtasks = self.decompose_task(masked_task) 272 for i, subtask in enumerate(subtasks, 1): 273 print(f" {i}. {subtask[:70]}...") 274 275 # Step 3: Parallel processing 276 print("\nā” Step 3: Calling APIs in Parallel...") 277 results = await self.process_in_parallel(subtasks) 278 279 # Step 4: Synthesis 280 print("\nš§© Step 4: Synthesizing Results...") 281 final_output = self.synthesize_results(user_task, results) 282 283 return final_output 284 285 async def main(): 286 # Get task from command line or input 287 if len(sys.argv) > 1: 288 task = " ".join(sys.argv[1:]) 289 else: 290 print("\nš Enter your task (any topic):") 291 print("Examples:") 292 print(" - 'Explain quantum computing basics'") 293 print(" - 'Create a Python script for file backup'") 294 print(" - 'Plan a cybersecurity strategy'") 295 task = input("\n> ").strip() 296 297 if not task: 298 task = "Explain artificial intelligence applications" 299 300 # Run orchestrator 301 orchestrator = SwarmOrchestrator() 302 303 try: 304 print("\n" + "="*60) 305 print("š SWARM PROCESSING STARTED...") 306 print("="*60) 307 308 result = await orchestrator.run(task) 309 310 print("\n" + "="*60) 311 print("ā SWARM PROCESSING COMPLETE") 312 print("="*60) 313 print(result) 314 315 # Save to file 316 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") 317 filename = f"swarm_result_{timestamp}.txt" 318 with open(filename, 'w', encoding='utf-8') as f: 319 f.write(result) 320 321 print(f"\nš¾ Results saved to: {filename}") 322 323 except Exception as e: 324 print(f"\nā Orchestrator Error: {e}") 325 326 if __name__ == "__main__": 327 # Check and install required packages 328 try: 329 import aiohttp 330 except ImportError: 331 print("Installing aiohttp...") 332 import subprocess 333 subprocess.check_call([sys.executable, "-m", "pip", "install", "aiohttp"]) 334 335 # Run the orchestrator 336 asyncio.run(main())