import json import os from pathlib import Path import subprocess import threading import time import urllib.error import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer MODEL_REVISION = "a5995439030829f0c6ef1980f69192660df86f38" MODEL_REPO = "CodeMasterCody3D/taardis-27b-full-ternary" MODEL_NAME = "TAARDIS-27B-Full-Ternary-V2-1.75bit.gguf" ADAPTER_NAME = "TAARDIS-27B-Doctors-V3.lora.gguf" EXPECTED_SIZES = {MODEL_NAME: 5904497568, ADAPTER_NAME: 322730176} MODEL_DIR = Path(os.environ.get("TAARDIS_MODEL_DIR", "/tmp/taardis-models")) RUNTIME = os.environ.get("TAARDIS_RUNTIME", "/opt/llama/bin/llama-server") RUNTIME_URL = "http://127.0.0.1:8080" state = {"status": "starting", "message": "Preparing the public TAARDIS model."} generation_lock = threading.Lock() runtime_process = None PAGE = r"""TAARDIS research demo | Internectual Labs
INTERNECTUAL LABSExplore the lab ↗
Experimental model / public research demo

Meet TAARDIS.

A compact representation of a 27-billion-parameter model. Try a short prompt using the public V2 weights and Doctors V3 corrections.

Runtime status
Checking model availability…
Free CPU hosting. Startup downloads about 6.2 GB and loads a custom runtime; responses may be slow.
One request at a time · 4K context · up to 128 output tokens
Response
Your response will appear here.
Model and limitations

This is a research model. Aggressive quantization can reduce quality. Responses can be incomplete or incorrect. Do not use this demo for consequential decisions or submit private information.

The interface does not retain prompts in an application database. Inputs are sent to the Space server for inference; hosting infrastructure may process operational logs.

Model card and Apache 2.0 attribution ↗ · Required public runtime ↗

This CPU demo has no GPU grant yet. GPU support would improve interactive use; performance is measured only after an actual successful run.

""" def download(filename, remote_path): dest = MODEL_DIR / filename if dest.exists() and dest.stat().st_size == EXPECTED_SIZES[filename]: return dest state.update(status="loading", message=f"Downloading public model files: {filename}.") url = f"https://huggingface.co/{MODEL_REPO}/resolve/{MODEL_REVISION}/{remote_path}" temp = dest.with_suffix(dest.suffix + ".part") with urllib.request.urlopen(url, timeout=600) as response, temp.open("wb") as output: while chunk := response.read(8 * 1024 * 1024): output.write(chunk) if temp.stat().st_size != EXPECTED_SIZES[filename]: raise ValueError("Model download was incomplete.") with temp.open("rb") as downloaded: if downloaded.read(4) != b"GGUF": raise ValueError("Model download is not a GGUF file.") temp.replace(dest) return dest def startup(): global runtime_process try: MODEL_DIR.mkdir(parents=True, exist_ok=True) model = download(MODEL_NAME, MODEL_NAME) adapter = download(ADAPTER_NAME, f"doctors/{ADAPTER_NAME}") state.update(status="loading", message="Loading TAARDIS V2 and Doctors V3 into the CPU runtime.") threads = str(max(1, min(4, int(os.environ.get("TAARDIS_THREADS", "2"))))) runtime_process = subprocess.Popen([ RUNTIME, "-m", str(model), "--lora", str(adapter), "--host", "127.0.0.1", "--port", "8080", "-c", "4096", "-b", "64", "-ub", "64", "-np", "1", "-ngl", "0", "-t", threads, "-tb", threads, "--no-warmup", ]) while runtime_process.poll() is None: try: with urllib.request.urlopen(RUNTIME_URL + "/health", timeout=3) as response: if response.status == 200: state.update(status="ready", message="TAARDIS is ready. CPU responses may take several minutes.") break except (urllib.error.URLError, TimeoutError): pass time.sleep(2) runtime_process.wait() state.update(status="error", message="The model runtime stopped. Please check the Space runtime logs.") except Exception as exc: print(f"Runtime setup failed: {exc}", flush=True) state.update(status="error", message="The model could not start. The owner can inspect Space runtime logs.") class Handler(BaseHTTPRequestHandler): def log_message(self, *_): pass def respond(self, status, body, content_type="application/json"): data = body.encode() if isinstance(body, str) else json.dumps(body).encode() self.send_response(status) self.send_header("Content-Type", content_type + "; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.send_header("Cache-Control", "no-store") self.send_header("X-Content-Type-Options", "nosniff") self.end_headers() self.wfile.write(data) def do_GET(self): if self.path in ("/", "/index.html"): self.respond(200, PAGE, "text/html") elif self.path == "/api/status": self.respond(200, state) else: self.respond(404, {"error": "Not found."}) def do_POST(self): if self.path != "/api/chat": self.respond(404, {"error": "Not found."}) return if state["status"] != "ready": self.respond(503, {"error": state["message"]}) return if not generation_lock.acquire(blocking=False): self.respond(429, {"error": "Another request is running. Please try again after it finishes."}) return try: length = int(self.headers.get("Content-Length", "0")) if not 0 < length <= 30000: raise ValueError("Invalid request size.") data = json.loads(self.rfile.read(length)) prompt = data.get("prompt", "") if not isinstance(prompt, str) or not 1 <= len(prompt.strip()) <= 6000: raise ValueError("Enter a prompt of 1 to 6000 characters.") payload = { "messages": [{"role": "user", "content": prompt}], "max_tokens": 128, "temperature": 0.6, "top_p": 0.95, "top_k": 20, "repeat_penalty": 1.3, "stream": True, "chat_template_kwargs": {"enable_thinking": False}, } start = time.monotonic() request = urllib.request.Request(RUNTIME_URL + "/v1/chat/completions", data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}) with urllib.request.urlopen(request, timeout=900) as response: self.send_response(200) self.send_header("Content-Type", "text/event-stream; charset=utf-8") self.send_header("Cache-Control", "no-store") self.send_header("X-Accel-Buffering", "no") self.end_headers() for line in response: if not line.startswith(b"data: "): continue data = line[6:].strip() if data == b"[DONE]": break chunk = json.loads(data) choices = chunk.get("choices", []) delta = choices[0].get("delta", {}) if choices else {} text = delta.get("content") or delta.get("reasoning_content") or "" if text: self.wfile.write(("data: " + json.dumps({"text": text}) + "\n\n").encode()) self.wfile.flush() self.wfile.write(("data: " + json.dumps({"done": True, "elapsed_seconds": round(time.monotonic() - start, 1)}) + "\n\n").encode()) self.wfile.flush() except (ValueError, json.JSONDecodeError) as exc: self.respond(400, {"error": str(exc)}) except Exception as exc: print(f"Generation failed: {type(exc).__name__}", flush=True) self.respond(502, {"error": "The model could not finish this request. Try a shorter prompt or wait for the runtime to recover."}) finally: generation_lock.release() if __name__ == "__main__": threading.Thread(target=startup, daemon=True).start() ThreadingHTTPServer(("0.0.0.0", int(os.environ.get("PORT", "7860"))), Handler).serve_forever()