#!/usr/bin/env python3 """ OpenRouter Research Swarm — Fire multiple cheap models for parallel research. Uses MiniMax M2.5, DeepSeek V3.2, Qwen3.5 Flash for maximum coverage. """ import os import sys import json import time import asyncio import aiohttp from pathlib import Path from datetime import datetime OPENROUTER_KEY = os.environ.get("OPENROUTER_API_KEY", "") if not OPENROUTER_KEY: # Try loading from secrets.env secrets_path = Path("/mnt/e/genesis-system/secrets.env") if secrets_path.exists(): for line in secrets_path.read_text().splitlines(): if "OPENROUTER" in line and "=" in line: OPENROUTER_KEY = line.split("=", 1)[1].strip().strip('"').strip("'") break if not OPENROUTER_KEY: print("ERROR: No OpenRouter API key found") sys.exit(1) OUTPUT_DIR = Path("/mnt/e/genesis-system/GHL/swarm_output/openrouter") OUTPUT_DIR.mkdir(parents=True, exist_ok=True) # Models to use (cheap + fast) MODELS = [ "minimax/minimax-m2.5", # 196K ctx, $0.30/$1.10 per MTok "qwen/qwen3.5-flash-02-23", # 1M ctx, $0.10/$0.40 per MTok "deepseek/deepseek-v3.2", # 164K ctx, $0.25/$0.40 per MTok ] # Research tasks TASKS = [ { "id": "OR-01-ghl-saas-case-studies", "model": MODELS[0], "prompt": """Research 10 successful GHL SaaS agencies and their business models as of 2026. For each agency, find: 1. Name and URL 2. What SaaS products they sell on GHL 3. Pricing (monthly/annual) 4. Number of customers (if public) 5. Revenue (if public) 6. What snapshots/templates they use 7. What vertical they serve Focus on agencies serving trades, local businesses, and SMBs. Include Extendly, SaaS Academy, GHL Elite, HighLevel Mastery examples. Output as a structured report with a comparison table.""" }, { "id": "OR-02-tradie-market-size", "model": MODELS[1], "prompt": """Research the Australian trades market size and digital adoption as of 2026. Answer these specific questions: 1. Total number of registered trade businesses in Australia (ABS data) 2. Breakdown by trade: plumbers, electricians, HVAC, builders, locksmiths 3. Average revenue per trade business 4. % currently using digital marketing 5. % using a CRM or business management software 6. Top pain points for Australian tradies 7. Digital marketing spend per tradie business 8. Growth rate of the trades sector 9. Average customer acquisition cost 10. Average lifetime value of a customer Cite sources. Use Australian Bureau of Statistics, IBISWorld, and industry reports.""" }, { "id": "OR-03-instantly-best-practices", "model": MODELS[2], "prompt": """Research Instantly.ai best practices for B2B cold email campaigns targeting Australian tradies. Cover: 1. Domain warm-up strategy (how many domains, warm-up period) 2. Email infrastructure setup (DNS records, DKIM, DMARC, SPF) 3. Optimal sending volume per domain per day 4. Subject line formulas that work for tradies 5. Email copy frameworks (PAS, AIDA, etc.) 6. Follow-up cadence (how many emails, spacing) 7. Australian anti-spam compliance (Spam Act 2003) 8. Reply handling automation 9. Lead scoring based on engagement 10. Integration with CRM (GHL) via webhooks Include 3 example email sequences for selling MCTB Chatbot to plumbers.""" }, { "id": "OR-04-review-software-landscape", "model": MODELS[0], "prompt": """Research the complete landscape of review management software as of 2026. Compare these platforms on features, pricing, and target market: 1. ReviewHarvest (Clay Lawrence) 2. Podium 3. Birdeye 4. Reputation.com 5. GatherUp 6. Grade.us 7. NiceJob 8. ReviewTrackers 9. Yext Reviews 10. Trustpilot 11. ReviewShake 12. Whitespark For each: pricing tiers, key features, review platforms supported, API availability, white-label options. Then explain exactly how our GHL-based Review Generator beats each one. Our advantages: $197 AUD/mo, unlimited requests, 40+ platforms, sentiment filtering, full CRM integration, Australian market focus.""" }, { "id": "OR-05-v0-api-deep-dive", "model": MODELS[1], "prompt": """Deep dive into the v0.dev Platform API for programmatic website generation. Research: 1. Complete API documentation at v0.app/docs/api 2. Authentication method (API keys, OAuth) 3. Available endpoints (list all) 4. Create project flow 5. Generate code flow 6. Deploy to Vercel flow 7. Custom domain setup 8. Rate limits and quotas 9. Pricing (credits, subscription tiers) 10. SDK (TypeScript) usage examples Then design a pipeline for generating 25 tradie websites: - Input: trade type, color scheme, business name, services, location - Process: v0 API generates, Claude reviews, auto-deploy to Vercel - Output: Live website on custom domain Include code examples for the API calls.""" }, { "id": "OR-06-ghl-marketplace-apps", "model": MODELS[2], "prompt": """Research the GHL Marketplace as of 2026. What apps and integrations are available? Find: 1. Total number of marketplace apps 2. Categories of apps 3. Top 20 most popular apps and what they do 4. Apps specifically useful for trades/local businesses 5. Review/reputation management apps 6. AI/chatbot apps 7. Website builder integrations 8. Pricing model for marketplace apps 9. How to list your own app on the marketplace 10. Revenue share for marketplace app developers Focus on finding apps that complement our product suite: - MCTB Chatbot - Review Generator - AI Receptionist - Website + SEO What marketplace apps should we install in our master templates?""" }, { "id": "OR-07-creator-economy-2026", "model": MODELS[0], "prompt": """Research the creator economy market size and trends as of 2026. Answer: 1. Total creator economy market size (global and Australia) 2. Number of content creators globally and in Australia 3. Revenue streams for creators (brand deals, courses, memberships, merch, affiliate) 4. Average income by tier (nano, micro, mid, macro, mega) 5. Top creator management platforms and their features 6. Pain points creators face managing their business 7. What tools do creators currently use? 8. Market gap analysis: what's missing? 9. Growth projections 2026-2028 10. Monetization trends (shifting from ad revenue to direct monetization) Then assess: can a GHL-based CreatorDashboard at $97-497/mo capture market share?""" }, { "id": "OR-08-composable-saas-examples", "model": MODELS[1], "prompt": """Research composable SaaS pricing models and how they work in practice. Find 10 examples of SaaS products that offer composable/modular pricing where customers pick individual features: 1. How they structure their pricing page 2. Technical implementation (feature flags, plan tiers, add-ons) 3. How they handle upgrades/downgrades 4. Billing infrastructure (Stripe, Paddle, etc.) 5. Conversion rates of modular vs fixed pricing 6. UX best practices for composable pricing pages Then apply these learnings to our GHL SaaS Configurator architecture: - 6 standalone products ($97-$497) - 5 bundles ($247-$1,297) - 20-plan limit constraint - Feature toggles per plan - Auto-provisioning from snapshots""" }, ] async def run_task(session: aiohttp.ClientSession, task: dict) -> dict: """Run a single research task via OpenRouter.""" task_id = task["id"] model = task["model"] prompt = task["prompt"] print(f"[{task_id}] Starting with {model}...") start = time.time() try: async with session.post( "https://openrouter.ai/api/v1/chat/completions", headers={ "Authorization": f"Bearer {OPENROUTER_KEY}", "Content-Type": "application/json", "HTTP-Referer": "https://genesis-system.local", "X-Title": f"Genesis Research: {task_id}", }, json={ "model": model, "messages": [ { "role": "system", "content": "You are a senior business research analyst. Provide comprehensive, well-structured research reports with specific data, citations, and actionable insights. Use markdown formatting. Be thorough but concise." }, {"role": "user", "content": prompt} ], "max_tokens": 8000, "temperature": 0.3, }, timeout=aiohttp.ClientTimeout(total=120) ) as resp: data = await resp.json() if "error" in data: error_msg = data["error"].get("message", str(data["error"])) print(f"[{task_id}] ERROR: {error_msg}") return {"id": task_id, "status": "error", "error": error_msg} content = data["choices"][0]["message"]["content"] usage = data.get("usage", {}) elapsed = time.time() - start # Save output output_path = OUTPUT_DIR / f"{task_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.md" output_path.write_text(f"# {task_id}\n**Model**: {model}\n**Generated**: {datetime.now().isoformat()}\n**Tokens**: {usage.get('total_tokens', '?')}\n**Time**: {elapsed:.1f}s\n\n---\n\n{content}") print(f"[{task_id}] DONE in {elapsed:.1f}s | {usage.get('total_tokens', '?')} tokens | -> {output_path.name}") return { "id": task_id, "status": "success", "tokens": usage.get("total_tokens"), "time": elapsed, "output": str(output_path) } except asyncio.TimeoutError: print(f"[{task_id}] TIMEOUT after 120s") return {"id": task_id, "status": "timeout"} except Exception as e: print(f"[{task_id}] EXCEPTION: {e}") return {"id": task_id, "status": "error", "error": str(e)} async def main(): print("=" * 60) print(" OPENROUTER RESEARCH SWARM") print(f" Tasks: {len(TASKS)}") print(f" Models: {', '.join(set(t['model'] for t in TASKS))}") print(f" Output: {OUTPUT_DIR}") print(f" Time: {datetime.now().isoformat()}") print("=" * 60) async with aiohttp.ClientSession() as session: # Fire all tasks concurrently results = await asyncio.gather(*[run_task(session, t) for t in TASKS]) # Summary print("\n" + "=" * 60) print(" SWARM RESULTS") print("=" * 60) success = sum(1 for r in results if r["status"] == "success") total_tokens = sum(r.get("tokens", 0) or 0 for r in results) total_time = max(r.get("time", 0) or 0 for r in results) for r in results: status_icon = "OK" if r["status"] == "success" else "FAIL" print(f" [{status_icon}] {r['id']}: {r.get('tokens', '?')} tokens, {r.get('time', 0):.1f}s") print(f"\n Success: {success}/{len(TASKS)}") print(f" Total tokens: {total_tokens:,}") print(f" Wall time: {total_time:.1f}s") print("=" * 60) if __name__ == "__main__": asyncio.run(main())