| import os, io, re, gc, json, base64, threading |
| from fnmatch import fnmatch |
| import numpy as np |
| import torch |
| import soundfile as sf |
| from PIL import Image |
| from gradio import Server |
| from fastapi.responses import HTMLResponse |
| from huggingface_hub import snapshot_download, HfApi, constants |
| from transformers import (AutoTokenizer, AutoModelForCausalLM, AutoModelForSeq2SeqLM, |
| WhisperProcessor, WhisperForConditionalGeneration, |
| VisionEncoderDecoderModel, ViTImageProcessor, |
| TextIteratorStreamer, StoppingCriteria, StoppingCriteriaList) |
|
|
| try: |
| import spaces |
| GPU = spaces.GPU |
| except Exception: |
| def GPU(duration=60): |
| return lambda f: f |
|
|
| app = Server() |
| HERE = os.path.dirname(os.path.abspath(__file__)) |
|
|
|
|
| @GPU(duration=10) |
| def _gpu_marker(): |
| return 0 |
|
|
|
|
| def M(repo, kind, size, group, bf16=False): |
| return {"repo": repo, "kind": kind, "size": size, "group": group, "bf16": bf16} |
|
|
| SM, MID, BIG = "💬 دردشة صغيرة (أقل من 1B)", "🧠 متوسطة (1–2B)", "🚀 كبيرة (3–4B) — بطيئة على CPU" |
| T5, CP, CU = "⚡ سريعة (سؤال وجواب)", "🐣 إكمال نصي", "🧩 مخصّص" |
| MODELS = { |
| |
| "qwen25": M("Qwen/Qwen2.5-0.5B-Instruct", "chat", "~1GB", SM), |
| "qwen3": M("Qwen/Qwen3-0.6B", "chat", "~1.2GB", SM), |
| "qwen2": M("Qwen/Qwen2-0.5B-Instruct", "chat", "~1GB", SM), |
| "qwen": M("Qwen/Qwen1.5-0.5B-Chat", "chat", "~1GB", SM), |
| "qcoder05": M("Qwen/Qwen2.5-Coder-0.5B-Instruct", "chat", "~1GB", SM), |
| "lfm2": M("LiquidAI/LFM2-350M", "chat", "~700MB", SM), |
| "lfm2_700": M("LiquidAI/LFM2-700M", "chat", "~1.4GB", SM), |
| "smol360": M("HuggingFaceTB/SmolLM2-360M-Instruct", "chat", "~720MB", SM), |
| "smol135": M("HuggingFaceTB/SmolLM2-135M-Instruct", "chat", "~270MB", SM), |
| |
| "qwen25_15": M("Qwen/Qwen2.5-1.5B-Instruct", "chat", "~3GB", MID), |
| "qwen3_17": M("Qwen/Qwen3-1.7B", "chat", "~3.4GB", MID), |
| "qcoder15": M("Qwen/Qwen2.5-Coder-1.5B-Instruct", "chat", "~3GB", MID), |
| "smol17": M("HuggingFaceTB/SmolLM2-1.7B-Instruct", "chat", "~3.4GB", MID), |
| "lfm2_12": M("LiquidAI/LFM2-1.2B", "chat", "~2.4GB", MID), |
| "tinyllama": M("TinyLlama/TinyLlama-1.1B-Chat-v1.0", "chat", "~2.2GB", MID), |
| |
| "qwen25_3": M("Qwen/Qwen2.5-3B-Instruct", "chat", "~6GB", BIG, True), |
| "smol3": M("HuggingFaceTB/SmolLM3-3B", "chat", "~6GB", BIG, True), |
| "phi35": M("microsoft/Phi-3.5-mini-instruct", "chat", "~7.6GB", BIG, True), |
| "qwen3_4": M("Qwen/Qwen3-4B-Instruct-2507", "chat", "~8GB", BIG, True), |
| |
| "lamini": M("MBZUAI/LaMini-Flan-T5-77M", "t5", "~300MB", T5), |
| "lamini248": M("MBZUAI/LaMini-Flan-T5-248M", "t5", "~1GB", T5), |
| "lamini783": M("MBZUAI/LaMini-Flan-T5-783M", "t5", "~3GB", T5), |
| "flansmall": M("google/flan-t5-small", "t5", "~300MB", T5), |
| "flanbase": M("google/flan-t5-base", "t5", "~1GB", T5), |
| "flanlarge": M("google/flan-t5-large", "t5", "~3GB", T5), |
| "t5small": M("google/t5-small", "t5", "~240MB", T5), |
| |
| "distilgpt2": M("distilbert/distilgpt2", "alpaca", "~350MB", CP), |
| "gpt2": M("openai-community/gpt2", "alpaca", "~500MB", CP), |
| "opt125": M("facebook/opt-125m", "alpaca", "~250MB", CP), |
| "opt350": M("facebook/opt-350m", "alpaca", "~650MB", CP), |
| "lgpt": M("MBZUAI/LaMini-GPT-124M", "alpaca", "~250MB", CP), |
| "lgpt774": M("MBZUAI/LaMini-GPT-774M", "alpaca", "~1.5GB", CP), |
| |
| "whisper": M("openai/whisper-tiny.en", "asr", "~150MB", "aux"), |
| "caption": M("nlpconnect/vit-gpt2-image-captioning", "caption", "~950MB", "aux"), |
| } |
| TEXT_KINDS = ("chat", "t5", "alpaca") |
|
|
| _pipes, _state, _cancels = {}, {}, {} |
| _lock = threading.Lock() |
| F32 = torch.float32 |
|
|
|
|
| def _st(mid): |
| return _state.setdefault(mid, {"state": "idle", "pct": 0, "error": ""}) |
|
|
|
|
| def _plan(repo): |
| """الملفات المطلوبة وحجمها الإجمالي (لحساب النسبة).""" |
| sib = HfApi().model_info(repo, files_metadata=True).siblings |
| ign = ["*.h5", "*.msgpack", "*.ot", "*.onnx", "*.onnx_data", "*.tflite", |
| "*.gguf", "onnx/*", "openvino/*", "coreml/*"] |
| if any(s.rfilename.endswith(".safetensors") for s in sib): |
| ign.append("*.bin") |
| keep = [s for s in sib if not any(fnmatch(s.rfilename, p) for p in ign)] |
| return ign, sum(s.size or 0 for s in keep) |
|
|
|
|
| def _watch(mid, repo, total, stop): |
| """يقيس حجم ما نزل في الكاش لتحديث النسبة.""" |
| d = os.path.join(constants.HF_HUB_CACHE, "models--" + repo.replace("/", "--")) |
| while not stop.is_set(): |
| n = 0 |
| for r, _, fs in os.walk(d): |
| for f in fs: |
| p = os.path.join(r, f) |
| try: |
| if not os.path.islink(p): |
| n += os.path.getsize(p) |
| except OSError: |
| pass |
| if total: |
| _st(mid)["pct"] = min(99.0, n * 100 / total) |
| stop.wait(0.7) |
|
|
|
|
| def _build(info): |
| k, r = info["kind"], info["repo"] |
| if k == "asr": |
| return {"proc": WhisperProcessor.from_pretrained(r), |
| "model": WhisperForConditionalGeneration.from_pretrained(r, dtype=F32).eval()} |
| if k == "caption": |
| return {"proc": ViTImageProcessor.from_pretrained(r), "tok": AutoTokenizer.from_pretrained(r), |
| "model": VisionEncoderDecoderModel.from_pretrained(r, dtype=F32).eval()} |
| cls = AutoModelForSeq2SeqLM if k == "t5" else AutoModelForCausalLM |
| dt = torch.bfloat16 if info.get("bf16") else F32 |
| return {"tok": AutoTokenizer.from_pretrained(r), "model": cls.from_pretrained(r, dtype=dt).eval()} |
|
|
|
|
| def _free_text(keep): |
| """نموذج نصي واحد في الذاكرة فقط، لتجنّب نفاد RAM مع النماذج الكبيرة.""" |
| for k in list(_pipes): |
| if k != keep and MODELS.get(k, {}).get("kind") in TEXT_KINDS: |
| _pipes.pop(k, None) |
| _st(k).update(state="idle", pct=0) |
| gc.collect() |
|
|
|
|
| def _load_sync(mid): |
| info, s, stop = MODELS[mid], _st(mid), threading.Event() |
| s.update(state="downloading", pct=0, error="") |
| try: |
| ign, total = _plan(info["repo"]) |
| threading.Thread(target=_watch, args=(mid, info["repo"], total, stop), daemon=True).start() |
| snapshot_download(info["repo"], ignore_patterns=ign) |
| stop.set() |
| s.update(state="loading", pct=100) |
| if info["kind"] in TEXT_KINDS: |
| _free_text(mid) |
| _pipes[mid] = _build(info) |
| s["state"] = "ready" |
| except Exception as e: |
| stop.set() |
| s.update(state="error", error=str(e)[:300]) |
|
|
|
|
| class StopOnCancel(StoppingCriteria): |
| def __init__(self, rid): |
| self.rid = rid |
|
|
| def __call__(self, input_ids, scores, **kw): |
| return torch.full((input_ids.shape[0],), bool(_cancels.get(self.rid)), |
| dtype=torch.bool, device=input_ids.device) |
|
|
|
|
| def _alpaca(msgs): |
| q = next((m["content"] for m in reversed(msgs) if m["role"] == "user"), "") |
| return ("Below is an instruction that describes a task. Write a response that " |
| f"appropriately completes the request.\n\n### Instruction:\n{q}\n\n### Response:\n") |
|
|
|
|
| |
| @app.api(name="models") |
| def api_models() -> dict: |
| return {"models": [{"id": k, **v} for k, v in MODELS.items()]} |
|
|
|
|
| @app.api(name="register") |
| def api_register(repo: str, kind: str = "chat") -> dict: |
| repo = repo.strip() |
| if not re.fullmatch(r"[\w.\-]+/[\w.\-]+", repo) or kind not in TEXT_KINDS: |
| return {"ok": False, "error": "صيغة غير صحيحة. مثال: Qwen/Qwen2.5-3B-Instruct"} |
| mid = "c_" + re.sub(r"\W", "_", repo) |
| MODELS.setdefault(mid, M(repo, kind, "؟", CU, True)) |
| return {"ok": True, "model": {"id": mid, **MODELS[mid]}} |
|
|
|
|
| @app.api(name="status") |
| def api_status() -> dict: |
| out = {} |
| for mid in MODELS: |
| s = _st(mid) |
| if mid in _pipes: |
| s["state"] = "ready" |
| out[mid] = {"state": s["state"], "pct": s["pct"], "error": s["error"]} |
| return out |
|
|
|
|
| @app.api(name="load") |
| def api_load(model_id: str) -> dict: |
| if model_id not in MODELS: |
| return {"ok": False, "error": "نموذج غير معروف"} |
| with _lock: |
| if model_id not in _pipes and _st(model_id)["state"] not in ("downloading", "loading"): |
| _st(model_id)["state"] = "downloading" |
| threading.Thread(target=_load_sync, args=(model_id,), daemon=True).start() |
| return {"ok": True} |
|
|
|
|
| @app.api(name="unload") |
| def api_unload(model_id: str = "") -> dict: |
| with _lock: |
| for mid in ([model_id] if model_id else list(_pipes)): |
| _pipes.pop(mid, None) |
| if mid in _state: |
| _state[mid].update(state="idle", pct=0) |
| return {"ok": True} |
|
|
|
|
| @app.api(name="cancel") |
| def api_cancel(req_id: str) -> dict: |
| _cancels[req_id] = True |
| return {"ok": True} |
|
|
|
|
| @app.api(name="generate") |
| def api_generate(model_id: str, messages: str, req_id: str, |
| max_tokens: int = 256, temperature: float = 0.7) -> dict: |
| _cancels[req_id] = False |
| try: |
| p = _pipes.get(model_id) |
| if p is None: |
| yield {"ok": False, "error": "النموذج غير محمّل"} |
| return |
| kind, tok, model = MODELS[model_id]["kind"], p["tok"], p["model"] |
| msgs = json.loads(messages) |
|
|
| special = True |
| if kind == "chat": |
| try: |
| prompt = tok.apply_chat_template(msgs, tokenize=False, |
| add_generation_prompt=True, enable_thinking=False) |
| special = False |
| except Exception: |
| prompt = _alpaca(msgs) |
| elif kind == "t5": |
| prompt = next((m["content"] for m in reversed(msgs) if m["role"] == "user"), "") |
| else: |
| prompt = _alpaca(msgs) |
|
|
| enc = tok(prompt, return_tensors="pt", add_special_tokens=special) |
| streamer = TextIteratorStreamer(tok, skip_prompt=(kind != "t5"), skip_special_tokens=True) |
| kw = dict(**enc, max_new_tokens=int(max_tokens), streamer=streamer, |
| stopping_criteria=StoppingCriteriaList([StopOnCancel(req_id)])) |
| if temperature > 0: |
| kw.update(do_sample=True, temperature=float(temperature), top_p=0.9) |
| else: |
| kw["do_sample"] = False |
| if kind != "t5": |
| kw.update(repetition_penalty=1.1, pad_token_id=tok.pad_token_id or tok.eos_token_id) |
|
|
| errs = [] |
|
|
| def run(): |
| try: |
| with torch.inference_mode(): |
| model.generate(**kw) |
| except Exception as e: |
| errs.append(str(e)) |
| streamer.end() |
|
|
| th = threading.Thread(target=run, daemon=True) |
| th.start() |
| out = "" |
| for piece in streamer: |
| if _cancels.get(req_id): |
| break |
| out += piece |
| yield {"ok": True, "text": out, "done": False} |
| th.join(timeout=2) |
| if errs: |
| yield {"ok": False, "error": errs[0]} |
| else: |
| yield {"ok": True, "text": out.strip(), "done": True} |
| except Exception as e: |
| yield {"ok": False, "error": str(e)} |
| finally: |
| _cancels.pop(req_id, None) |
|
|
|
|
| @app.api(name="asr") |
| def api_asr(audio_b64: str) -> dict: |
| try: |
| p = _pipes.get("whisper") |
| if p is None: |
| return {"ok": False, "error": "نموذج الصوت غير محمّل"} |
| data, sr = sf.read(io.BytesIO(base64.b64decode(audio_b64)), dtype="float32") |
| if data.ndim > 1: |
| data = data.mean(axis=1) |
| if sr != 16000: |
| data = np.interp(np.linspace(0, len(data) - 1, int(len(data) * 16000 / sr)), |
| np.arange(len(data)), data).astype("float32") |
| feats = p["proc"](data[:16000 * 30], sampling_rate=16000, return_tensors="pt").input_features |
| with torch.inference_mode(): |
| ids = p["model"].generate(feats, max_new_tokens=220) |
| return {"ok": True, "text": p["proc"].batch_decode(ids, skip_special_tokens=True)[0].strip()} |
| except Exception as e: |
| return {"ok": False, "error": str(e)} |
|
|
|
|
| @app.api(name="caption") |
| def api_caption(image_b64: str) -> dict: |
| try: |
| p = _pipes.get("caption") |
| if p is None: |
| return {"ok": False, "error": "نموذج الصور غير محمّل"} |
| img = Image.open(io.BytesIO(base64.b64decode(image_b64))).convert("RGB") |
| pix = p["proc"](images=img, return_tensors="pt").pixel_values |
| with torch.inference_mode(): |
| ids = p["model"].generate(pix, max_new_tokens=32, num_beams=4) |
| return {"ok": True, "text": p["tok"].decode(ids[0], skip_special_tokens=True).strip()} |
| except Exception as e: |
| return {"ok": False, "error": str(e)} |
|
|
|
|
| @app.get("/", response_class=HTMLResponse) |
| async def serve_ui(): |
| with open(os.path.join(HERE, "index.html"), "r", encoding="utf-8") as f: |
| return f.read() |
|
|
|
|
| app.launch(server_name="0.0.0.0", |
| server_port=int(os.environ.get("GRADIO_SERVER_PORT", 7860)), |
| ssr_mode=False, show_error=True) |
