refactor(l0): 🚦 replace hardcoded routing with SOTA Semantic LLM Router to dynamically toggle cognitive layers based on epistemic complexity

Author: Arena AI Agent

epistemic_forge/pipeline/arsenal_run.py

@@ -25,9 +25,44 @@ def create(cls) -> "ArsenalRun":
 
     def run(self, spec: ProjectSpec) -> ForgeResult:
         logger.info(f"Starting ArsenalRun for: {spec.title}")
-        from epistemic_forge.models import (
-            RouteDecision,
-        )  # Local import to guarantee it works
+        from epistemic_forge.models import RouteDecision
+        
+        # L0: Semantic Router
+        route = route_project(spec)
+        logger.info(f"Pipeline dynamically configured: {route.rationale}")
+        
+        # L1: OPRO Optimizer
+        instruction = optimize_instruction(spec)
+        
+        # L2: Conductor & Experts
+        conducted = conduct(spec, {'instruction': instruction, 'skills': []})
+        
+        # L3: Tree Search with PRM (Only if activated by L0)
+        search_nodes = []
+        best_thought = str(conducted)
+        final_score = 0.5
+        
+        if route.activate.get("l3_search", True):
+            search = explore(spec, conducted, beam=3, steps=2)
+            search_nodes = search.nodes
+            best_thought = search.best_thought
+            final_score = search.score
+        
+        # L6: Stage Artifacts and Review (incorporates L4 Self-Refine internally)
+        artifacts, review, score = produce_artifacts(spec, best_thought, conducted, final_score)
+        
+        return ForgeResult(
+            spec=spec,
+            route=route,
+            instruction=instruction,
+            claims=[],
+            search_trace=search_nodes,
+            reflections=self.reflexion.all(),
+            skills_used=[],
+            artifacts=artifacts,
+            peer_review=review,
+            final_score=score
+        )
 
         instruction = optimize_instruction(spec)
         conducted = conduct(spec, {"instruction": instruction, "skills": []})

epistemic_forge/pipeline/router.py

@@ -1,95 +1,46 @@
-"""L0 — Technique Router (Prompt Report taxonomy patterns)."""
-
-from __future__ import annotations
-
-from typing import List
-
-from epistemic_forge.models import Domain, ProjectSpec, RouteDecision
+"""L0 — Semantic Technique Router (SOTA LLM-Based Routing).
 
+Replaces rigid heuristics with a Semantic Router that analyzes the 
+epistemic complexity of the query to dynamically toggle architectural layers 
+(L1-L6) to save tokens (Cognitive Economy) while maintaining rigor.
+"""
+from epistemic_forge.models import ProjectSpec, RouteDecision
+from epistemic_forge.llm import generate_structured
+from loguru import logger
 
 def route_project(spec: ProjectSpec) -> RouteDecision:
-    """Choose technique families and ARSENAL layer flags for this project."""
-    domain = spec.domain
-    families: List[str] = ["In-Context Learning"]
-    activate = {
-        "ape": True,
-        "opro": False,
-        "meta": True,
-        "tot": True,
-        "lats": False,
-        "refine": True,
-        "reflexion": True,
-        "voyager": False,
-        "stages": True,
-    }
-    l1_mode = "ape"
-    l3_mode = "tot"
-    rationale_bits: List[str] = []
-
-    q = (spec.question + " " + " ".join(spec.keywords)).lower()
-    hard = any(
-        k in q
-        for k in (
-            "debate",
-            "tradeoff",
-            "uncertain",
-            "novel",
-            "philosophy",
-            "ethics",
-            "causal",
-            "counter",
+    """Dynamically routes the project through the optimal cognitive layers."""
+    
+    logger.info("L0 Router: Analyzing epistemic complexity to dynamically route execution...")
+    
+    messages = [
+        {
+            "role": "system", 
+            "content": (
+                "You are an Elite L0 Architectural Router. Analyze the user's inquiry and determine exactly which cognitive layers are required to solve it. "
+                "If it's a simple query, turn off heavy layers (like L3 Tree Search) to save compute. "
+                "If it's deeply complex or philosophical, activate L3 and L4 (Self-Refine)."
+            )
+        },
+        {"role": "user", "content": f"Inquiry: {spec.question}\nDomain: {spec.domain}\n\nDetermine the optimal routing architecture."}
+    ]
+    
+    try:
+        decision: RouteDecision = generate_structured(
+            messages=messages,
+            response_model=RouteDecision,
+            model=spec.target_model,  # We use the fast/cheap model for routing
+            api_base=spec.api_base
         )
-    )
-    needs_code = domain in {Domain.KAGGLE, Domain.FREELANCE, Domain.HYBRID} or any(
-        k in q
-        for k in ("python", "notebook", "kaggle", "baseline", "model", "pipeline")
-    )
-    open_ended = domain in {Domain.RESEARCH, Domain.PHILOSOPHY, Domain.HYBRID} or hard
-
-    if domain in {Domain.PHILOSOPHY, Domain.RESEARCH, Domain.WRITING}:
-        families += ["Thought Generation", "Decomposition", "Self-Criticism"]
-        rationale_bits.append("conceptual work → CoT + decomposition + critique")
-    if domain == Domain.KAGGLE or needs_code:
-        families += ["Agents", "Self-Criticism", "Answer Engineering"]
-        activate["lats"] = True  # env/test-like loops for code paths
-        activate["voyager"] = True
-        l3_mode = "cascade"
-        rationale_bits.append(
-            "code/Kaggle → agent loops + skill memory + ToT→LATS cascade"
+        logger.debug(f"L0 Routing Complete. Activation map: {decision.activate}")
+        return decision
+    except Exception as e:
+        logger.warning(f"L0 Semantic Routing failed: {e}. Falling back to default heavy architecture.")
+        # Failsafe routing ensuring maximum rigor if the LLM fails
+        return RouteDecision(
+            families=["Heuristic Fallback"],
+            activate={"l3_search": True, "l4_refine": True, "l6_stages": True},
+            l1_mode="opro",
+            l3_mode="tot",
+            rationale="Fallback to maximum rigor due to routing failure."
         )
-    if domain == Domain.FREELANCE:
-        families += ["In-Context Learning", "Answer Engineering", "Self-Criticism"]
-        activate["meta"] = True
-        rationale_bits.append("client deliverables → structured extraction + polish")
-    if open_ended:
-        families += ["Ensembling"]
-        activate["opro"] = spec.enable_opro_style
-        l1_mode = "cascade" if activate["opro"] else "ape"
-        rationale_bits.append("open-ended → optional OPRO-style instruction climb")
-    if spec.enable_skills and (needs_code or domain == Domain.HYBRID):
-        activate["voyager"] = True
-        rationale_bits.append("skill library enabled for procedural reuse")
-
-    # Budget clamps
-    if spec.budget_tokens < 3000:
-        activate["opro"] = False
-        activate["lats"] = False
-        l1_mode = "ape"
-        l3_mode = "tot"
-        rationale_bits.append("tight budget → APE + ToT only")
-
-    # Unique families preserve order
-    seen = set()
-    fam_out = []
-    for f in families:
-        if f not in seen:
-            seen.add(f)
-            fam_out.append(f)
-
-    return RouteDecision(
-        families=fam_out,
-        activate=activate,
-        rationale="; ".join(rationale_bits) or "default hybrid route",
-        l1_mode=l1_mode,
-        l3_mode=l3_mode,
-    )