""" Genesis Triage Agent ==================== Fast keyword-based classifier: routes any task string to the correct general + model tier in < 2 seconds — no LLM call required. Usage: from core.triage_agent import GenesisTriageAgent agent = GenesisTriageAgent() result = agent.triage("Build a Stripe checkout flow for ReceptionistAI") print(result) Output shape: { "general": "FORGE", "complexity": "moderate", "model_tier": 2, "priority": "high", "suggested_model": "minimax/minimax-01", "estimated_cost_usd": 0.006, "parallel_safe": True, "requires_browser": False, "requires_memory_write": False } """ import re import time from typing import Any # --------------------------------------------------------------------------- # Keyword routing tables — order matters (first match wins for general) # --------------------------------------------------------------------------- GENERAL_KEYWORDS: list[tuple[str, list[str]]] = [ ("MNEMOSYNE", [ "memory", "kg", "knowledge graph", "axiom", "rlm", "reward model", "preference", "ingest", "embedding", "titan", "indexing", "vector store", "episodic", "semantic memory", ]), ("PROMETHEUS", [ "mcp", "capability", "install", "evaluate", "model evaluation", "benchmark", "new tool", "skill", "plugin", "integration test", "capability audit", "new model", "evaluate model", ]), ("ATLAS", [ "research", "find", "search", "competitor", "analyse", "analyze", "scrape", "crawl", "market", "pricing", "intelligence", "discover", "audit", "report", "top 20", "survey", "trend", "agency list", "lead list", ]), ("HERMES", [ "demo", "ghl", "email", "outreach", "client", "agency", "onboard", "browser", "login", "fill form", "click", "setup ghl", "workflow", "funnel", "landing page", "sales", "lead capture", "crm", "stripe", "checkout flow", "website", "dns", ]), ("FORGE", [ "build", "code", "implement", "fix", "write", "create", "develop", "deploy", "architect", "prd", "story", "test", "debug", "refactor", "backend", "frontend", "api", "database", "migration", "script", "endpoint", "function", "class", "module", ]), ] # Complexity detection by word count COMPLEXITY_TIERS: list[tuple[str, int, int]] = [ ("simple", 0, 20), ("moderate", 20, 50), ("complex", 50, 150), ("deep", 150, 9999), ] # Model selection per (general, complexity) MODEL_SELECTION: dict[tuple[str, str], str] = { # ATLAS — research ("ATLAS", "simple"): "google/gemini-2.5-flash", ("ATLAS", "moderate"): "google/gemini-2.5-flash", ("ATLAS", "complex"): "google/gemini-2.5-flash", ("ATLAS", "deep"): "deepseek/deepseek-chat", # FORGE — coding ("FORGE", "simple"): "minimax/minimax-01", ("FORGE", "moderate"): "minimax/minimax-01", ("FORGE", "complex"): "minimax/minimax-01", ("FORGE", "deep"): "anthropic/claude-sonnet-4-5", # HERMES — browser/GTM ("HERMES", "simple"): "google/gemini-2.5-flash", ("HERMES", "moderate"):"moonshot/moonshot-v1-8k", ("HERMES", "complex"): "moonshot/moonshot-v1-8k", ("HERMES", "deep"): "anthropic/claude-sonnet-4-5", # MNEMOSYNE — memory ("MNEMOSYNE", "simple"): "deepseek/deepseek-chat", ("MNEMOSYNE", "moderate"): "google/gemini-2.5-flash", ("MNEMOSYNE", "complex"): "google/gemini-2.5-flash", ("MNEMOSYNE", "deep"): "deepseek/deepseek-chat", # PROMETHEUS — capabilities ("PROMETHEUS", "simple"): "anthropic/claude-haiku-4-5", ("PROMETHEUS", "moderate"): "anthropic/claude-haiku-4-5", ("PROMETHEUS", "complex"): "anthropic/claude-sonnet-4-5", ("PROMETHEUS", "deep"): "anthropic/claude-sonnet-4-5", } # Model tier mapping TIER_BY_COMPLEXITY: dict[str, int] = { "simple": 1, "moderate": 2, "complex": 2, "deep": 3, } # Estimated per-call cost in USD (based on HYPERSCALE_HIVE_ARCHITECTURE.md cost table) COST_BY_MODEL: dict[str, float] = { "google/gemini-2.5-flash": 0.002, "minimax/minimax-01": 0.006, "deepseek/deepseek-chat": 0.001, "moonshot/moonshot-v1-8k": 0.003, "anthropic/claude-haiku-4-5": 0.004, "anthropic/claude-sonnet-4-5": 0.012, "anthropic/claude-opus-4-6": 0.020, } # Priority detection keywords PRIORITY_KEYWORDS: dict[str, list[str]] = { "critical": ["urgent", "asap", "critical", "blocker", "now", "immediately", "production down"], "high": ["george demo", "demo", "client", "launch", "ship", "revenue", "high priority"], "low": ["later", "eventually", "nice to have", "someday", "backlog"], } # Browser requirement keywords BROWSER_KEYWORDS = [ "browser", "playwright", "login", "ghl", "click", "fill form", "website", "dns", "stripe checkout", "godaddy", "cloudflare", ] # Memory write requirement keywords MEMORY_WRITE_KEYWORDS = [ "memory", "kg", "axiom", "ingest", "knowledge graph", "rlm", "titan", "learn", "store", "remember", "commit to kg", ] class GenesisTriageAgent: """ Routes any task to the right general + model tier. Pure keyword classification — no LLM call, no network I/O. Target: < 2 seconds (typically < 1 ms). """ def __init__(self) -> None: self.version = "1.0.0" # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def triage(self, task: str) -> dict[str, Any]: """ Classify a task string and return full routing metadata. Parameters ---------- task : str Any free-text task description. Returns ------- dict with keys: general, complexity, model_tier, priority, suggested_model, estimated_cost_usd, parallel_safe, requires_browser, requires_memory_write, triage_latency_ms, matched_keywords """ t0 = time.perf_counter() task_lower = task.lower() words = re.findall(r"\b\w+\b", task_lower) word_count = len(words) general, matched_general_keywords = self._detect_general(task_lower) complexity = self._detect_complexity(word_count) priority = self._detect_priority(task_lower) model_tier = TIER_BY_COMPLEXITY[complexity] suggested_model = MODEL_SELECTION.get((general, complexity), "google/gemini-2.5-flash") estimated_cost = COST_BY_MODEL.get(suggested_model, 0.005) requires_browser = self._detect_browser(task_lower) requires_memory_write = self._detect_memory_write(task_lower) parallel_safe = not requires_browser # browser tasks serialise themselves latency_ms = round((time.perf_counter() - t0) * 1000, 3) return { "general": general, "complexity": complexity, "model_tier": model_tier, "priority": priority, "suggested_model": suggested_model, "estimated_cost_usd": estimated_cost, "parallel_safe": parallel_safe, "requires_browser": requires_browser, "requires_memory_write": requires_memory_write, "word_count": word_count, "triage_latency_ms": latency_ms, "matched_keywords": matched_general_keywords, } def triage_batch(self, tasks: list[str]) -> list[dict[str, Any]]: """Triage a list of tasks and return a list of routing dicts.""" return [self.triage(t) for t in tasks] def explain(self, task: str) -> str: """Human-readable explanation of triage decision.""" result = self.triage(task) lines = [ f"Task : {task[:100]}{'...' if len(task) > 100 else ''}", f"General : {result['general']}", f"Complexity: {result['complexity']} ({result['word_count']} words)", f"Tier : {result['model_tier']}", f"Model : {result['suggested_model']}", f"Priority : {result['priority']}", f"Est. Cost: ${result['estimated_cost_usd']:.4f}", f"Browser : {result['requires_browser']}", f"Mem Write: {result['requires_memory_write']}", f"Parallel : {result['parallel_safe']}", f"Latency : {result['triage_latency_ms']} ms", f"Keywords : {', '.join(result['matched_keywords']) or 'none — default FORGE'}", ] return "\n".join(lines) # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ def _detect_general(self, task_lower: str) -> tuple[str, list[str]]: """Return (general_name, matched_keywords). FORGE is the default.""" for general, keywords in GENERAL_KEYWORDS: matched = [kw for kw in keywords if kw in task_lower] if matched: return general, matched # Default: if no keyword matched, lean towards FORGE (build) return "FORGE", [] def _detect_complexity(self, word_count: int) -> str: for label, low, high in COMPLEXITY_TIERS: if low <= word_count < high: return label return "complex" def _detect_priority(self, task_lower: str) -> str: for level in ("critical", "high", "low"): for kw in PRIORITY_KEYWORDS.get(level, []): if kw in task_lower: return level return "medium" def _detect_browser(self, task_lower: str) -> bool: return any(kw in task_lower for kw in BROWSER_KEYWORDS) def _detect_memory_write(self, task_lower: str) -> bool: return any(kw in task_lower for kw in MEMORY_WRITE_KEYWORDS) # --------------------------------------------------------------------------- # CLI self-test # --------------------------------------------------------------------------- if __name__ == "__main__": agent = GenesisTriageAgent() TEST_TASKS = [ "Build the Stripe checkout flow for ReceptionistAI", "Research top 20 AU digital agencies in Brisbane", "Set up GHL workflow for George Bunker demo", "Ingest yesterday's session transcript into the knowledge graph", "Evaluate Gemini 2.5 Flash vs MiniMax for coding tasks", "Write the landing page copy for the AI voice widget", "Fix the broken test in core/triage_agent.py", "Find competitor pricing for AI receptionist products", "Login to GoDaddy and update DNS records for sunaiva.com", "Build the voice widget checkout flow", ] print("=" * 70) print("GENESIS TRIAGE AGENT — Self Test") print("=" * 70) for task in TEST_TASKS: result = agent.triage(task) print(f"\n[{result['general']:10}] [{result['complexity']:8}] [T{result['model_tier']}] " f"${result['estimated_cost_usd']:.4f} {task[:60]}") print(f" Model: {result['suggested_model']} " f"Priority: {result['priority']} " f"Browser: {result['requires_browser']} " f"Latency: {result['triage_latency_ms']}ms") print("\n" + "=" * 70) print("DETAILED EXPLAIN — first task:") print(agent.explain(TEST_TASKS[0]))