""" Qwen3.6-27B MTP TQ3_4S — Custom HTML chat server for Hugging Face Spaces. Free CPU tier (2 vCPUs, 16 GB RAM). Architecture: - FastAPI server on port 7860 - GET / → HTML chat interface - POST /api/chat → SSE stream of reasoning + response tokens + live metrics - No Gradio — pure HTML/CSS/JS frontend Reasoning (Qwen3 thinking mode): The model's embedded chat template defaults to thinking ENABLED. We don't pass chat_template_kwargs (not supported in llama-cpp-python 0.3.32). The model emits reasoning tokens in delta.reasoning_content automatically. """ from __future__ import annotations import asyncio import json import os import re import time import traceback import uuid from threading import Lock from typing import Iterator, List import uvicorn from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse from huggingface_hub import hf_hub_download from llama_cpp import Llama from sse_starlette.sse import EventSourceResponse # ============================================================================= # Configuration # ============================================================================= REPO_ID = "openbmb/MiniCPM5-1B-GGUF" FILENAME = "MiniCPM5-1B-Q8_0.gguf" # --- CPU / memory tuning (free tier = 2 vCPUs, 16 GB RAM) ------------------ # MiniCPM5-1B is only 1.07 GB (Q8_0) — tons of headroom on 16 GB RAM. # # 100k context window — MiniCPM5 supports up to 262k (n_ctx_train). # KV cache at q8_0 for 100k context ≈ 2.3 GB (fits easily in 16 GB). # # N_CTX = 102400 : 100k token context (model supports up to 262k) # N_BATCH = 512 : prompt-eval batch size # KV_CACHE = q8_0 : optimal for CPU (memory bandwidth bound) N_THREADS = 2 N_CTX = 102400 N_BATCH = 512 KV_CACHE_DTYPE = "q8_0" N_GPU_LAYERS = 0 # --- Generation defaults --------------------------------------------------- # Max output = 30k tokens. At ~12 tok/s, 30k tokens ≈ 42 minutes. # Default is 4096 (reasonable for chat); users can bump to 30k via slider. MAX_TOKENS_DEFAULT = 4096 TEMPERATURE_DEFAULT = 0.7 TOP_P_DEFAULT = 0.9 REPEAT_PENALTY_DEFAULT = 1.05 PORT = int(os.environ.get("PORT", 7860)) SYSTEM_PROMPT_DEFAULT = ( "You are MiniCPM5, a helpful and concise assistant. " "Answer in clear, well-structured prose." ) # ChatML template — the model's embedded template already handles thinking, # but we set this explicitly as a fallback for models that omit it. CHAT_TEMPLATE = ( "{% for message in messages %}" "{% if message['role'] == 'system' %}" "<|im_start|>system\n{{ message['content'] }}<|im_end|>\n" "{% elif message['role'] == 'user' %}" "<|im_start|>user\n{{ message['content'] }}<|im_end|>\n" "{% elif message['role'] == 'assistant' %}" "<|im_start|>assistant\n{{ message['content'] }}<|im_end|>\n" "{% endif %}" "{% endfor %}" "{% if add_generation_prompt %}<|im_start|>assistant\n{% endif %}" ) # ============================================================================= # Quant label parser # ============================================================================= def _parse_quant(filename: str) -> str: m = re.search(r"(Q[0-9]+_[A-Z]+|IQ[0-9]+_[A-Z]+|TQ[0-9]+_[A-Z0-9]+|F16|F32)", filename, re.IGNORECASE) return m.group(1).upper() if m else "unknown" # ============================================================================= # Model download + load # ============================================================================= print(f"[boot] Downloading {FILENAME} from {REPO_ID} ...", flush=True) model_path = hf_hub_download( repo_id=REPO_ID, filename=FILENAME, repo_type="model", ) print(f"[boot] Model cached at: {model_path}", flush=True) model_size_mb = os.path.getsize(model_path) / (1024 * 1024) quant_label = _parse_quant(FILENAME) print(f"[boot] Model file size: {model_size_mb:.1f} MB ({quant_label})", flush=True) if model_size_mb < 100: raise RuntimeError( f"Model file is suspiciously small ({model_size_mb:.1f} MB). " f"Try clearing the HF cache and restarting." ) import llama_cpp as _llama_cpp_mod print(f"[boot] llama-cpp-python version: {_llama_cpp_mod.__version__}", flush=True) print("[boot] Loading Llama model (this can take 1-2 minutes) ...", flush=True) LOAD_ATTEMPTS = [ # MiniCPM5 has its own embedded chat template — do NOT override it. # q8_0 KV cache is optimal for CPU (tested: fp16 was 2x slower due to # memory bandwidth bottleneck). ("q8_0 KV cache, model's embedded template", dict( kv_cache_dtype=KV_CACHE_DTYPE, )), ("fp16 KV cache, model's embedded template", dict( kv_cache_dtype="fp16", )), ("bare minimum (all defaults)", dict()), ] llm = None last_error = None for label, extra_kwargs in LOAD_ATTEMPTS: print(f"[boot] Attempt: {label}", flush=True) try: llm = Llama( model_path=model_path, n_threads=N_THREADS, n_ctx=N_CTX, n_batch=N_BATCH, n_gpu_layers=N_GPU_LAYERS, use_mlock=False, use_mmap=False, # eager-load 1GB model into RAM (avoids 73s # mmap cold-start on first request over HF's # network-attached storage) verbose=False, # suppress per-request debug output (was True # for debugging load issues — no longer needed) **extra_kwargs, ) print(f"[boot] SUCCESS with: {label}", flush=True) break except Exception as exc: last_error = exc print(f"[boot] FAILED with: {label}", flush=True) print(f"[boot] error: {exc}", flush=True) print("-" * 70, flush=True) llm = None continue if llm is None: raise RuntimeError(f"All model load attempts failed. Last error: {last_error}") print("[boot] Model ready. Starting FastAPI server ...", flush=True) LLM_LOCK = Lock() # ============================================================================= # FastAPI app # ============================================================================= app = FastAPI(title="Qwen3.6-27B TQ3_4S Chat") # CORS — allow browser-based apps and tools to call the API directly. app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) MODEL_ID = FILENAME.replace(".gguf", "") def _build_messages(history: List[dict], system_prompt: str) -> List[dict]: """ Build the messages list for create_chat_completion. The HTML frontend sends the full conversation history (including the latest user message) via the `history` field. We prepend a system message if one isn't already present in the history. The Qwen3.6 chat template requires at least one user message — otherwise it raises "No user query found in messages." """ messages: List[dict] = [] # Check if the history already starts with a system message. has_system = ( len(history) > 0 and isinstance(history[0], dict) and history[0].get("role") == "system" ) if not has_system: messages.append({ "role": "system", "content": system_prompt or SYSTEM_PROMPT_DEFAULT, }) for msg in history: if isinstance(msg, dict) and msg.get("role") and msg.get("content") is not None: messages.append({ "role": msg["role"], "content": msg["content"], }) # Safety check: the Qwen3.6 template raises "No user query found" # if there's no user message. Log a warning if we're about to send # a messages list with no user role. has_user = any(m.get("role") == "user" for m in messages) if not has_user: print(f"[warn] _build_messages: no user message in history " f"(roles: {[m['role'] for m in messages]})", flush=True) return messages def _run_llm_stream( messages: List[dict], temperature: float, top_p: float, max_tokens: int, repeat_penalty: float, queue: asyncio.Queue, loop: asyncio.AbstractEventLoop, ) -> None: """ Synchronous worker that runs llama-cpp-python's blocking create_chat_completion in a background thread. Pushes events to the asyncio queue so the async SSE generator can yield them immediately without blocking the event loop. This is CRITICAL for streaming — if we ran create_chat_completion directly in the async path, each token would block the event loop and the SSE response wouldn't flush until the entire generation finished. """ def _put(event): # Thread-safe put into the asyncio queue. loop.call_soon_threadsafe(queue.put_nowait, event) try: _put({"type": "status", "stage": "processing_prompt"}) state = { "prompt_start": time.perf_counter(), "first_token": None, "first_reasoning_token": None, "first_response_token": None, "last_reasoning_token": None, "last_response_token": None, "last_token": None, "reasoning_tokens": 0, "response_tokens": 0, } def _tps(first, last, count): if first is None or last is None or count == 0: return 0.0 elapsed = last - first return count / elapsed if elapsed > 0 else 0.0 # State machine for parsing tags from content. # MiniCPM5 (and some other models) emit reasoning as ... # inside the regular content field, NOT as a separate reasoning_content # field. We parse these tags in real-time and split into reasoning vs # response events. # # States: # "thinking" — inside block, emit as reasoning # "responding" — after , emit as response # "buffering" — haven't seen yet, buffering to check think_state = "buffering" content_buffer = "" with LLM_LOCK: stream = llm.create_chat_completion( messages=messages, max_tokens=int(max_tokens), temperature=float(temperature), top_p=float(top_p), repeat_penalty=float(repeat_penalty), stream=True, ) for chunk in stream: delta = (chunk.get("choices") or [{}])[0].get("delta", {}) or {} now = time.perf_counter() # Check for reasoning_content field first (Qwen3 / GPT-oss style) reasoning_token = delta.get("reasoning_content") if reasoning_token: if state["first_token"] is None: state["first_token"] = now if state["first_reasoning_token"] is None: state["first_reasoning_token"] = now state["last_reasoning_token"] = now state["last_token"] = now state["reasoning_tokens"] += 1 _put({ "type": "reasoning", "token": reasoning_token, "count": state["reasoning_tokens"], "tps": _tps( state["first_reasoning_token"], state["last_reasoning_token"], state["reasoning_tokens"], ), }) continue # Content tokens — may contain tags (MiniCPM5 style) token = delta.get("content") if not token: continue # Parse tags from the token stream content_buffer += token while content_buffer: if think_state == "buffering": # Check if we have in the buffer if "" in content_buffer: # Emit anything before as response (rare) idx = content_buffer.index("") if idx > 0: pre = content_buffer[:idx] if state["first_token"] is None: state["first_token"] = now if state["first_response_token"] is None: state["first_response_token"] = now state["last_response_token"] = now state["last_token"] = now state["response_tokens"] += 1 _put({ "type": "response", "token": pre, "count": state["response_tokens"], "tps": _tps( state["first_response_token"], state["last_response_token"], state["response_tokens"], ), }) content_buffer = content_buffer[idx + 7:] # skip think_state = "thinking" if state["first_token"] is None: state["first_token"] = now if state["first_reasoning_token"] is None: state["first_reasoning_token"] = now state["last_reasoning_token"] = now state["last_token"] = now state["reasoning_tokens"] += 1 _put({ "type": "reasoning", "token": "", "count": state["reasoning_tokens"], "tps": _tps( state["first_reasoning_token"], state["last_reasoning_token"], state["reasoning_tokens"], ), }) elif len(content_buffer) > 7: # Not enough to contain — emit as response # but keep last 7 chars in case "" spans chunks safe = content_buffer[:-7] content_buffer = content_buffer[-7:] if safe and state["first_token"] is None: state["first_token"] = now if safe and state["first_response_token"] is None: state["first_response_token"] = now if safe: state["last_response_token"] = now state["last_token"] = now state["response_tokens"] += 1 _put({ "type": "response", "token": safe, "count": state["response_tokens"], "tps": _tps( state["first_response_token"], state["last_response_token"], state["response_tokens"], ), }) break else: break # buffer too short, wait for more elif think_state == "thinking": # Check for in the buffer if "" in content_buffer: idx = content_buffer.index("") reasoning_text = content_buffer[:idx] if reasoning_text: if state["first_token"] is None: state["first_token"] = now state["last_reasoning_token"] = now state["last_token"] = now state["reasoning_tokens"] += 1 _put({ "type": "reasoning", "token": reasoning_text, "count": state["reasoning_tokens"], "tps": _tps( state["first_reasoning_token"], state["last_reasoning_token"], state["reasoning_tokens"], ), }) content_buffer = content_buffer[idx + 8:] # skip think_state = "responding" elif len(content_buffer) > 8: # Emit as reasoning but keep last 8 chars safe = content_buffer[:-8] content_buffer = content_buffer[-8:] if safe: state["last_reasoning_token"] = now state["last_token"] = now state["reasoning_tokens"] += 1 _put({ "type": "reasoning", "token": safe, "count": state["reasoning_tokens"], "tps": _tps( state["first_reasoning_token"], state["last_reasoning_token"], state["reasoning_tokens"], ), }) break else: break # buffer too short elif think_state == "responding": # Everything goes to response if state["first_response_token"] is None: state["first_response_token"] = now state["last_response_token"] = now state["last_token"] = now state["response_tokens"] += 1 _put({ "type": "response", "token": content_buffer, "count": state["response_tokens"], "tps": _tps( state["first_response_token"], state["last_response_token"], state["response_tokens"], ), }) content_buffer = "" break # Final metrics total_tokens = state["reasoning_tokens"] + state["response_tokens"] ttft_ms = (state["first_token"] - state["prompt_start"]) * 1000 if state["first_token"] else 0 gen_time = (state["last_token"] - state["first_token"]) if state["first_token"] and state["last_token"] else 0 prompt_time = (state["first_token"] - state["prompt_start"]) if state["first_token"] else 0 total_time = prompt_time + gen_time total_tps = total_tokens / gen_time if gen_time > 0 else 0 reasoning_tps = _tps(state["first_reasoning_token"], state["last_reasoning_token"], state["reasoning_tokens"]) response_tps = _tps(state["first_response_token"], state["last_response_token"], state["response_tokens"]) _put({ "type": "metrics", "ttft_ms": round(ttft_ms, 0), "prompt_time_s": round(prompt_time, 2), "gen_time_s": round(gen_time, 2), "total_time_s": round(total_time, 2), "reasoning_tokens": state["reasoning_tokens"], "response_tokens": state["response_tokens"], "total_tokens": total_tokens, "reasoning_tps": round(reasoning_tps, 2), "response_tps": round(response_tps, 2), "total_tps": round(total_tps, 2), }) _put({"type": "done"}) except Exception as exc: traceback.print_exc() _put({"type": "error", "message": str(exc)}) async def _stream_chat_async( messages: List[dict], temperature: float, top_p: float, max_tokens: int, repeat_penalty: float, ) -> asyncio.Queue: """ Launch the blocking LLM stream in a background thread and return an asyncio.Queue that yields events as they arrive. Each token is available immediately — the event loop stays responsive for SSE flushing. """ queue: asyncio.Queue = asyncio.Queue() loop = asyncio.get_event_loop() # Run the sync worker in a thread pool. The worker pushes events to the # queue via call_soon_threadsafe, so the async side can await them. asyncio.get_event_loop().run_in_executor( None, _run_llm_stream, messages, temperature, top_p, max_tokens, repeat_penalty, queue, loop, ) return queue @app.get("/") async def index() -> HTMLResponse: return HTMLResponse(content=HTML_PAGE) @app.post("/api/chat") async def chat(request: Request): body = await request.json() messages = _build_messages( body.get("history", []), body.get("system_prompt", SYSTEM_PROMPT_DEFAULT), ) queue = await _stream_chat_async( messages=messages, temperature=body.get("temperature", TEMPERATURE_DEFAULT), top_p=body.get("top_p", TOP_P_DEFAULT), max_tokens=body.get("max_tokens", MAX_TOKENS_DEFAULT), repeat_penalty=body.get("repeat_penalty", REPEAT_PENALTY_DEFAULT), ) async def _event_generator(): while True: event = await queue.get() yield {"data": json.dumps(event)} if event.get("type") in ("done", "error"): break return EventSourceResponse(_event_generator()) @app.get("/api/health") async def health(): return JSONResponse({ "status": "ok", "model": FILENAME, "repo": REPO_ID, "quant": quant_label, "size_mb": round(model_size_mb, 1), "llama_cpp_version": _llama_cpp_mod.__version__, "n_threads": N_THREADS, "n_ctx": N_CTX, "kv_cache_dtype": KV_CACHE_DTYPE, }) # ============================================================================= # OpenAI-compatible API (/v1/*) # # These endpoints implement the OpenAI Chat Completions API spec so any # client that speaks OpenAI (Python `openai` library, LangChain, curl, etc.) # can connect directly: # # from openai import OpenAI # client = OpenAI(base_url="https://YOUR_SPACE.hf.space/v1", api_key="x") # resp = client.chat.completions.create( # model="Qwen3.6-27B-MTP-TQ3_4S", # messages=[{"role": "user", "content": "Hello"}], # stream=True, # ) # # Streaming format follows OpenAI SSE conventions: # data: {"choices":[{"delta":{"content":"..."}}]} # data: {"choices":[{"delta":{"reasoning_content":"..."}}]} ← Qwen3 thinking # data: [DONE] # # Non-streaming returns a standard chat.completion object. # ============================================================================= @app.get("/v1/models") async def list_models(): return { "object": "list", "data": [{ "id": MODEL_ID, "object": "model", "created": int(time.time()), "owned_by": REPO_ID.split("/")[0], }], } @app.post("/v1/chat/completions") async def chat_completions(request: Request): body = await request.json() messages = body.get("messages", []) stream = body.get("stream", False) temperature = body.get("temperature", TEMPERATURE_DEFAULT) top_p = body.get("top_p", TOP_P_DEFAULT) # OpenAI uses max_tokens (legacy) or max_completion_tokens (newer) max_tokens = body.get("max_completion_tokens") or body.get("max_tokens") or MAX_TOKENS_DEFAULT repeat_penalty = body.get("repeat_penalty", REPEAT_PENALTY_DEFAULT) stop = body.get("stop") # Normalize stop to a list (OpenAI accepts string or list) if isinstance(stop, str): stop = [stop] completion_id = f"chatcmpl-{uuid.uuid4().hex[:12]}" created = int(time.time()) model_name = body.get("model", MODEL_ID) if stream: queue = await _stream_chat_async(messages, temperature, top_p, max_tokens, repeat_penalty) async def _openai_sse(): # Initial role delta (OpenAI convention) yield {"data": json.dumps({ "id": completion_id, "object": "chat.completion.chunk", "created": created, "model": model_name, "choices": [{ "index": 0, "delta": {"role": "assistant"}, "finish_reason": None, }], })} while True: event = await queue.get() etype = event.get("type") if etype == "reasoning": yield {"data": json.dumps({ "id": completion_id, "object": "chat.completion.chunk", "created": created, "model": model_name, "choices": [{ "index": 0, "delta": {"reasoning_content": event["token"]}, "finish_reason": None, }], })} elif etype == "response": yield {"data": json.dumps({ "id": completion_id, "object": "chat.completion.chunk", "created": created, "model": model_name, "choices": [{ "index": 0, "delta": {"content": event["token"]}, "finish_reason": None, }], })} elif etype == "done": yield {"data": json.dumps({ "id": completion_id, "object": "chat.completion.chunk", "created": created, "model": model_name, "choices": [{ "index": 0, "delta": {}, "finish_reason": "stop", }], })} yield {"data": "[DONE]"} break elif etype == "error": yield {"data": json.dumps({ "error": {"message": event.get("message", "unknown error")}, })} break return EventSourceResponse(_openai_sse()) else: # Non-streaming: collect all tokens from the queue, return a single JSON. queue = await _stream_chat_async(messages, temperature, top_p, max_tokens, repeat_penalty) reasoning_content = "" content = "" final_metrics = None while True: event = await queue.get() etype = event.get("type") if etype == "reasoning": reasoning_content += event["token"] elif etype == "response": content += event["token"] elif etype == "metrics": final_metrics = event elif etype == "error": return JSONResponse( status_code=500, content={"error": {"message": event.get("message", "unknown error")}}, ) elif etype == "done": break prompt_tokens = 0 try: # Count prompt tokens using the model's tokenizer full_prompt = " ".join(m.get("content", "") for m in messages) prompt_tokens = len(llm.tokenize(full_prompt.encode("utf-8"))) except Exception: pass completion_tokens = (final_metrics or {}).get("total_tokens", 0) message_obj = { "role": "assistant", "content": content, } if reasoning_content: message_obj["reasoning_content"] = reasoning_content return { "id": completion_id, "object": "chat.completion", "created": created, "model": model_name, "choices": [{ "index": 0, "message": message_obj, "finish_reason": "stop", }], "usage": { "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "total_tokens": prompt_tokens + completion_tokens, }, } # ============================================================================= # HTML / CSS / JS — custom chat interface # ============================================================================= HTML_PAGE = r""" MiniCPM5-1B · CPU Chat

🤖 MiniCPM5-1B (Q8_0) · CPU

100k context · Max output 30k · Hybrid Reasoning ON · OpenAI API at /v1/chat/completions
Explain transformer attention in 3 sentences
Write a Python dedup function
Three habits for senior DevOps
Draft a bug-fix release note
""" # ============================================================================= # Launch # ============================================================================= if __name__ == "__main__": uvicorn.run( app, host="0.0.0.0", port=PORT, log_level="info", )