Text Classification
syfox
decision-engine
decision-making
decision-model
structured-prediction
system-one
si-substrate
intent-classification
routing
abstention
uncertainty-quantification
selective-prediction
hebbian
energy-based-model
associative-memory
no-transformer
no-neural-network
no-pattern-matching
moderation
spam-detection
multilingual
| #!/usr/bin/env python3 | |
| # ============================================================================ | |
| # SyFox HTTP bridge v3 — a Jev-compatible System One API over the SI substrate. | |
| # | |
| # Endpoints: | |
| # POST /v1/systemone { "state": str|obj, "questions": {qid: {...}}, | |
| # "lang": "auto"|"<slug>", (optional) | |
| # "options": { ... CLI-equal knobs } (optional) } | |
| # GET /v1/models model dir discovery (meta.json, no fabric load) | |
| # GET /v1/health core version, flags, uptime, deferral count | |
| # GET /v1/deferrals v2.2 active-learning: logged deferrals | |
| # DELETE /v1/deferrals clear the deferral buffer | |
| # | |
| # options (per request, CLI-equal semantics): | |
| # evidence bool M3: machine-auditable evidence JSON | |
| # energy_norm bool M1: measurement gain for corpus-scale fabrics | |
| # salience_gating bool SI selection mode | |
| # miller_window bool SI selection mode (Miller [cap-4,cap] window) | |
| # ngrams "on"|"off" v2.2 trigram lane policy | |
| # | |
| # Server flags mirror the CLI: --lang off|auto|<slug>, --ngrams auto|on|off, | |
| # --energy-norm, --salience-gating, --miller-window, --threads N, | |
| # --log-deferrals FILE. | |
| # | |
| # The Python layer is a TOOL: it speaks HTTP and JSON. Every decision is | |
| # made by the C++ SI substrate core (energy injection -> settle -> resonance | |
| # readout -> honest silence). No transformer, no classifier lives here. | |
| # ============================================================================ | |
| import argparse | |
| import json | |
| import threading | |
| import time | |
| from pathlib import Path | |
| from fastapi import FastAPI, Request | |
| from fastapi.responses import JSONResponse | |
| import uvicorn | |
| from syfox_bridge import SyFoxEngine, core_version, detect_script, _LIB | |
| app = FastAPI(title="SyFox System One API", | |
| version=f"{core_version()}.api3") | |
| STATE = { | |
| "base_model": "model-tickets", | |
| "lang": "off", # off | auto | <slug> | |
| "ngrams": "auto", # auto | on | off (auto = CLI routing policy) | |
| "log_deferrals": None, | |
| "started": time.time(), | |
| } | |
| ENGINES: dict[str, SyFoxEngine] = {} # model_dir -> loaded engine cache | |
| DEFERRALS: list[dict] = [] # in-memory ring (mirrored to file) | |
| GLOBAL_LOCK = threading.Lock() # serializes routing + decide (ngrams is | |
| # a process-global policy in the core) | |
| SERVER_FLAGS: dict[str, bool] = { | |
| "energy_norm": False, "salience_gating": False, | |
| "miller_window": False, "parallel_settle": False, "threads": 0, | |
| } | |
| def _norm_questions(questions) -> tuple[dict | None, JSONResponse | None]: | |
| """Validate + normalize the question schema (tool-side, v0.1-compatible).""" | |
| if not isinstance(questions, dict) or not questions: | |
| return None, JSONResponse(status_code=400, content={ | |
| "error": {"message": "questions must be a non-empty JSON object"}}) | |
| norm = {} | |
| for qid, q in questions.items(): | |
| if not isinstance(q, dict) or "type" not in q: | |
| return None, JSONResponse(status_code=400, content={ | |
| "error": {"message": f"question '{qid}' needs a type: choice|score|noul"}}) | |
| t = q["type"] | |
| if t not in ("choice", "score", "noul"): | |
| return None, JSONResponse(status_code=400, content={ | |
| "error": {"message": f"question '{qid}': unknown type '{t}'"}}) | |
| nq = {"type": t, "instructions": str(q.get("instructions", ""))} | |
| if t in ("choice", "score"): | |
| if "criteria" not in q: | |
| return None, JSONResponse(status_code=400, content={ | |
| "error": {"message": f"question '{qid}' needs criteria"}}) | |
| nq["criteria"] = q["criteria"] | |
| norm[qid] = nq | |
| return norm, None | |
| def _get_engine(model_dir: str) -> SyFoxEngine: | |
| eng = ENGINES.get(model_dir) | |
| if eng is None: | |
| eng = SyFoxEngine(model_dir) | |
| eng.set_energy_norm(SERVER_FLAGS["energy_norm"]) | |
| eng.set_source_modes(SERVER_FLAGS["salience_gating"], | |
| SERVER_FLAGS["miller_window"]) | |
| eng.set_parallel_settle(SERVER_FLAGS["parallel_settle"], | |
| SERVER_FLAGS["threads"]) | |
| ENGINES[model_dir] = eng | |
| return eng | |
| def _route(state: str, lang_req: str | None) -> tuple[str, str]: | |
| """CLI-equal routing: returns (model_dir, note). | |
| off -> base model, no detection (v2.1 behavior) | |
| auto -> detect script; fall back to base with an honest note | |
| slug -> force the family; same fallback | |
| """ | |
| lang = (lang_req or STATE["lang"] or "off").lower() | |
| if lang in ("", "off"): | |
| return STATE["base_model"], "" | |
| sc = detect_script(state) | |
| slug = lang if lang not in ("auto",) else (sc if sc != "unknown" else "") | |
| if not slug or slug == "unknown": | |
| return STATE["base_model"], "script=unknown; using base model" | |
| cand = f"{STATE['base_model']}-{slug}" | |
| if Path(cand).is_dir(): | |
| return cand, f"routed to {cand} (script={slug})" | |
| return STATE["base_model"], f"no {cand} substrate; using base model (honest fallback)" | |
| def _ngrams_policy(ngrams_req, script_slug: str) -> None: | |
| """v2.2 trigram policy: explicit wins; 'auto' = non-Latin routed -> on.""" | |
| g = ngrams_req if ngrams_req in ("on", "off") else STATE["ngrams"] | |
| if g == "on": | |
| _LIB.syfox_engine_set_ngrams(None, 1) | |
| elif g == "off": | |
| _LIB.syfox_engine_set_ngrams(None, 0) | |
| else: # auto: non-Latin routing is load-bearing, Latin keeps word stream | |
| _LIB.syfox_engine_set_ngrams( | |
| None, 1 if (script_slug and script_slug != "latin" | |
| and STATE["lang"] != "off") else 0) | |
| async def models(): | |
| """Discover model dirs from meta.json (no fabric load).""" | |
| out = [] | |
| for d in sorted(Path(".").glob("model-*")): | |
| if not d.is_dir(): | |
| continue | |
| meta = {} | |
| mp = d / "meta.json" | |
| if mp.exists(): | |
| try: | |
| meta = json.loads(mp.read_text()) | |
| except Exception: | |
| meta = {} | |
| out.append({ | |
| "id": d.name, | |
| "object": "model", | |
| "owned_by": "Mr-DS-ML-85", | |
| "core": meta.get("core", "si-substrate"), | |
| "version": meta.get("version"), | |
| "nodes": meta.get("nodes"), | |
| "lanes": meta.get("lanes"), | |
| "evidence_records": meta.get("evidence_records"), | |
| "contradictions": meta.get("contradictions"), | |
| "loaded": d.name in {Path(k).name for k in ENGINES}, | |
| }) | |
| return {"object": "list", "core_version": core_version(), "data": out} | |
| async def health(): | |
| info = {} | |
| if ENGINES: | |
| try: | |
| info = ENGINES[STATE["base_model"]].info() | |
| except Exception: | |
| info = {} | |
| return { | |
| "status": "ok", | |
| "core_version": core_version(), | |
| "model": STATE["base_model"], | |
| "lang": STATE["lang"], | |
| "ngrams": STATE["ngrams"], | |
| "flags": SERVER_FLAGS, | |
| "fabric": info, | |
| "deferrals": len(DEFERRALS), | |
| "uptime_s": round(time.time() - STATE["started"], 1), | |
| } | |
| async def deferrals(): | |
| """v2.2 active-learning loop, step 1: rows the fabric deferred on.""" | |
| return {"object": "list", "count": len(DEFERRALS), "data": DEFERRALS} | |
| async def deferrals_clear(): | |
| n = len(DEFERRALS) | |
| DEFERRALS.clear() | |
| return {"cleared": n} | |
| async def systemone(req: Request): | |
| body = await req.json() | |
| state = body.get("state", "") | |
| if isinstance(state, (dict, list)): | |
| state = json.dumps(state, ensure_ascii=False) | |
| norm, err = _norm_questions(body.get("questions", {})) | |
| if err: | |
| return err | |
| opts = body.get("options", {}) or {} | |
| if not isinstance(opts, dict): | |
| return JSONResponse(status_code=400, content={ | |
| "error": {"message": "options must be a JSON object"}}) | |
| with GLOBAL_LOCK: # field is single-threaded per engine; ngrams is global | |
| model_dir, note = _route(state, body.get("lang")) | |
| script_slug = "" | |
| lang_req = (body.get("lang") or STATE["lang"] or "off").lower() | |
| if lang_req not in ("", "off"): | |
| sc = detect_script(state) | |
| script_slug = lang_req if lang_req != "auto" else sc | |
| _ngrams_policy(opts.get("ngrams"), script_slug) | |
| try: | |
| eng = _get_engine(model_dir) | |
| answers, usage, evidence = eng.decide(state, norm, opts) | |
| except RuntimeError as e: | |
| return JSONResponse(status_code=422, content={ | |
| "error": {"message": str(e)}}) | |
| # options.defer_margin P: honest-uncertainty disclosure at the decision | |
| # layer (CLI-equal). The substrate still decided; a p1-p2 below P is | |
| # disclosed as a deferral (reason "low_margin") instead of a | |
| # confident-looking label on tied candidates. | |
| dm = opts.get("defer_margin") | |
| if isinstance(dm, (int, float)) and dm > 0: | |
| for a in answers.values(): | |
| if a.get("deferred"): | |
| continue | |
| vals = sorted(a.get("probabilities", {}).values(), reverse=True) | |
| if len(vals) >= 2 and vals[0] - vals[1] < dm: | |
| a["deferred"] = True | |
| a["reason"] = "low_margin" | |
| # v2.2 active-learning: log deferrals (state + full question schema) | |
| deferred = {qid: a for qid, a in answers.items() if a.get("deferred")} | |
| if deferred: | |
| row = {"ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"), | |
| "state": state, "questions": norm, | |
| "reasons": {qid: a.get("reason", "silence") for qid, a in deferred.items()}} | |
| DEFERRALS.append(row) | |
| if STATE["log_deferrals"]: | |
| with open(STATE["log_deferrals"], "a") as f: | |
| f.write(json.dumps(row, ensure_ascii=False) + "\n") | |
| out = {"answers": answers, "usage": usage, | |
| "model": body.get("model", Path(STATE["base_model"]).name)} | |
| if note: | |
| out["lang_note"] = note | |
| out["routed_model"] = Path(model_dir).name | |
| if evidence is not None: | |
| out["evidence"] = evidence | |
| return out | |
| def main(): | |
| ap = argparse.ArgumentParser(description="SyFox System One server (v3 API)") | |
| ap.add_argument("--model", default="model-tickets", help="base model directory") | |
| ap.add_argument("--port", type=int, default=8010) | |
| ap.add_argument("--host", default="127.0.0.1") | |
| ap.add_argument("--lang", default="off", | |
| help="off | auto | <script-slug> (v2.2 per-script routing)") | |
| ap.add_argument("--ngrams", default="auto", choices=["auto", "on", "off"], | |
| help="trigram lane policy (v2.2; auto = CLI routing policy)") | |
| ap.add_argument("--energy-norm", action="store_true", | |
| help="M1 decide-side measurement gain (corpus-scale fabrics)") | |
| ap.add_argument("--salience-gating", action="store_true", | |
| help="SI salience selection mode (off by default)") | |
| ap.add_argument("--miller-window", action="store_true", | |
| help="SI Miller [cap-4,cap] window mode (off by default)") | |
| ap.add_argument("--threads", type=int, default=0, | |
| help="M5 OMP parallel settle (needs an OMP build: make omp)") | |
| ap.add_argument("--log-deferrals", default=None, | |
| help="append every deferral to FILE.jsonl (active learning)") | |
| args = ap.parse_args() | |
| if not Path(args.model).is_dir(): | |
| raise SystemExit(f"syfox: model dir not found: {args.model}") | |
| STATE["base_model"] = args.model | |
| STATE["lang"] = args.lang | |
| STATE["ngrams"] = args.ngrams | |
| STATE["log_deferrals"] = args.log_deferrals | |
| SERVER_FLAGS.update(energy_norm=args.energy_norm, | |
| salience_gating=args.salience_gating, | |
| miller_window=args.miller_window, | |
| parallel_settle=args.threads > 0, threads=args.threads) | |
| _get_engine(args.model) # load the base fabric up front | |
| print(f"SyFox v{core_version()} serving on http://{args.host}:{args.port} " | |
| f"(model: {args.model}, core: si-substrate, lang: {args.lang}, " | |
| f"ngrams: {args.ngrams})") | |
| uvicorn.run(app, host=args.host, port=args.port, log_level="warning") | |
| if __name__ == "__main__": | |
| main() | |