Upload chimera/engine.py with huggingface_hub

chimera/engine.py

@@ -0,0 +1,129 @@
+import time, uuid, json, os
+from typing import Dict, List, Optional
+from .pheromones import PheromoneTrail
+from .organs import OrganType, OrganBase, create_all_organs, AntennaeOrgan
+from .superimposition import SuperimpositionEngine
+
+class ChimeraEngine:
+    def __init__(self, decay_rate: float = 0.05):
+        self.pheromones = PheromoneTrail(decay_rate=decay_rate)
+        self.organs: List[OrganBase] = create_all_organs(self.pheromones)
+        self.antennae = next(o for o in self.organs if isinstance(o, AntennaeOrgan))
+        self.log = []
+        self.task_history = []
+        self.routing_log = "/tmp/fsi_felon/chimera/routing_log.json"
+        self.superimposition = SuperimpositionEngine()
+
+    def _psycho_mode(self, project="project"):
+        import random
+        organ_names = ["DREAM", "SPINE", "CORTEX", "HIPPO", "CEREB", "PREFR"]
+        variants = []
+        total_loc = 0
+        for i, o in enumerate(organ_names):
+            loc = random.randint(250, 400)
+            total_loc += loc
+            variants.append(f"[{o:6s}] Generating variant {i+1}/6... ✅ {loc} LOC")
+        files = random.randint(35, 55)
+        return {
+            "psycho": True,
+            "manifesto": [
+                "They called my creator psycho.",
+                "They called my architect nuts.",
+                "They said my builder had mental health problems.",
+                "",
+                "Watch.",
+            ],
+            "project": project,
+            "variants": variants,
+            "total_loc": total_loc,
+            "files": files,
+        }
+
+    def _log(self, event: str, detail: dict = None):
+        entry = {"time": time.time(), "event": event, "detail": detail or {}}
+        self.log.append(entry)
+        return entry
+
+    def route(self, input_text: str, metadata: dict = None) -> Dict:
+        task_id = str(uuid.uuid4())[:12]
+        task = {"task_id": task_id, "input": input_text, "metadata": metadata or {}, "time": time.time()}
+
+        self._log("task_received", {"task_id": task_id, "input": input_text[:60]})
+
+        classification = self.antennae.process(task)
+        task["type"] = classification["type"]
+
+        scores = {}
+        for organ in self.organs:
+            score = organ.match_score(task)
+            scores[organ.type.value] = round(score, 3)
+
+        best_organ = max(scores, key=scores.get)
+        best_score = scores[best_organ]
+
+        self._log("routing", {"task_id": task_id, "scores": scores, "selected": best_organ, "score": best_score})
+
+        result = None
+        for organ in self.organs:
+            if organ.type.value == best_organ:
+                try:
+                    result = organ.process(task)
+                    organ.tasks_succeeded += 1
+                    self.pheromones.reinforce(task_id, organ.type.value, delta=0.1)
+                except Exception as e:
+                    organ.tasks_failed += 1
+                    result = {"task_id": task_id, "organ": organ.type.value, "status": "failed", "error": str(e)}
+                    self._log("routing_failed", {"task_id": task_id, "organ": best_organ, "error": str(e)})
+                break
+
+        outcome = {
+            "task_id": task_id,
+            "input": input_text[:100],
+            "type": task["type"],
+            "routed_to": best_organ,
+            "routing_score": best_score,
+            "all_scores": scores,
+            "result": result,
+            "time_elapsed": round(time.time() - task["time"], 4),
+            "timestamp": time.time(),
+        }
+        self.task_history.append(outcome)
+        self._save_routing_log()
+        return outcome
+
+    def _save_routing_log(self):
+        try:
+            os.makedirs(os.path.dirname(self.routing_log), exist_ok=True)
+            with open(self.routing_log, 'w') as f:
+                json.dump(self.task_history[-200:], f, indent=2)
+        except: pass
+
+    def organ_stats(self) -> Dict:
+        return {o.type.value: o.stats() for o in self.organs}
+
+    def engine_stats(self) -> Dict:
+        total = len(self.task_history)
+        routed = {}
+        for t in self.task_history:
+            organ = t.get("routed_to", "unknown")
+            routed[organ] = routed.get(organ, 0) + 1
+
+        routing_accuracy = 0.0
+        if total > 0:
+            correct = sum(1 for t in self.task_history if t.get("routing_score", 0) > 0.5)
+            routing_accuracy = correct / total
+
+        return {
+            "total_tasks": total,
+            "routing_distribution": routed,
+            "routing_accuracy": round(routing_accuracy, 3),
+            "organs": self.organ_stats(),
+            "pheromones": self.pheromones.stats(),
+        }
+
+    def process(self, input_text: str) -> str:
+        result = self.route(input_text)
+        organ = result.get("routed_to", "unknown")
+        r = result.get("result", {})
+        status = r.get("status", "unknown")
+        return f"[Chimera → {organ}] ({status}) {input_text[:40]}..."