Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| PRACTICALITY SYSTEM 28.0 — MODULAR ORACLE ENGINE | |
| ══════════════════════════════════════════════════════════════════════════ | |
| Architecture: | |
| - practicality_core.py : Tensor Math, Interval Arithmetic, Adam Optimizer | |
| - practicality_axioms.py : Ray Tracer, UCB1 Bandit, Hypothesis Generation | |
| - practicality_oracle.py : Ground Truth Verification, Heisenberg Quantum Sim | |
| - app.py (This file) : Orchestrator, LLM Bridge, AXL Parser, Gradio UI | |
| """ | |
| import os, time, threading, json, re, textwrap, copy | |
| import traceback, html, itertools, math | |
| import requests # Explicitly declared | |
| from collections import deque, defaultdict | |
| from typing import Dict, List, Tuple, Any, Optional, Set | |
| from dataclasses import dataclass, field | |
| import sympy as sp | |
| from sympy.parsing.sympy_parser import parse_expr | |
| import gradio as gr | |
| from google import genai | |
| from google.genai import types | |
| # Import our decoupled engines | |
| import practicality_core as core | |
| import practicality_axioms as axioms | |
| import practicality_oracle as oracle | |
| DEFAULT_GEMINI_API_KEY = os.environ.get("AI_Key", "") | |
| GEMINI_MODEL = "gemini-2.5-flash" | |
| print(f"[SYSTEM 28.0] Orchestrator Initialized. Compute: {core.DEVICE.type.upper()}") | |
| # ══════════════════════════════════════════════════════════════════════ | |
| # SECTION 1: GLOBAL STATE & UI LOCKS | |
| # ══════════════════════════════════════════════════════════════════════ | |
| STATE_LOCK = threading.Lock() | |
| FINGERPRINT_LOCK = threading.Lock() | |
| FINGERPRINT_STORE: Dict[str, List[Dict]] = defaultdict(list) | |
| GLOBAL_SOLVE_STATE = { | |
| "is_running": False, "phase": "IDLE", "best_ce": float('inf'), | |
| "total_fired": 0, "best_ray": "—", | |
| "recent_traces": deque(maxlen=25), "high_level_logs": [], | |
| "answer": "", "raw_results_json": "", "infeasibility_report": "", | |
| "minimize_result": "", | |
| } | |
| def _update_state(**kwargs): | |
| with STATE_LOCK: | |
| for k, v in kwargs.items(): GLOBAL_SOLVE_STATE[k] = v | |
| def _add_log(line: str): | |
| with STATE_LOCK: GLOBAL_SOLVE_STATE["high_level_logs"].append(line) | |
| print(f"[LOG] {line}") | |
| # ══════════════════════════════════════════════════════════════════════ | |
| # SECTION 2: AXL / PSL PARSER (The Bridge) | |
| # ══════════════════════════════════════════════════════════════════════ | |
| class AXLInvariant: | |
| name: str; expr: str; tolerance: float = 1e-3; mode: str = "eq" | |
| compiled_func: Any = None; syms_used: List[str] = field(default_factory=list) | |
| def compile(self, variables: List[str]): | |
| self.expr = self.expr.replace("^", "**") | |
| syms = {v: sp.Symbol(v) for v in variables} | |
| parsed = parse_expr(self.expr, local_dict=syms) | |
| if getattr(parsed, 'is_Equality', False) or getattr(parsed, 'is_Relational', False): | |
| parsed = parsed.lhs - parsed.rhs | |
| for s in list(parsed.free_symbols): | |
| if str(s) not in variables: parsed = parsed.subs(s, 1.0) | |
| self.syms_used = [str(s) for s in parsed.free_symbols if str(s) in variables] | |
| self.compiled_func = sp.lambdify([sp.Symbol(v) for v in self.syms_used], parsed, modules="math") | |
| class AXLProblemDef: | |
| name: str; description: str; axioms: Set[str]; variables: List[Dict]; constraints: List[Dict] | |
| scopes: List[Dict]; anchors: List[AXLInvariant]; observations: Dict[str, float] = field(default_factory=dict) | |
| minimize_var: str = "" | |
| class AXLtoPSLCompiler: | |
| def compile(self, prob: AXLProblemDef) -> str: | |
| lines = [f"SPACE {prob.name}"]; emitted_vars = set() | |
| for v in prob.variables: | |
| lo, hi = v["lo"], v["hi"] | |
| if v["name"] in prob.observations: val = prob.observations[v["name"]]; lo = hi = val | |
| is_int_flag = " INT" if v.get("is_int", False) else "" | |
| lines.append(f"VAR {v['name']} {lo} {hi}{is_int_flag}") | |
| emitted_vars.add(v["name"]) | |
| for obs_name, obs_val in prob.observations.items(): | |
| if obs_name not in emitted_vars: | |
| is_int_flag = " INT" if float(obs_val).is_integer() else "" | |
| lines.append(f"VAR {obs_name} {obs_val} {obs_val}{is_int_flag}") | |
| emitted_vars.add(obs_name) | |
| for c in prob.constraints: | |
| wt = c.get("weight", 1.0); expr = str(c['expr']).replace("^", "**") | |
| lines.append(f"{c['kind']} {expr} WEIGHT {wt}") | |
| return "\n".join(lines) | |
| AXL_COMPILER = AXLtoPSLCompiler() | |
| def parse_template_as_axl(template_text: str) -> AXLProblemDef: | |
| lines = template_text.strip().split('\n') | |
| variables, constraints, anchors, observations = [], [], [], {} | |
| minimize_var, name = "", "template" | |
| for line in lines[:6]: | |
| if "HYPOTHESIS" in line.upper(): | |
| name = re.sub(r'[^a-zA-Z0-9_]', '_', line.split(":", 1)[-1].strip())[:28] | |
| break | |
| for line in lines: | |
| stripped = line.strip() | |
| if not stripped or stripped.startswith('#'): continue | |
| if stripped.upper().startswith("MINIMIZE:"): | |
| minimize_var = stripped.split(":", 1)[1].strip() | |
| continue | |
| var_match = re.match(r'^([a-zA-Z_0-9,\s]+)\s+in\s+\[([^\]]+)\]\s*(INT)?', stripped, re.IGNORECASE) | |
| if var_match: | |
| vnames = [v.strip() for v in var_match.group(1).split(',')] | |
| parts = [p.strip() for p in var_match.group(2).split(',')] | |
| lo, hi = float(parts[0]), float(parts[1]) | |
| is_int = bool(var_match.group(3)) | |
| for vn in vnames: | |
| if vn: variables.append({"name": vn, "lo": lo, "hi": hi, "is_int": is_int}) | |
| continue | |
| con_match = re.match(r'^(EQ|GEQ|LEQ)(?:\s+weight\s*=\s*([0-9.]+))?\s*:\s*(.+)$', stripped, re.IGNORECASE) | |
| if con_match: | |
| kind, wt, expr = con_match.group(1).upper(), float(con_match.group(2) or 1.0), con_match.group(3).replace('^', '**') | |
| constraints.append({"kind": kind, "weight": wt, "expr": expr}) | |
| continue | |
| anc_match = re.match(r'^Anchor:\s*(.+)$', stripped, re.IGNORECASE) | |
| if anc_match: | |
| anchors.append(AXLInvariant(name=f"anc_{len(anchors)}", expr=anc_match.group(1).replace('^', '**'))) | |
| continue | |
| obs_match = re.match(r'^Observation:\s*([a-zA-Z_0-9]+)\s*=\s*([0-9.e+\-]+)', stripped, re.IGNORECASE) | |
| if obs_match: | |
| observations[obs_match.group(1)] = float(obs_match.group(2)) | |
| return AXLProblemDef( | |
| name=name, description=template_text[:100], axioms=set(), | |
| variables=variables, constraints=constraints, scopes=[], | |
| anchors=anchors, observations=observations, minimize_var=minimize_var | |
| ) | |
| def build_core_problem(axl_def: AXLProblemDef) -> core.Problem: | |
| psl_text = AXL_COMPILER.compile(axl_def) | |
| bounds, int_vars = {}, set() | |
| for v in axl_def.variables: | |
| bounds[v['name']] = (v['lo'], v['hi']) | |
| if v.get('is_int'): int_vars.add(v['name']) | |
| for obs_k, obs_v in axl_def.observations.items(): | |
| bounds[obs_k] = (obs_v, obs_v) | |
| var_names = list(bounds.keys()) | |
| compiled_constraints = [] | |
| kind_map = {'EQ': 'equality', 'GEQ': 'inequality', 'LEQ': 'inequality'} | |
| dir_map = {'EQ': 'eq', 'GEQ': 'geq', 'LEQ': 'leq'} | |
| for c in axl_def.constraints: | |
| k = c['kind'].upper() | |
| mc = core.compile_mc(kind_map[k], c['expr'], dir_map[k], var_names, c.get('weight', 1.0)) | |
| compiled_constraints.append(mc) | |
| prob = core.Problem( | |
| pid=axl_def.name, variables=var_names, bounds=bounds, | |
| compiled_constraints=compiled_constraints, int_vars=int_vars, | |
| minimize_var=axl_def.minimize_var | |
| ) | |
| return prob | |
| # ══════════════════════════════════════════════════════════════════════ | |
| # SECTION 3: EXTERNAL / INTERNAL LLM BRIDGE & ROBUST JSON EXTRACTION | |
| # ══════════════════════════════════════════════════════════════════════ | |
| def _extract_json_from_text(text: str) -> str: | |
| """Robust, lexer-style JSON extractor that handles LaTeX groupings.""" | |
| text = text.strip() | |
| if not text: | |
| return "" | |
| blocks = re.findall(r'```(?:json)?\s*(\{.*?\})\s*```', text, re.DOTALL | re.IGNORECASE) | |
| for b in blocks: | |
| cleaned = b.strip() | |
| if any(k in cleaned for k in ['"variables"', '"constraints"', '"hypothesis_markdown"', '"name"']): | |
| return cleaned | |
| if blocks: | |
| return blocks[0].strip() | |
| candidates = [] | |
| for match in re.finditer(r'\{', text): | |
| start_idx = match.start() | |
| depth = 0 | |
| for i in range(start_idx, len(text)): | |
| if text[i] == '{': | |
| depth += 1 | |
| elif text[i] == '}': | |
| depth -= 1 | |
| if depth == 0: | |
| candidates.append(text[start_idx:i+1]) | |
| break | |
| for cand in candidates: | |
| cleaned = cand.strip() | |
| if any(k in cleaned for k in ['"variables"', '"constraints"', '"hypothesis_markdown"', '"name"']): | |
| return cleaned | |
| if candidates: | |
| return max(candidates, key=len).strip() | |
| return text | |
| def call_external_ai(url: str, model: str, system_prompt: str, prompt: str) -> str: | |
| url = url.rstrip("/") | |
| try: | |
| res = requests.post(f"{url}/api/chats", timeout=10) | |
| res.raise_for_status() | |
| chat_id = res.json()["id"] | |
| payload = {"model": model, "prompt": prompt, "system_prompt": system_prompt} | |
| stream_res = requests.post(f"{url}/api/chats/{chat_id}/stream", json=payload, stream=True, timeout=300) | |
| stream_res.raise_for_status() | |
| for _ in stream_res.iter_content(chunk_size=8192): pass | |
| chat_res = requests.get(f"{url}/api/chats/{chat_id}", timeout=10) | |
| chat_res.raise_for_status() | |
| chat_data = chat_res.json() | |
| try: requests.delete(f"{url}/api/chats/{chat_id}", timeout=10) | |
| except: pass | |
| messages = chat_data.get("messages", []) | |
| if not messages: raise ValueError("No messages returned from external AI") | |
| last_msg = messages[-1] | |
| if last_msg.get("role") != "assistant": raise ValueError("Last message is not from assistant") | |
| return last_msg.get("content", "") | |
| except Exception as e: | |
| raise RuntimeError(f"External AI Call Failed: {str(e)}") | |
| def encode_problem(text: str, api_key: str, use_ext: bool, ext_url: str, ext_model: str) -> AXLProblemDef: | |
| if use_ext: | |
| _add_log(f"Calling External AI Encoder ({ext_model})...") | |
| raw = call_external_ai(ext_url, ext_model, oracle.ENCODER_PROMPT, text) | |
| else: | |
| _add_log("Calling Internal Gemini Encoder...") | |
| client = genai.Client(api_key=api_key) | |
| config = types.GenerateContentConfig( | |
| system_instruction=[types.Part.from_text(text=oracle.ENCODER_PROMPT)], | |
| response_mime_type="application/json", temperature=0.1) | |
| resp = client.models.generate_content( | |
| model=GEMINI_MODEL, | |
| contents=[types.Content(role="user", parts=[types.Part.from_text(text=text)])], | |
| config=config) | |
| raw = resp.text | |
| raw = raw.strip() | |
| if not raw: | |
| raise ValueError("The AI model returned an empty response.") | |
| json_str = _extract_json_from_text(raw) | |
| if not json_str: | |
| raise ValueError(f"Could not extract JSON from AI response.\nRaw output: {raw[:200]}...") | |
| try: | |
| d = json.loads(json_str) | |
| except json.JSONDecodeError as e: | |
| raise ValueError(f"JSON Parse Error: {e}\nExtracted String: {json_str[:200]}") | |
| if "error" in d: raise ValueError(f"Isomorphism Invalid: {d['error']} - {d.get('reason', '')}") | |
| anchors = [AXLInvariant(name=a["name"], expr=a["expr"]) for a in d.get("anchors", [])] | |
| variables = [{"name": v["name"], "lo": float(v.get("lo", -1000.0)), "hi": float(v.get("hi", 1000.0)), | |
| "is_int": bool(v.get("is_int", False))} for v in d["variables"]] | |
| return AXLProblemDef( | |
| name=d["name"], description=d.get("description", ""), axioms=set(), | |
| variables=variables, constraints=d.get("constraints", []), | |
| scopes=[], anchors=anchors, observations=d.get("observations", {}) | |
| ) | |
| def collapse_solution(payload: dict, api_key: str, use_ext: bool, ext_url: str, ext_model: str) -> str: | |
| if use_ext: | |
| _add_log(f"Calling External AI Collapser ({ext_model})...") | |
| raw = call_external_ai(ext_url, ext_model, oracle.COLLAPSER_PROMPT, json.dumps(payload)) | |
| else: | |
| _add_log("Calling Internal Gemini Collapser...") | |
| client = genai.Client(api_key=api_key) | |
| config = types.GenerateContentConfig( | |
| system_instruction=[types.Part.from_text(text=oracle.COLLAPSER_PROMPT)], | |
| temperature=0.2) | |
| resp = client.models.generate_content( | |
| model=GEMINI_MODEL, | |
| contents=[types.Content(role="user", parts=[types.Part.from_text(text=json.dumps(payload))])], | |
| config=config) | |
| raw = resp.text | |
| raw = raw.strip() | |
| if not raw: | |
| return "The AI model returned an empty response during collapse." | |
| json_str = _extract_json_from_text(raw) | |
| try: | |
| data = json.loads(json_str) | |
| return data.get("hypothesis_markdown", "No markdown generated.") | |
| except Exception as e: | |
| return f"JSON Decode Error from Collapser: {e}\n\nRaw Output:\n{raw}" | |
| # ══════════════════════════════════════════════════════════════════════ | |
| # SECTION 4: THE ORCHESTRATOR LOOP (Adversary Pre-Flight Removed) | |
| # ══════════════════════════════════════════════════════════════════════ | |
| def global_solve_task(original_text: str, api_key: str, template_name: str = None, | |
| use_ext: bool = False, ext_url: str = "", ext_model: str = ""): | |
| _update_state( | |
| is_running=True, phase="ENCODING", high_level_logs=[], answer="", | |
| best_ce=float('inf'), total_fired=0, best_ray="—", | |
| recent_traces=deque(maxlen=25), raw_results_json="", | |
| infeasibility_report="", minimize_result="" | |
| ) | |
| try: | |
| if template_name: | |
| _add_log(f"Loading Template: {template_name}") | |
| template_text = oracle.STRUCTURAL_HYPOTHESIS_TEMPLATES[template_name] | |
| axl_def = parse_template_as_axl(template_text) | |
| else: | |
| axl_def = encode_problem(original_text, api_key, use_ext, ext_url, ext_model) | |
| _add_log(f"`{axl_def.name}` | {len(axl_def.variables)} vars") | |
| if axl_def.minimize_var: _add_log(f" MINIMIZE: {axl_def.minimize_var}") | |
| is_factorization = any(v['name'] in ['p', 'q', 't', 's', 'd'] for v in axl_def.variables) | |
| _update_state(phase="RAY TRACING") | |
| base_prob = build_core_problem(axl_def) | |
| init_b, init_ce = axioms._mprt_sample(base_prob, 1000) | |
| _add_log("Searching continuous relaxation...") | |
| templates = axioms.generate_templates(base_prob, axioms.compute_sketch(base_prob)) | |
| tracer = axioms.SequenceRayTracer( | |
| log_callback=_add_log, | |
| update_state_callback=lambda **kwargs: _update_state(**kwargs) | |
| ) | |
| binding, ce, traces, best_ray_name = tracer.trace( | |
| base_prob, init_b, init_ce, templates, max_rays=1500 | |
| ) | |
| _add_log(f"Deduction Engine Complete. Best CE = {ce:.6f}") | |
| g1_pass = ce < core.SOLVE_THRESHOLD | |
| g2_pass = True | |
| if is_factorization and len(axl_def.observations) > 0: | |
| N_target = int(list(axl_def.observations.values())[0]) | |
| # Call ground_truth_verify via the oracle module import | |
| gt_passed, gt_msg = oracle.ground_truth_verify(binding, N_target) | |
| _add_log(f"GROUND TRUTH: {gt_msg}") | |
| if not gt_passed: | |
| _update_state(infeasibility_report=f"GROUND TRUTH FAILED: {gt_msg}") | |
| g2_pass = False | |
| _update_state(phase="COLLAPSING", best_ce=ce) | |
| payload = { | |
| "solved": g1_pass and g2_pass, | |
| "raw_binding": {k: core.safe_round(v, 4) for k, v in binding.items()}, | |
| "infeasibility_report": GLOBAL_SOLVE_STATE["infeasibility_report"] | |
| } | |
| ans = collapse_solution(payload, api_key, use_ext, ext_url, ext_model) | |
| _update_state(answer=ans, raw_results_json=json.dumps(payload, indent=2), phase="DONE") | |
| _add_log("Pipeline Complete.") | |
| except Exception as e: | |
| tb_str = traceback.format_exc() | |
| print(tb_str) | |
| error_msg = f"💥 FATAL ERROR ({type(e).__name__}): {str(e)}" | |
| _add_log(error_msg) | |
| md_error = f"## 💥 SYSTEM CRASH\n\n**Error Type:** `{type(e).__name__}`\n**Message:** {str(e)}\n\n### Traceback\n```python\n{tb_str}\n```" | |
| _update_state(phase="ERROR", answer=md_error) | |
| finally: | |
| _update_state(is_running=False) | |
| def trigger_solve(problem_text, api_key, template_name=None, use_ext=False, ext_url="", ext_model=""): | |
| if GLOBAL_SOLVE_STATE["is_running"] or not problem_text.strip(): return | |
| threading.Thread(target=global_solve_task, args=(problem_text, api_key, template_name, use_ext, ext_url, ext_model), daemon=True).start() | |
| def trigger_hyp_solve(template_name, api_key, use_ext=False, ext_url="", ext_model=""): | |
| if GLOBAL_SOLVE_STATE["is_running"]: return | |
| threading.Thread(target=global_solve_task, args=("", api_key, template_name, use_ext, ext_url, ext_model), daemon=True).start() | |
| def run_quantum_demo(): | |
| _update_state(is_running=True, phase="QUANTUM SIMULATION", answer="Running Physics Simulator...\nCheck console for output.", high_level_logs=[]) | |
| def _run(): | |
| try: | |
| oracle.run_shors_demo(15) | |
| oracle.run_shors_demo(35) | |
| oracle.benchmark_heisenberg_vs_schrodinger(20) | |
| _update_state(phase="DONE", answer="Quantum simulation complete. Output logged to console.") | |
| except Exception as e: | |
| tb_str = traceback.format_exc() | |
| print(tb_str) | |
| _add_log(f"💥 QUANTUM DEMO CRASH: {str(e)}") | |
| _update_state(phase="ERROR", answer=f"## 💥 QUANTUM CRASH\n\n```python\n{tb_str}\n```") | |
| finally: | |
| _update_state(is_running=False) | |
| threading.Thread(target=_run, daemon=True).start() | |
| # ══════════════════════════════════════════════════════════════════════ | |
| # SECTION 5: GRADIO UI | |
| # ══════════════════════════════════════════════════════════════════════ | |
| CSS = """ | |
| body{background:#090909;color:#e0e0e0;font-family:monospace;} | |
| .solver-log{font-size:0.75em;background:#0d0d0d;border:1px solid #181818;} | |
| .answer-box{font-size:0.85em;background:#0d0d0d;border:1px solid #222;padding:10px;} | |
| .infeas-banner{background:#1a0000;border:1px solid #5C0000;color:#FF6B6B;padding:8px;border-radius:4px;font-size:0.8em;margin:4px 0;} | |
| """ | |
| def build_live_html(): | |
| with STATE_LOCK: | |
| phase = GLOBAL_SOLVE_STATE["phase"] | |
| ce = GLOBAL_SOLVE_STATE["best_ce"] | |
| logs = GLOBAL_SOLVE_STATE["high_level_logs"] | |
| infeas = GLOBAL_SOLVE_STATE["infeasibility_report"] | |
| phase_colors = {"DONE": "#4CAF50", "ERROR": "#F44336", "ADVERSARY PRE-FLIGHT": "#FF9800"} | |
| phase_color = phase_colors.get(phase, "#26C6DA" if phase != "IDLE" else "#888") | |
| log_html = "<ul style='margin:0;padding-left:20px;color:#aaa;font-size:0.82em'>" | |
| for l in logs[-15:]: | |
| safe_l = html.escape(str(l)).replace('\n', '<br>') | |
| if "FATAL ERROR" in safe_l or "CRASH" in safe_l: | |
| log_html += f"<li style='margin-bottom:4px;color:#FF6B6B;font-weight:bold;'>{safe_l}</li>" | |
| else: | |
| log_html += f"<li style='margin-bottom:4px'>{safe_l}</li>" | |
| log_html += "</ul>" | |
| infeas_html = f"<div class='infeas-banner'>🚫 {html.escape(infeas[:300])}</div>" if infeas else "" | |
| return (f"<div style='border:1px solid #222;border-radius:6px;padding:12px;background:#0d0d0d;min-height:300px;'>" | |
| f"<div style='color:{phase_color};font-weight:bold;margin-bottom:8px'>● Phase: {phase}</div>" | |
| f"<div style='margin-bottom:8px;border-bottom:1px solid #222;padding-bottom:8px'>{log_html}</div>" | |
| f"{infeas_html}" | |
| f"</div>") | |
| def fetch_dashboard(d_cache): | |
| new_html = build_live_html() | |
| if d_cache.get("html") == new_html: return gr.skip(), d_cache | |
| d_cache["html"] = new_html | |
| return new_html, d_cache | |
| def fetch_answers(a_cache): | |
| if a_cache.get("finalized", False): return gr.skip(), gr.skip(), a_cache | |
| with STATE_LOCK: | |
| ans, raw, phase = GLOBAL_SOLVE_STATE["answer"], GLOBAL_SOLVE_STATE["raw_results_json"], GLOBAL_SOLVE_STATE["phase"] | |
| ans_changed = a_cache.get("ans") != ans | |
| raw_changed = a_cache.get("raw") != raw | |
| if not ans_changed and not raw_changed: return gr.skip(), gr.skip(), a_cache | |
| a_cache.update({"ans": ans, "raw": raw}) | |
| if phase in ("DONE", "ERROR"): a_cache["finalized"] = True | |
| return (ans if ans_changed else gr.skip(), raw if raw_changed else gr.skip(), a_cache) | |
| def _reset(c): | |
| c.update({"finalized": False, "ans": "", "raw": ""}) | |
| return c | |
| def handle_solve(text, key, use_ext, ext_url, ext_model, cache): | |
| trigger_solve(text, key, template_name=None, use_ext=use_ext, ext_url=ext_url, ext_model=ext_model) | |
| return _reset(cache) | |
| def handle_hyp_solve(name, key, use_ext, ext_url, ext_model, cache): | |
| trigger_hyp_solve(name, key, use_ext=use_ext, ext_url=ext_url, ext_model=ext_model) | |
| return _reset(cache) | |
| def handle_q_run(cache): | |
| run_quantum_demo() | |
| return _reset(cache) | |
| with gr.Blocks(title="Practicality 28.0") as demo: | |
| gr.Markdown("## ⚗ Practicality 28.0 — Modular Oracle Engine\n`[Core Math] → [Axioms] → [Ground Truth Oracle]`") | |
| dash_cache = gr.State({"html": ""}) | |
| ans_cache = gr.State({"ans": "", "raw": "", "finalized": False}) | |
| with gr.Row(): | |
| with gr.Column(scale=2): | |
| with gr.Group(): | |
| gr.Markdown("### ⚙️ AI Backend Configuration") | |
| api_key_input = gr.Textbox(label="Gemini API Key (Internal)", type="password", value=DEFAULT_GEMINI_API_KEY) | |
| use_ext_toggle = gr.Checkbox(label="🌐 Use External AI Server", value=False) | |
| ext_url_input = gr.Textbox(label="External Server URL", value="http://localhost:7860", placeholder="http://localhost:7860") | |
| ext_model_dropdown = gr.Dropdown(choices=["claude", "haiku", "maverick"], value="claude", label="External Model") | |
| gr.Markdown("### 1. Novel Discovery") | |
| prob_input = gr.Textbox(label="Target Description", lines=3) | |
| solve_btn = gr.Button("⚗ Solve Natural Language", variant="primary") | |
| gr.Markdown("### 2. Verified Templates") | |
| hyp_dropdown = gr.Dropdown(choices=list(oracle.STRUCTURAL_HYPOTHESIS_TEMPLATES.keys()), value=list(oracle.STRUCTURAL_HYPOTHESIS_TEMPLATES.keys())[0]) | |
| hyp_solve_btn = gr.Button("📋 Run Template directly against Oracle", variant="secondary") | |
| gr.Markdown("### 3. Heisenberg Stabilizer Simulation") | |
| q_btn = gr.Button("⚛ Run Quantum Physics Simulator (Check Console)", variant="secondary") | |
| with gr.Column(scale=2): | |
| live_html = gr.HTML(value=build_live_html()) | |
| ans_box = gr.Textbox(label="Grounded Hypothesis (LLM Output)", lines=12) | |
| raw_box = gr.Textbox(label="Raw JSON", lines=4) | |
| # NATIVE PY-SIDE TIMER | |
| ui_timer = gr.Timer(1.0) | |
| ui_timer.tick(fn=fetch_dashboard, inputs=[dash_cache], outputs=[live_html, dash_cache]) | |
| ui_timer.tick(fn=fetch_answers, inputs=[ans_cache], outputs=[ans_box, raw_box, ans_cache]) | |
| solve_btn.click(fn=handle_solve, inputs=[prob_input, api_key_input, use_ext_toggle, ext_url_input, ext_model_dropdown, ans_cache], outputs=[ans_cache]) | |
| hyp_solve_btn.click(fn=handle_hyp_solve, inputs=[hyp_dropdown, api_key_input, use_ext_toggle, ext_url_input, ext_model_dropdown, ans_cache], outputs=[ans_cache]) | |
| q_btn.click(fn=handle_q_run, inputs=[ans_cache], outputs=[ans_cache]) | |
| if __name__ == "__main__": | |
| demo.launch(server_name="0.0.0.0", server_port=7860, share=False, css=CSS) |