Update prettybird_brain.py

prettybird_brain.py

@@ -1,117 +1,242 @@
-import torch
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+"""
+PrettyBird Skull Engine
+- GGUF = mathematical optimization brain (skull)
+- Bodies = interchangeable (text/image/audio/video/3D adapters)
+- Single-file, backend-clean, optimizer-compatible
+"""
+
 import json
 import re
-import os
-import transformers
-from transformers import LogitsProcessor, LogitsProcessorList, AutoModelForCausalLM, AutoTokenizer
-
-CONTROLLED_REASONING_CORE = "You are a helpful assistant with a Controlled Reasoning Core. Please reason step by step."
-
-class InterventionLogitsProcessor(LogitsProcessor):
-    def __init__(self, boost_token_id, boost_value=2.0):
-        self.boost_token_id = boost_token_id
-        self.boost_value = boost_value
-
-    def __call__(self, input_ids: torch.LongTensor, scores: torch.FloatTensor) -> torch.FloatTensor:
-        scores[:, self.boost_token_id] += self.boost_value
-        return scores
-
-class prettybird_bce_basic_brain_mini:
-    def __init__(self, model_path="qwen_merged", device="cuda" if torch.cuda.is_available() else "cpu"):
-        self.device = device
-        print(f"Transformers version: {transformers.__version__}")
-
-        local_path = model_path
-        if not os.path.exists(local_path):
-            if os.path.exists(os.path.join("llama.cpp", model_path)):
-                local_path = os.path.join("llama.cpp", model_path)
-            elif os.path.exists("/content/qwen_merged"):
-                local_path = "/content/qwen_merged"
-        
-        final_path = local_path if os.path.exists(local_path) else "Qwen/Qwen2.5-Math-1.5B-Instruct"
-        print(f"Loading model from {final_path}...")
+import ast
+import numpy as np
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, Tuple
+
+from llama_cpp import Llama
+
+
+# ============================================================
+# 1) SYSTEM PROMPT (FINAL – kilitli)
+# ============================================================
+SYSTEM_PROMPT = """You are a controlled reasoning core operating as a mathematical optimization brain.
+
+You are NOT an autonomous agent.
+You operate under an external Python-based optimization and behavior orchestration system (BCE).
+
+Hard rules:
+- Output MUST be valid JSON.
+- Output MUST contain ONLY JSON.
+- Do NOT reveal chain-of-thought.
+- Use double quotes only.
+- Keep structure deterministic across revisions.
+
+If information is missing, list it in "needs".
+
+JSON CONTRACT:
+{
+  "version": "1.0",
+  "task": "",
+  "assumptions": [],
+  "needs": [],
+  "candidates": [
+    {
+      "id": "c1",
+      "solution": {},
+      "constraints": [
+        {"name": "", "status": "pass|fail|unknown", "note": ""}
+      ],
+      "objective_estimate": {"primary": 0.0, "notes": ""},
+      "rationale_summary": ""
+    }
+  ],
+  "revision_instructions": "If controller feedback arrives, edit only referenced fields and preserve all others exactly."
+}
+"""
+
+
+# ============================================================
+# 2) Güvenli mini-tool (opsiyonel, math destek)
+# ============================================================
+_ALLOWED_AST = {
+    ast.Expression, ast.BinOp, ast.UnaryOp, ast.Constant,
+    ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Pow, ast.Mod,
+    ast.USub, ast.UAdd,
+}
 
+def safe_calc(expr: str) -> Optional[float]:
+    if not re.fullmatch(r"[0-9\.\s\+\-\*\/\(\)]+", expr):
+        return None
+    try:
+        tree = ast.parse(expr, mode="eval")
+        for n in ast.walk(tree):
+            if type(n) not in _ALLOWED_AST:
+                return None
+        return float(eval(compile(tree, "<calc>", "eval"), {"__builtins__": {}}))
+    except Exception:
+        return None
+
+
+# ============================================================
+# 3) Skull (GGUF Math Brain)
+# ============================================================
+@dataclass
+class Skull:
+    gguf_path: str
+    n_ctx: int = 8192
+    n_gpu_layers: int = 0
+    chat_format: str = "chatml"
+    verbose: bool = False
+
+    def __post_init__(self):
+        self.llm = Llama(
+            model_path=self.gguf_path,
+            n_ctx=self.n_ctx,
+            n_gpu_layers=self.n_gpu_layers,
+            chat_format=self.chat_format,
+            verbose=self.verbose,
+        )
+
+    def _parse_json(self, text: str) -> Dict[str, Any]:
+        t = text.strip()
         try:
-            # Attempt Native Load (trust_remote_code=False) first
-            self.tokenizer = AutoTokenizer.from_pretrained(final_path, trust_remote_code=False)
-            self.model = AutoModelForCausalLM.from_pretrained(
-                final_path,
-                device_map=device,
-                trust_remote_code=False,
-                torch_dtype=torch.float16
-            )
-            print("Loaded natively.")
-        except Exception as e:
-            print(f"Native load failed: {e}. Trying remote code...")
-            try:
-                self.tokenizer = AutoTokenizer.from_pretrained(final_path, trust_remote_code=True)
-                self.model = AutoModelForCausalLM.from_pretrained(
-                    final_path,
-                    device_map=device,
-                    trust_remote_code=True,
-                    torch_dtype=torch.float16
-                )
-                print("Loaded with remote code.")
-            except Exception as e2:
-                raise RuntimeError(f"Failed to load model: {e2}")
-        
-        if self.tokenizer.pad_token is None:
-            self.tokenizer.pad_token = self.tokenizer.eos_token
-
-    def math_reward(self, response):
+            return json.loads(t)
+        except json.JSONDecodeError:
+            s, e = t.find("{"), t.rfind("}")
+            if s != -1 and e != -1 and e > s:
+                return json.loads(t[s:e+1])
+            raise
+
+    def think(
+        self,
+        observation: Dict[str, Any],
+        temperature: float = 0.2,
+        top_p: float = 0.9,
+        max_tokens: int = 512,
+    ) -> Dict[str, Any]:
+
+        messages = [
+            {"role": "system", "content": SYSTEM_PROMPT},
+            {"role": "user", "content": json.dumps(observation, ensure_ascii=False)},
+        ]
+
+        resp = self.llm.create_chat_completion(
+            messages=messages,
+            temperature=temperature,
+            top_p=top_p,
+            max_tokens=max_tokens,
+            response_format={"type": "json_object"},
+        )
+
+        content = resp["choices"][0]["message"]["content"]
+        return self._parse_json(content)
+
+
+# ============================================================
+# 4) Objective + Constraint (C)
+# ============================================================
+class ObjectiveEngine:
+    """
+    GGUF çıktısını tekrar değerlendiren deterministik katman.
+    """
+
+    def score(self, result: Dict[str, Any]) -> float:
         score = 0.0
-        if re.search(r"\\boxed\{.*?\}", response):
-            score += 1.0
-        if len(response) > 50:
+
+        # valid JSON already guaranteed
+        cands = result.get("candidates", [])
+        if not cands:
+            return -1e9
+
+        c = cands[0]
+
+        # constraint satisfaction
+        for con in c.get("constraints", []):
+            if con.get("status") == "pass":
+                score += 1.0
+            elif con.get("status") == "fail":
+                score -= 2.0
+
+        # model's own estimate
+        oe = c.get("objective_estimate", {})
+        if isinstance(oe.get("primary"), (int, float)):
+            score += float(oe["primary"])
+
+        # small structure bonus
+        if isinstance(c.get("solution"), dict):
             score += 0.5
+
         return score
 
-    def parameter_editing(self, layer_idx=0, noise_scale=1e-5):
-        print(f"Editing parameters in layer {layer_idx}...")
-        try:
-            with torch.no_grad():
-                if hasattr(self.model, 'model'):
-                    layers = self.model.model.layers
-                else:
-                    layers = self.model.layers
-                weights = layers[layer_idx].self_attn.q_proj.weight
-                noise = torch.randn_like(weights) * noise_scale
-                weights.add_(noise)
-            print("Parameter editing complete.")
-        except Exception as e:
-            print(f"Error editing params: {e}")
-
-    def run_tool(self, tool_name, query):
-        if tool_name == "calculator":
-            try:
-                clean_query = re.sub(r"[^0-9+\-*/(). ]", "", query)
-                if not clean_query.strip(): return "Invalid"
-                return str(eval(clean_query))
-            except:
-                return "Error"
-        return "Unknown"
-
-    def generate_response(self, query, use_tool=False, use_intervention=False):
-        input_text = query
-        if use_tool or "calculate" in query.lower():
-            match = re.search(r"([\d\.\s\+\-\*\/\(\)]+)", query)
-            if match and len(match.group(1).strip()) > 3:
-                res = self.run_tool("calculator", match.group(1))
-                input_text += f"\nTool Result: {res}"
-
-        messages = [{"role": "system", "content": CONTROLLED_REASONING_CORE}, {"role": "user", "content": input_text}]
-        inputs = self.tokenizer.apply_chat_template(messages, tokenize=True, add_generation_prompt=True, return_tensors="pt").to(self.device)
-        
-        logits_processor = LogitsProcessorList()
-        if use_intervention:
-            print("Applying intervention...")
-            ids = self.tokenizer.encode("Therefore", add_special_tokens=False)
-            if ids:
-                logits_processor.append(InterventionLogitsProcessor(ids[0], 5.0))
-
-        outputs = self.model.generate(inputs, max_new_tokens=100, logits_processor=logits_processor, do_sample=True)
-        response = self.tokenizer.decode(outputs[0][len(inputs[0]):], skip_special_tokens=True)
-        
-        print(f"Response: {response}")
-        print(f"Reward: {self.math_reward(response)}")
-        return response
+
+# ============================================================
+# 5) Body (örnek: text body)
+# ============================================================
+class TextBody:
+    def observe(self, text: str) -> Dict[str, Any]:
+        # İleride image/audio/video/3D body'ler aynı fonksiyonu sağlar
+        return {
+            "task": "optimization_request",
+            "body": "text",
+            "input": text,
+        }
+
+
+# ============================================================
+# 6) Orchestrator (brain loop)
+# ============================================================
+class BrainSystem:
+    def __init__(self, skull: Skull, body: Any):
+        self.skull = skull
+        self.body = body
+        self.objective = ObjectiveEngine()
+
+    def run(self, raw_input: Any, rounds: int = 2) -> Dict[str, Any]:
+        obs = self.body.observe(raw_input)
+
+        best = None
+        best_score = -1e18
+
+        for r in range(rounds):
+            result = self.skull.think(obs)
+            score = self.objective.score(result)
+
+            if score > best_score:
+                best = result
+                best_score = score
+
+            # revise loop (hafif)
+            if result.get("needs"):
+                obs["_feedback"] = {
+                    "issue": "missing_data",
+                    "needs": result["needs"],
+                }
+
+        return {
+            "best_score": best_score,
+            "decision": best,
+        }
+
+
+# ============================================================
+# 7) Demo
+# ============================================================
+if __name__ == "__main__":
+    skull = Skull(
+        gguf_path="prettybird_bce_basic_brain_mini_q4_k_m.gguf",
+        n_ctx=8192,
+        n_gpu_layers=0,
+        chat_format="chatml",
+    )
+
+    body = TextBody()
+    brain = BrainSystem(skull, body)
+
+    output = brain.run(
+        "5 işi 2 makineye ata ve makespan minimize et. Süreler: [3,5,2,6,4].",
+        rounds=2,
+    )
+
+    print(json.dumps(output, ensure_ascii=False, indent=2))