import base64 import gc import hashlib import json import os import platform import shutil import subprocess import sys import tempfile import threading import zlib from pathlib import Path from contextlib import asynccontextmanager from dataclasses import dataclass from typing import Any, Iterable, Literal, Mapping, Optional, Sequence # Must be configured before importing torch / llama-cpp. CPU_THREADS = int(os.getenv("CPU_THREADS", "4")) MAX_CONTEXT_TOKENS = int(os.getenv("MAX_CONTEXT_TOKENS", "4096")) MAX_NEW_TOKENS = int(os.getenv("MAX_NEW_TOKENS", "1024")) DEFAULT_MODEL = os.getenv("DEFAULT_MODEL", "bachvnju-vbpt-1-0.5B") MAX_JSON_SCHEMA_BYTES = int(os.getenv("MAX_JSON_SCHEMA_BYTES", "24576")) MAX_JSON_SCHEMA_DEPTH = int(os.getenv("MAX_JSON_SCHEMA_DEPTH", "16")) os.environ.setdefault("OMP_NUM_THREADS", str(CPU_THREADS)) os.environ.setdefault("MKL_NUM_THREADS", str(CPU_THREADS)) os.environ.setdefault("TOKENIZERS_PARALLELISM", "false") import torch from fastapi import FastAPI, HTTPException from huggingface_hub import HfApi, hf_hub_download, snapshot_download from jsonschema import Draft202012Validator, SchemaError, ValidationError from pydantic import BaseModel, Field, model_validator # GGUF and ONNX remain optional: the dedicated VBPT path only needs PyTorch # and Transformers, so a missing optional backend must not block API startup. try: from llama_cpp import Llama except ImportError: Llama = None # type: ignore[assignment] try: from optimum.onnxruntime import ORTModelForCausalLM except ImportError: ORTModelForCausalLM = None # type: ignore[assignment] from transformers import AutoModelForCausalLM, AutoTokenizer # ----------------------------------------------------------------------------- # Embedded VBPT loader (activated only for MODEL_CATALOG entries with # ``loader == "vbpt"``). This API has no dependency on vbpt_dataloader.py. # ----------------------------------------------------------------------------- VBPT_MODEL_KEY = "bachvnju-vbpt-1-0.5B" VBPT_REPO_ID = "bachvnju/vbpt-1-0.5B" # ----------------------------------------------------------------------------- # Embedded audited GDN runtime # ----------------------------------------------------------------------------- # The public VBPT repository currently ships a remote model implementation that # imports FLA unconditionally. The benchmark supplied with this project embeds # a compatible GDN v4.8 implementation with two explicit execution backends: # * ``fla`` GPU / flash-linear-attention # * ``reference`` pure PyTorch CPU reference (slow, but no FLA required) # # We download weights/tokenizer/config from the VBPT repository but replace only # its custom Python runtime in a local overlay. This prevents arbitrary latest # remote Python from being executed at startup and preserves the supplied # benchmark's CPU fallback. Payload hashes are verified before materialization. VBPT_RUNTIME_FILES = {'__init__.py': {'compressed_b85': 'c-l)V%Fk8MOU}p&kkV8tq<)_N!a=HB2og~S}cD>H4HM=pZbd2#gl(*;XKt3|;+eO@FsV8*5M>bQrU1aKygM%7uH<}iP-ri~OCt5@*nN-AZ8WS9+$V4FP4WX0@spyx=ZDsK$SaC&XRRj@iG;>EJR|vw{oYssd>JxYFH~_xE)beDdN8-jnIYzh~J2!l{)#>S%tA+fnj*V`OHf2N6TV$IJR!;R4loMFq~n0z}(5(lMip+O^9wH4&>K&@27r@DlC7NL)tm0)AuLDOx2dWH)YpFkFaAGBuu|bd;{SfVJD{(tQ2>=sda8qvRuk6gAk9<_a`3+Se2SX*h(%NK&*;dSab3yoFlCI8Pi7LBIvwm6FN&6&8Caukn(FenL1t^li_}5AIH3rs?VOq*o|x3pbfcFQf~&A-%JKb)}FB<*XApD^o{+OeW@#4Zu~`uv)i2w+ua=r^J<(?x@oH|nGyn4&70FRwyf&{5Lcqmj%h3!y`yXivradY`R}}_mB3hR$0ed+-`+d}_`Zc@X(Z3`a*Tr8t)A$%nNB<3{f&eWlDgoxd6m_-L;nY|+%^(Uu2O9~ER+EwNV@Rj<751{ICcVbdBI9q)@nBeW`&!v6LMYDM#tQTj@u}RZW#+|n+!LW-DNQ80Z07p%@|oiVhOC~G&g@e4v?^((zv;?gpkqP=#g1|13NW0JANzk^~OwwPvueQ0@UBN44@|t^k;U7(T6CvFrx*DI=1;z6p&}q^{%M3=8%RmISlZ*(7aI2ZajG#howc{5VsZ)SU-AX3U31NeE@T^qetBeRkjVcDL2(T8jpWVoo6i^%w>KD72`Dn#fH|5a+5-`R@+Ujpbi^b$AYr4g5p)Ht{T#Par%~IRB0LxR`Po|-3P$2+uhBHpE|Q5k-F!0HJ+ThMRc3Y)LU6aw?I*^;b2NPGTk;cbVCInE74!7C55QflBWWY`zn>mVaaNhIcGr}x~z5Q6VJ2Gt38EP{wS~!*^dEVzT84GtaT^psZEyx7ilVi9NfHF+_5Pl`gKRB?D@}bH#xA5fwdsn&_-05P#ODp#VS!sz1V-+f;V7s)_JgK>SmT8rt3@&LxuVLsZ)86IosD;^E8M(KUE{f#)afP7+_%6*94?eoh-MERvV!5t*wrX8Za0mkV(7$mWtm?iLfL>mxe}JS5?EJ(Ng1Y{2t+jX>_{?MuZl<3hc7*2S{R)!jo7d;LAVo(;+yLD3ev85PE3i8qIHJh*0)Ac&;LB}el?56y2>_2^?iAx}CFb?`Xt-5Qz^6tY1;Vdb0+Z4X9hUq3jJcvwmShdq9?7ftx!V0n^+SQTJZ96K__VKWSE4@XI352YdxJyFOWI|=+e8tLSeMWe`H-+(;!)&sdWNWdq1y+9t*^C%`&^}76R7BJ6oieXn=cjBUbdIs{ip3hAEf&Kot6-ror3$)K)rU+T>-H-Ilg3@8wD9F&MnrR#ujtzZKB!f->8P#*j@=pI%Qi1EPJlDdNuknBZe)rzLEGo6-~})F=BnVkR*r>eHYYDz;YbLcv!maqXOoF!2M;PR@-$4kOAEV&V~6Z3b~16-TES9e@(%uI_K2RvSq|a%|F`aehV30X`F*pJjsxGL>8lQJ~hR-{dG|3c9B_V=UXP87~4AmG6T%pMs7%`xJzu|B&vAw(t0q%U1$ich+;92?)rD`=xs6n`wL~jjcs|{O9*0HY}Nv0F?>&1SLe)>N>8b*e5A=mc(cDakWNIa5v4P+?g-=awvR7?w=l$jVKWLRK@sWX&;t@vZpWptP!i>9yI=UZ{R~V?*2zQr{2^vzOItCf{QMGQjwYaRQ#Z=aMl}|KSnm!coq0*d=GvuI0iqBRTfP)@n`#3!CHR5*u`lrcCa<_gow=goq0+dngQUa--0@qxrVOo0yHAzeR}9zdP-*NH0jJ9l82Yad)xtC3tBGqY=565XJjYHV%&kg{?{Zt>5wJ&Cf0DkMG%KU|JZSQOx}Fy7(V5SjpAd%aR&Lx={z9CP}gQJUv^_cK|G)ZZK32EE&-4<5GXEMQ*&D@XUPTp-50(2!T$hT_L%q', 'sha256': '1f6629d282e91ba48d67729daeb8b2487f4cad7a19f33720efaab9cef2b46e68', 'size': 8556}, 'modeling_gdn_mca_mla.py': {'compressed_b85': 'c-rNCX?NSmwcvOC3XFVa0?Ht4DV@BGI_wj-m+n5XJD#>X?}Ls8f(S@NL;@55YH^bP{_awH6+p_eb24xGBr*Y1-P&&5x_j;K@Bg^DysYZW=uJK^qg7Ls%cxqdm*uLg`@FB3I=X1us9%)P4|!h}(^uuP&wnoa=xOxs<%{XNw=bf3zUlI1^!8o4xA##Nj8wr@byv30rYpPX4U`6mKfM1n%IgCDA2sW;rDmq(^`^SXm(XnV{{Car&KFTzUX*QFL&u=f%lE&g(NBF;mp1^e+njfO)o*%$n)f}_l3lf1Pxtos_xCQ^W))@G#irl1WtQOxo3@Ac>xKu}?d^$otGr*x-%TffE%I)$tj^`*ZQj-}X&nPD@;;w0^A1NS3aD3HqQ76`9E*>CS!0v=QnlIRXsw5|sJgzb&Nn!gyo>PBE>PF_zOLQ3>BXj=Gkh%YMs$F_Zu)B3rSnBOzgjmHOth%0-|UIvZC-aUnX9tx(#x_IYb4A6px(Tz?y5RahY$&doHd(%z3J7we4lsaJN)(+djFqQzj&W_eG(7Res8aD?~nH)_>U^2^QOM2E;l^3?6Rn{)jZEu%Uq58)z5(SFR9pGc{eZDee@H3dfm27dmKeiV6ED6-bdwKIo~i=QfC%b2l%-x0D0r`j-X+ms{VeWd+@Lh?BO^7-b@Y%)~e6=Wt?Soz5-;Q?CtISN#i5^m0*IC9*18~c%ooCeduBH=4}!E_(rW;#0wqWLTMQKdRfh@9yV}S=D3rKD!;57*s=32#SKZImMiFM*Bz@3al&Z&X4&NZ$FlC4_7ukUb5oa8e$}rZR^ihF;g_qXz0dN^-Og277&b1s;fTB;;vj?B+;TOin7kS>JxMj2tyK;YnaIkgbrX2%jxe$W&E$(02z-bsRA<5p8#BXTUM6~tOqT>&s)Gt7)i_>h7NYI(cHwT3++Nh_`9e23VRwTaN%?nD0vB8WT-p?lW6*#^+I>KM!HY{WUG2~Ad0u$b=$|PH2!B5Olc-y$m`}dKANPfGOy#JTFs`1Npw|~YyADO-INnJDCIa5xJlYKu???)GosW(Z%a6;YT3&Ja~cQh%4Pm=TmA6%*Iibb$<47vRUWXcm(Y5)ZkxaH%Dx5qo3}9oO^qd$4W)ocB@U>!z!S)2IpuL8E`r(IHL{uE9g!fN*ePfwL-}>iKvQoqV4}KPJ(~BzkqKUZ3ave33*Mj<&w+@h3DxLrA+tzAiQ5Tv$DffP=!`99v}BDlv;Jgpv&ud7HdT#>-B?-pDoQHq^X|u&Z>zrirR=I=lP^8$1N`B~+hz*#PnoZfg`{FEMy3%8m@AYzE~EUeBAnooiI5DL=tW4SHLLmM&psH@)uL&Ux-y3%4vE!bp4s2*deC>|(5t=+vqL@YqWI+Gw<(arDf~T6qLV2-;lt^~J0}aA?aQ3jco^dJF#V6;xs{>QeR4IG&2iM{<5eDUL%_>tl^4KidYpy6trtpN3jjeQ`vqg}y?-6ZNux?s$JWZS~rgG79Xc1OJ4J0D8CIWz%-18;s6#}c}PH}Ov?+;D|Q_^Q7p=l$YuNc$Gmz94HF8<5a(iyDwXP0@aJxoRpNq=~x|#d=#MOaxV(p7@iDeLYFjG>M*{0wJO%56stxr<(ndYNZd*Lb))oTCnsMa;8X)A$R_DD9*&Pa!ncKJogv;kqmZiMp_#{gm`LF8{R!_W@u7UYcV?JhH3m2F)v^&7x~Z3YjLvEEo5F(;09Duysx{2Z1uQwilb_a@v)^TA;eOs>KdXU9M4bbhlLjyA8Q)n`OL5ltQX;b6D8oJCI=12jW(XVI7YoOU5XW=+#PW_LBQM^RQ})*l^i4n{_v%-spOhlj9`9|8Z|mU1}wSx%-LRB_em#<)+yHkLJ@Wp}!#wcaT;IGG(aJn+sC9pPjwgbY)=zfIW{Buc!K9TiaC!{qD=}VXgW3{5?A;Ek14s!OB165Pm|708sPI5*WX}i8|O%i)V-i0+Q6t-Qht<>=<&*d=ld0L~n)LroAH67c7)UFXNnN1r`=fK`O}GGTu>1HL>;*wRJ#{Z6%as-)^e%Ha>_?4o{%~w^svL+#ZvOElCONk5OI_TL4EUK}|@axpTEy3i}CbrCle4{;!c9O_fCnEk=e3Nt&^6B`i?UGQS7Of>-|If@O-jLGzGdTG9q5vO=vK~2*I458j3K}`ih1%*^Z;y#vTlM#B|A8Lhps`?&rW8E#qqhx^Y5)z}3OJ3nL|z420ib}x=fH<7n_Ku7X*+LToDJP?>;}|;xq_Y}SKqWiOmmPfC8<)Y)s68N18NVCDrxQRW|b|<9JLzOdwiEkT^6AC(eHk*Ke7%WyU}#*2&66x=8XY0q#|L78-#qD<*~o{{7+)3r8AGV>hJjyW<0CQ%M7;R4N3<-oG(v_)pR1&mB0NWH+~`HjLc6$g+mOE`N8QHF4Gk*6!uE2S)j;`C<5~w%VI!W&6_~o8^5l*UI4;Ye2o(8LELjJHWmCQV7{(NNdgRS?sm5tBK*AN$Tu@^7`a2FONeT#yZ#R~_<4tjF$#j@tO9Qb(k0*6i4}GEpxzWWwc9Ot1+{zs@$FT-DT7rVKxg0ve)vkk(-`TO5sFeo)!k7xRUj*M(bU<^=@Rno6&kVZawVhHA?}`sO&>FanD%0%U6#BfxW*O2XW&AxnUp%V$&udeFl50{RVaia^1iw_lx7`?a>rTK)ytN30QmBS6x+8j14)u>2hDqBAA<@kwWhe%GYlP<@EevN=;0MkZ3Eg+^h_Wu*=dqUrPWclN}a(5{A{>3y5uN?Sn?J$OjEd;y{Xijh-s_Q5>pb60IsUbsWjuYthG;-Q=+OB2?+J9|20L5a-s_LS2a5i%UT>1NC7s5>Mif^!3Ht2FPD_9Whe!%8#YJcp7b$kMPU;<6xo}qmFh$nKxSAxngllYGs@%}%Lozo8|=dXx=nveyK<~VJ{vmBjolMs>uFLP5wNH;lv%XXg?wYqvjr<)OM^P5@Mmk6t>TWVaTf56@X9NuRhGsVN}DW5lvfKG&=LlXnpR&=t?UQHk)Gev_K>>T=(a@)-KKumg<@=GDLY-I=giV|db-TnHv=EFEE|{vAui8s9j48Z`KC_glyCk=?(u*NDQWAKC%-jAd-lM{JO4HPgA!_vy6CwNnRk`p#qX~^Q5uX|!9|hSi#wr$g93ZiEv8KNr)rH)a326qw;zV~MkwVqx_2eH_YaX-T~Ls0l;lxK?>8{v*@Dpyq5Z!Fw&R&IGj=dO4`B%(tuWI}tbuc{a}4HDtb%$I+g?iYt#VTWlB{t6<9aB7LIn7r5DweKaY#Z(i^MSSRJEGR6u%YPn%Kn-yDGdWX1aP>0RJ{$#G1=+8Mh}umgab#e>5Ukq8oH!g##Mo(|dhV8>$?@ae4so4meiMNW?J$fQ0`C5gZVE%eEp(ut=kKD8R_ii~Oc^K$1*BqDX=cCKrH-WxH+xC=w$&e8`3dNU&Pt(Ey#iDY1hXjl@ie2@xnZ9IsHkjeDWy+J{2~2LorK>m}kaUi(-qTQBo@xu6gmz3}$#=yK#CkQJpiPL3n;+dBF~BHke8mPk_LLuSG6eBtSzc7adiT5}C#qFIMR`|LmvD}oy9FcI+FoxmY2z&P=CR#x>9{h{2}u&{kNE<{w@#ycho@Ch;-UyeN(*HEhKYzYT@hBz={Kf6};GvT=usx98Nt1cU`on2?=p?+CVM$Gk|jnA=d-t0bXD7(wJp;&m}cKz)mhw{Op+i<+G$H7(CN~uww07QEd6LrA?+eB#BdRg_cjlB-8xsh+A(N~5^+%K`Y1=YjID5?6@&FXMdorPLE$Hpk_4H4jE`#p(SUK9_6_j;(zRX;aw>QU9Av6D(nl(KP1=L5ES2{@2Bt(;AQ|^ncX5xB%k|Xg@zcZTkAN-s=UNL|B*PUI2HbY6Y(mj&Q(sMqaf$LZP~E#~#nRg?nL05#?_Q7{5FSJYO_djUE8J-*x&=dFqq4YU+W;I1ORij%DE9%o3u|AYXo6~p`_!BsXZ-*d-M~;|$rE$^Vb`lQS7D*pX;N0l)$|Jg{_pM8{r~KrF%kai#|{iT;=V@SX7?nwoe9&6&}9!~ho=2PBWh5NG*vITA-a_L=G0NP*CXfnGUph_Zn&P@p$a2-L{Vu#PnBN(ykwyNu0A7+!8cmth**G3d(ao~eQBF@~(Ho6Y4S0zpV1n(`w*AoKizM(Cb&Auz)tNENo90q?v-S7|6y^c!wk7$7w*OAXG4FQascHrbiPyYl`G)r`G*Z&~Ubt>ShO;?a=KR@3+8b*l{ydjZC9J5!IQQ-m_i-4Pl^=vF~Qy^e2v4-wQ-f0$wUgOYkNoha%v(8sd?Q4NQy@XPJ!AoQk6`!L44?T+Q$g5F0o$iNT&d#}MMG%lNc#IeJGOG8&o0JcuYrOz872-t>&Jn#Asuz}uDik^`E;$^d1mp$+tsAR^Y4$j7Dw`}_8wyiL;3x{XE|LsxK%^N&v@K|Bj2v(wY|2}$*3m9&Ooi`L2^iHHGqEAjdIyxmeRp{Mt71Ye569Jo1Wc)f8H@*eh5plcZl*CT$W{b|(J=eO~-!^1$XaO^#;vU0G*p>o29^w)C-Eo9>_sCDg~j@&2T-odU)%FOgubrt>O|e#nfyAte1NfFaSZYY10B%e2LM-S?rDC0mZRdSI0tgn8~x(k>GvB!l1w&-T@h(SYHUNnzkKBEm9GHrla6asboJAtrtfJ*bK?Rhisfv>l8Xg~W343x9p+QqV!4S&A@&8pXP^6R&$%-t1}2C$$@zRS{qB6V*iVd}2nM#}v31>mU+X1v7N$Qbsfc*>j3wdr13ETA2W<~<6^NqYCmK(^U=;nu^gbsa2J4HAB-T+X6l89=-Vuv~o^bX6D1{06A;)qir%n-Xs|#323q(noddgUD_^6l7%0M%qKO@{;lG}J3Bn<}ikllLs%`BN$(Rr3D4o|JnSoThF!WP`M_f{OZC_G_w{XZEG{$|zMiF=JH@H-a#mJs6PB(oGJ;`gvEKOfg~BO;!v_z1aW;pt*3p6GzQf~dXP*ec^V!6rN;xc9rlYo4ZI2`-LY9oU!A==q=hQp#kSc6?LxPpyJpk_B(h@BehIC_&lZDp@At4cEg6m(~5te?l`%!?VTo9c5j#^Pq1;Fws)PkUHR_ii+68d{6e{&Lwc@XN|^P&Sx*lcEipCx8VSWQdC^5|f5Ei;O&eV)WmD><6R94`lCPr5$fK7O1xYkE5cwlUqWKB8Ce2+jB}-9V;6Zv$9*zH#u6_2?3qO+#yrtXB{*UEE>{07zz5g`X7-nN>YLrwmHDN?9#G=q{2Zr2bnyIy~v{W<@ILJ{7_~VZ2B>M@n4A(dG4e9d`rB#G11%%Av=js3c!z8d6PFW@NF4v13IohKjY!ckN&3P7w#t$E}WvjgH>?T0)K{7fFmPZy(hak_F)jEmBzHUcKB$3{k8^SkP(9CWwFRc(L1X99z=Oq~-<N=SD%>agy9iNVz{+;dImGb}#Mr#Ocl_ynrrd#`7(qR*r=jwnbjeR6QE`Lq1?!6g1uEe_;RTR8L-j{RBJG@L{>wN^dMVS&rX!p6BXZDVc>GD%ySRAzxbys)uUkRaGx6hwf2f4k%RH41rWT+f7r5(8v79WlSp3RBFn|^rRJ*JPvP~XRl`$58ph!fC@rE;t;S%*to(r?fDUDYsM8n{3ch2;HSv&;I*`=tYMBB=W^Phc?wiVWw{dg##(_aj6pnFz$pdm0zKOFKL!c0h%wpw0p5n|l9GMjDf4bKtMcs{Cn!jo*Ez8g;$5K(E>HErX{u=_ZM+ro`BukYhPunkFjjF22U^(C;8c1&vzgs8O#K16$#tmD;ESGjm~5_UBcSTra8jwy?&f%#*I%2Y}vIDoC%Q>+;aYw8olCp^^}p+*9E^x^7F%|BT7TV1E?NDgadAk+}H^qmVe)7`b|MybRPq&Xhy8Im)g}juJdNKOSBg_SDTI-E=f$Lp|BLl#7TCStp$4ntDxuuAikx0hou7Z?&65a>K1Zd;Juwkf5Co(i^f02n3RA+AYAYY$$#WyQ2xmnK;FVY$aiWpWXUedMUeow0}A4TYbj>x?Xsa|D=6G}uxp<;iQSfuvlOD-Zz1AlXZYWPFtU3-xpwP#Bt1W6<y?^cWEv5r*fN)zuDzEO-jC8wMvra@5{f?Ph&%?<5G0TzR&D3fCl$od?&PY@G({c?ZYb@N@?W&;uLa8O;JyfA}aovjHY{x10^_WGo|aZi0-U$l1;*obSW^fKy!3k_VQtDrvplml4GTZNwAzarbR)18a&s@pI;Oeuk5mp5zE=E!cc4$qB%}X);4%pFf8NYDd{j=_oC00aHOKabj&#(=kTL{%v3d-g(nF~coUdfB~uRF3r()d31cS2Gipuzv^@iQc^jf#PpkP7OSL91X;Vr-vlj}ZR0N)WFK{Nre8}r)kss847v_)@e~27}qlcepMbt2p`y_e|gbqa&c}vVixm?N&tYWL-7M^a_>A*yciuw+A88fzP9d9oDisCXhOXKZDMLM|=lj!eOGy8gb>>nF@#Xej1^{}CofegV?YPe8V{SFtkM+xQa7rN3vm1&zz$eB)^@;%hK`WM_#i*8TjIqaO4TDq8z4=9x2>u7XvsO{^DtI23#o_)!@6a;-rh4TGkzlilcq%)?~2_IGRc}x_@LJky3YHiOl4<(mLy5bBp13@fn9eh^s0ml>N`RN|JQHbrThU5EU>p;##N*`yxv84GbDjiZnWd57}+gqFr5#N%&2jVF&aznSh4(EG_LZcP3H+)3s8>0rr%-pC$w39ZV$*uX1rMXAWzEBf;$J9Fl6}WBiJVggImaDDt@z`l;VNE>0BJ)lM@d-qC#e*{00JV@EcPOkUTLA(*b)feL&JM=XwzQ4g{ojL5F~!cW8Ozq+~+g$VXt?)pMz6jhn`;7lY{=oZL0h}}UcRH2>=JfZ#Mlh_d-Gxr`8Z;Y{mDetS~3l>!SER;RU(GC4nDdF2pcO*p7D6B=RpB(QEcPJ8S5)Ccf#7I?Fcjd=wjIw?CZTZkm5DeG~qS9xg=a(x&oB+r@3u^c2U*%Q&!Hw&Nk3wLd9{d~oM52;hR}l#D>!LQT0(Fz%oH@bU2Hlb$l8zxr8lD`#h_@sxZf&xmzcx8)T03RJ%We{)lHXdS&oQ!_NQEZS1ptASOhT4V=G8CL=v4dt1|Hp!{}?952Lxe@~nX#*zMcpaTFE;7=0XT0zYaE=Fc(;uK=5r+zMPKoq!Jd9R+6=^4JtFoy04)GjjE1xf97I@(AHg%C>q|dU1Gbf9qf?o1G#O6?Rb=-i8k_dh+c#fd+b=rc6b`NgdldR?r+A_Z!->`>j-Kv)TYjXxH^n-X_4n}E7;pMBpl)oNF)bE^&Yg)T8u<03XnL`|$Q6J*=p>u?EJNz^jPLl*-HJDs>*>!RS#sg+Fe7iN`YfRaUbV9(}wd(j15o#WrhzzOd=GtHvM&CvBnd33RS=N96w_uF*>`W*`1fd4FF>*|(KwC;6OJoqM$tbRY%O<%D9l?TfsZv`@Cj%KMz>N8TRH)+aj~ombA^p?NP)Pw&e=;5~{4)6`(RIyrOT?yKpW^!BPMHpN5*qJfqp<0ypYEvWWIEJ<03kEe8yQxvN##aT3Yu3@f5*1%9-Kk~3WZp?`OYMx=B4b#FZyAHt`N=+7ja7L96H^yUzH4Yi1embDsC&Dsc@0b156~7dYniPJ9V&z|zI0~5IZ2mZ-HXY6)3h%sJ;mF1LJnxw;#__A@x1{O&*2{3O3ESu!6P3dVdgHGq?&SF0^wrinYK@8ra~=USovn|7hCaelOM0}m?*+{=EG;my64?c&qdmgtuj5`PN+L6a)peDo_tx(N7x_Yrkqe0P^UpxJDPy@hh+E%&5TFksmrNqo9K?b`p23(ZAKZo9nQRS@j~@L7|>J;=&u9t5P`{ekS>xQB7yq>FRx*Y5qWk;{G*0YLRHnMp<{}TkYXbQb?Ck)OtT6M3vWo?dpVpP^EvQNm`xKm;tGXaVLv4*3=`3;qF1lptE&oAldXEtneE`DJhy#+fJvml?jG8F9J=8U=-#Guyu2u$SxWtZP;ePBpl%Rb!ZbSGixiwY1coDbcY*`6QdH=%Y+blX?5%vS>0VG2r(!5R!EV~Xtl}6FSj!oG_=R5nWu1Qb_1$D^UF!HQl~;zsb~MPHdqZ${_lfX~CU2Yia@^lRCr{YT>p?Zs@s?Iiy}`vAuNU&h*xti69Jo6lRSy(nAvALgKs(e>6Sk!|pNJVU0^`Hty0Nspwvu098D=`FV&b~jn;*kzEk}40xAWGH8S1A6v$}97^zJidKOZ=7>^!5}gZv(NNfXilgW5~9uzPHkaN_r5Lt~+F!)b!;8+`Dfc4x;i+D`@%+F$3#h0uStcfRWZ`><*|_~?h6c1fPkB-8eM&(}LB_Eo4WoZtnZ=+Qaq1Nb&wwkwn0upuv}k4EowWa6^X53I1V*|C{K2H-7&U1;2SFEU@Ym#?#{957y`hfR_oy?Ev$QDdasRl>1K4q~KOIpm+dIP43kGkqHp$G>S#;<^4sz=roDiM&-#ztc|K7O2v*|JvxHfou-#)uvnqJh+O5AyLnAr0**`kX?mE*rOjLJjLXWaiCqpHG~=)w4>tFtrZtC!#-O0<$_;MudUpi_9{R*ZgzC3+qnY_XU`!&1s&y-84Z+)UdsImpD2=K8_qQ8I~qhk#cG^|dg2i-KkMJ~+oKw2XnUC_iWVVdiyZHpBuOmW#g`+JbR_BS1N3RZeUD>%iA<)aH&v$9>c$eoc|6?uqs&&;^Ai`~K@fVXth5liQ`_r#+u8?-!W5X9t5nIMdYf`ACb#2%11ffqzq!lYdK_xUrf=*}>00Uzz+qC3kKoL07fp*A-vhU0%-uqsNXe{cMu^&e;!=3;WND_&sW6Rh+w)%j&qK-jwSQA02Vs#CD_oxWDK5t>fgP*H{Tr2X>KloCWxdMkx>l)<1ZWP&H*0b_X_mC+}7->z`cDWF~_!4xt3|MzGTH7Fq>sgn9ZG2}G*b6KuU*|O^S2&mA(saXTAUjbh>o$}zNNvc_kvk`hOJ+=YWfL~!@(O3;2Pz7oDrW9APSz$xu=HCNVWNYK5LZ7uiQqJ2arhU~^+t$gH?uv5)f-(7&hLKcB61`Sf@Yw^h+E91SUNNJ_cEN=ZviqEaKOgopvn*~#H)!mtO^Ar94&=@GfMpO7yLr~Dkr%O{4y$e_`}sj=s&G;kV#+3JJpj3A{^S^vb>hfua_0VR1H)>xP;+{#noT&wLla!}^QnRs%uS@}>nBy=(?EGCmn+!M7;6TN5lIw(+Oep?GVz^09?!HxZjWVt4ioygOyJ&OF3C3Lk3alMC#A`lZetleMv>CR0qsg0)dtQ5X6^=?ar!R8_z+Vu@!!L(!N!vv92)-p+#q&!YIwNU#LjLInCp~jKcFlOv|Q7TS7>$G$yY+?U4&+9o5RG{GpJM0iO6JS@guRbYXnZu)UN@B9vd(0$Ha#v5qD-iC4&Z6+e$L3nslD+!rBu;{YnC932mI2g4=+5xQXB)xw*Lm3r!1<9ZuYoo=w0p0+(~BsNW*zzD^r3O*Sl_}g>-TE-=>9Ycvf-p-hnc?haj`fnM7m*sBA3N_S{V3BBbeafO~LE1`K^Hi=Pjl?GL*g_M(CSp)ssqOIW)=2~L9MUzI}H{TlSqvA<}h=DX;~iP*&OdLl(5vYF$-!(0Z6PqynM_E1W)R-fYl19Nnm7>)aocr`M$&|x-7x0x6$(9tpF%eUB`05>kz4gYa2&I7Q_6lHiDx$KX%R#Ygro=`PYJw^6x#}c@u%-u~E18V(YO#mtGioj{yE8GEN7+?JreHS%t3lpp;$L+H(Zuk(u=8LNBuq2;!v^G&?*TZ1CVhGg3{5X)`sZiD(&T2Qe?AVDVm@Xb$MQMgYz|`9tpY>2k&7R}G%ZuJfwrLs#!Wmq$%)VL;%n_`FwP@d-QTXV{Xp#03hmXV{{L}M(Ha+t#?keEm?w?L(_fPh1hy2q~hhwv5x^C97F-%WKXBIXu)GH~|5D!t_sTamdIBLp#1<>z&l%@seV&*FUQeJN=wAa0O|5LPRnk!D=2w-5hV5WgoBLYm0h!0ra8({;UncC^yma8Z~J7CPT;ctb34mzG^j=nb*DMr2l&V_t<;{6x!o2 bool: raw = os.getenv(name) if raw is None: return default value = raw.strip().lower() if value in {"1", "true", "yes", "on"}: return True if value in {"0", "false", "no", "off"}: return False raise RuntimeError(f"{name} must be one of 1/0, true/false, yes/no, on/off.") def _embedded_runtime_bytes(name: str) -> bytes: try: item = VBPT_RUNTIME_FILES[name] except KeyError as exc: raise RuntimeError(f"Embedded VBPT runtime is missing {name!r}.") from exc try: payload = zlib.decompress(base64.b85decode(item["compressed_b85"].encode("ascii"))) except Exception as exc: raise RuntimeError(f"Embedded VBPT runtime payload is corrupt for {name}: {exc}") from exc digest = hashlib.sha256(payload).hexdigest() if digest != item["sha256"] or len(payload) != int(item["size"]): raise RuntimeError( f"Embedded VBPT runtime verification failed for {name}: " f"sha256={digest}, size={len(payload)}" ) return payload def _runtime_manifest() -> dict[str, str]: return {name: str(item["sha256"]) for name, item in VBPT_RUNTIME_FILES.items()} def _safe_component(value: str) -> str: cleaned = "".join(ch if ch.isalnum() or ch in "._-" else "_" for ch in value) return cleaned[:96] or "default" def _link_or_copy(source: Path, target: Path) -> None: target.parent.mkdir(parents=True, exist_ok=True) if target.exists() or target.is_symlink(): if target.is_dir() and not target.is_symlink(): shutil.rmtree(target) else: target.unlink() try: target.symlink_to(source.resolve(), target_is_directory=source.is_dir()) return except OSError: pass if source.is_dir(): shutil.copytree(source, target, dirs_exist_ok=True) return try: os.link(source.resolve(), target) return except OSError: shutil.copy2(source, target) def _materialize_vbpt_overlay(repo_dir: Path, revision: Optional[str]) -> Path: """Create a local weight/tokenizer overlay with verified embedded GDN code.""" root = Path(os.getenv(VBPT_OVERLAY_DIR_ENV, str(Path(tempfile.gettempdir()) / "vbpt_gdn_overlay"))) revision_key = _safe_component(revision or "main") runtime_key = hashlib.sha256(json.dumps(_runtime_manifest(), sort_keys=True).encode("utf-8")).hexdigest()[:16] destination = root / f"vbpt-{revision_key}-{runtime_key}" manifest_path = destination / ".vbpt_runtime_manifest.json" expected_manifest = { "runtime_version": VBPT_RUNTIME_VERSION, "runtime_sha256": _runtime_manifest(), "source_repo": VBPT_REPO_ID, "source_revision": revision, } if manifest_path.exists(): try: loaded_manifest = json.loads(manifest_path.read_text(encoding="utf-8")) if loaded_manifest == expected_manifest: return destination except Exception: pass staging = root / f".{destination.name}.staging-{os.getpid()}" if staging.exists(): shutil.rmtree(staging) staging.mkdir(parents=True, exist_ok=False) runtime_names = set(VBPT_RUNTIME_FILES) try: for source in repo_dir.rglob("*"): rel = source.relative_to(repo_dir) if rel.name in runtime_names or "__pycache__" in rel.parts: continue target = staging / rel if source.is_dir(): target.mkdir(parents=True, exist_ok=True) else: _link_or_copy(source, target) for name in sorted(runtime_names): (staging / name).write_bytes(_embedded_runtime_bytes(name)) manifest_path_staging = staging / ".vbpt_runtime_manifest.json" manifest_path_staging.write_text(json.dumps(expected_manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8") if destination.exists(): shutil.rmtree(destination) destination.parent.mkdir(parents=True, exist_ok=True) staging.replace(destination) except Exception: shutil.rmtree(staging, ignore_errors=True) raise return destination def _fla_available() -> tuple[bool, Optional[str]]: try: from fla.layers import GatedDeltaNet # noqa: F401 return True, None except Exception as exc: return False, f"{type(exc).__name__}: {exc}" def _resolve_vbpt_execution() -> tuple[str, torch.device, torch.dtype]: requested_backend = os.getenv(VBPT_RUNTIME_BACKEND_ENV, "auto").strip().lower() requested_device = os.getenv(VBPT_DEVICE_ENV, "auto").strip().lower() if requested_backend not in {"auto", "fla", "reference"}: raise RuntimeError(f"{VBPT_RUNTIME_BACKEND_ENV} must be auto, fla, or reference.") if requested_device not in {"auto", "cuda", "cpu"}: raise RuntimeError(f"{VBPT_DEVICE_ENV} must be auto, cuda, or cpu.") cuda_ready = torch.cuda.is_available() fla_ready, fla_error = _fla_available() if requested_backend == "fla": if not cuda_ready: raise RuntimeError("VBPT_RUNTIME_BACKEND=fla requires a CUDA-enabled PyTorch runtime.") if not fla_ready: raise RuntimeError( "VBPT_RUNTIME_BACKEND=fla requires flash-linear-attention[cuda]. " f"Import detail: {fla_error}" ) if requested_device == "cpu": raise RuntimeError("VBPT_RUNTIME_BACKEND=fla cannot run on VBPT_DEVICE=cpu.") dtype = torch.bfloat16 if torch.cuda.is_bf16_supported() else torch.float16 return "fla", torch.device("cuda"), dtype if requested_backend == "reference": if requested_device == "cuda": raise RuntimeError( "The verified VBPT reference runtime is intended for CPU. " "Use VBPT_RUNTIME_BACKEND=fla with CUDA for GPU inference." ) return "reference", torch.device("cpu"), torch.float32 # auto: choose fast FLA only when both CUDA and FLA are actually usable; # otherwise use the embedded exact-operation PyTorch reference on CPU. if requested_device == "cuda": if not cuda_ready: raise RuntimeError("VBPT_DEVICE=cuda was requested but torch.cuda.is_available() is False.") if not fla_ready: raise RuntimeError( "VBPT_DEVICE=cuda requires flash-linear-attention[cuda] for VBPT. " f"Import detail: {fla_error}" ) dtype = torch.bfloat16 if torch.cuda.is_bf16_supported() else torch.float16 return "fla", torch.device("cuda"), dtype if requested_device == "cpu": return "reference", torch.device("cpu"), torch.float32 if cuda_ready and fla_ready: dtype = torch.bfloat16 if torch.cuda.is_bf16_supported() else torch.float16 return "fla", torch.device("cuda"), dtype return "reference", torch.device("cpu"), torch.float32 def _download_vbpt_assets(repo_id: str, revision: Optional[str]) -> Path: # We need weights/tokenizer/config only. The public repository's custom # Python files are deliberately not used; the local overlay writes the # verified embedded runtime instead. return Path(snapshot_download( repo_id=repo_id, repo_type="model", revision=revision, local_files_only=_parse_bool_env(VBPT_LOCAL_FILES_ONLY_ENV, default=False), ignore_patterns=list(VBPT_RUNTIME_FILES), )) def _load_vbpt_with_embedded_runtime(repo_id: str) -> tuple[Any, Any, "VBPTDataLoader", str, str, str, Optional[str]]: revision = os.getenv(VBPT_REVISION_ENV) or None runtime_backend, device, dtype = _resolve_vbpt_execution() # The embedded runtime reads this setting during model construction. os.environ["GDN_GATED_DELTANET_BACKEND"] = runtime_backend source_dir = _download_vbpt_assets(repo_id, revision) overlay_dir = _materialize_vbpt_overlay(source_dir, revision) print( f"[vbpt-api] VBPT loader=embedded-overlay backend={runtime_backend} device={device} overlay={overlay_dir}", flush=True, ) model, tokenizer, loader = VBPTDataLoader.load_pretrained( str(overlay_dir), device=device, dtype=dtype, trust_remote_code=True, local_files_only=True, ) return model, tokenizer, loader, runtime_backend, str(device), str(dtype).replace("torch.", ""), revision @dataclass(frozen=True) class VBPTPreparedBatch: """Tokenized, left-padded batch ready for `model.generate`.""" inputs: Mapping[str, torch.Tensor] prompts: list[str] prompt_style: str prompt_token_count: int class VBPTDataLoader: """Model-specific prompt renderer and batch tokenizer for VBPT. The name is intentional: it owns data preparation for VBPT and can create batches, while model loading remains explicit and auditable. """ def __init__(self, tokenizer: Any, *, device: torch.device | str = "cpu") -> None: self.tokenizer = tokenizer self.device = torch.device(device) self._repair_special_tokens() # Left padding keeps every generated continuation after the common, # padded input width. It also preserves the newest user turns when # context truncation is enabled. self.tokenizer.padding_side = "left" self.tokenizer.truncation_side = "left" @classmethod def load_pretrained( cls, repo_id: str = VBPT_REPO_ID, *, device: torch.device | str = "cpu", dtype: torch.dtype = torch.float32, trust_remote_code: bool = True, local_files_only: bool = False, ) -> tuple[Any, Any, "VBPTDataLoader"]: """Load VBPT from a local verified overlay with the benchmark dtype fallback.""" from transformers import AutoModelForCausalLM, AutoTokenizer tokenizer = AutoTokenizer.from_pretrained( repo_id, trust_remote_code=trust_remote_code, local_files_only=local_files_only, ) # `dtype=` is the newer Transformers spelling used by the supplied # benchmark. Older installations require `torch_dtype=`. common = { "trust_remote_code": trust_remote_code, "low_cpu_mem_usage": True, "local_files_only": local_files_only, } try: model = AutoModelForCausalLM.from_pretrained( repo_id, dtype=dtype, **common, ) except TypeError as exc: text = str(exc).lower() if "dtype" not in text and "unexpected keyword" not in text: raise model = AutoModelForCausalLM.from_pretrained( repo_id, torch_dtype=dtype, **common, ) model.to(device) model.eval() loader = cls(tokenizer, device=device) loader.apply_generation_token_ids(model) return model, tokenizer, loader def _repair_special_tokens(self) -> None: """Guarantee padding works without adding/resizing vocabulary entries.""" if self.tokenizer.pad_token_id is None: if self.tokenizer.eos_token_id is None or self.tokenizer.eos_token is None: raise RuntimeError( "VBPT tokenizer has no pad token and no eos token; refusing to " "invent a vocabulary token because that would require embedding resize." ) self.tokenizer.pad_token = self.tokenizer.eos_token def apply_generation_token_ids(self, model: Any) -> None: """Synchronize tokenizer pad/eos IDs into config and GenerationConfig.""" pad_id = int(self.tokenizer.pad_token_id) eos_id = self.tokenizer.eos_token_id for config in (getattr(model, "config", None), getattr(model, "generation_config", None)): if config is None: continue if getattr(config, "pad_token_id", None) is None: config.pad_token_id = pad_id if eos_id is not None and getattr(config, "eos_token_id", None) is None: config.eos_token_id = eos_id if getattr(model, "generation_config", None) is not None: model.generation_config.use_cache = True @staticmethod def _normalize_messages(messages: Sequence[Mapping[str, str]]) -> list[dict[str, str]]: normalized: list[dict[str, str]] = [] allowed_roles = {"system", "user", "assistant"} for item in messages: role = str(item.get("role", "")).strip().lower() content = str(item.get("content", "")) if role not in allowed_roles: raise ValueError(f"VBPT does not support chat role {role!r}.") if not content.strip(): continue normalized.append({"role": role, "content": content}) if not normalized: raise ValueError("VBPT requires at least one non-empty message.") return normalized @staticmethod def _raw_benchmark_compatible_prompt(messages: Sequence[Mapping[str, str]]) -> str: """Fallback compatible with the benchmark's raw-prompt route. For the common one-turn request, this is simply system text followed by the user text: no unverified ChatML/Qwen markers are injected. Multi-turn histories use minimal Vietnamese role headers only to keep turns distinct. """ normalized = VBPTDataLoader._normalize_messages(messages) non_system = [item for item in normalized if item["role"] != "system"] systems = [item["content"].strip() for item in normalized if item["role"] == "system"] if len(non_system) <= 1: sections = [part for part in systems + [item["content"].strip() for item in non_system] if part] return "\n\n".join(sections) role_names = { "system": "Hệ thống", "user": "Người dùng", "assistant": "Trợ lý", } chunks = [ f"{role_names[item['role']]}:\n{item['content'].strip()}" for item in normalized ] return "\n\n".join(chunks) def render_messages(self, messages: Sequence[Mapping[str, str]]) -> tuple[str, str]: """Render one request using native template first, raw fallback second.""" normalized = self._normalize_messages(messages) template = getattr(self.tokenizer, "chat_template", None) if template: try: rendered = self.tokenizer.apply_chat_template( normalized, tokenize=False, add_generation_prompt=True, ) if isinstance(rendered, str) and rendered.strip(): return rendered, "native_chat_template" except Exception: # The raw fallback is deliberate: the supplied benchmark also # ignores broken templates and preserves the original prompt. pass return self._raw_benchmark_compatible_prompt(normalized), "raw_benchmark_compatible" def prepare_batch( self, messages_batch: Iterable[Sequence[Mapping[str, str]]], *, context_tokens: int, ) -> VBPTPreparedBatch: """Render, tokenize, left-pad, truncate, and move a VBPT batch to device.""" if context_tokens < 1: raise ValueError("context_tokens must be positive.") rendered = [self.render_messages(messages) for messages in messages_batch] if not rendered: raise ValueError("VBPT batch must contain at least one request.") prompts, styles = zip(*rendered) encoded = self.tokenizer( list(prompts), return_tensors="pt", padding=True, truncation=True, max_length=context_tokens, return_attention_mask=True, ) device_inputs = { key: value.to(self.device) for key, value in encoded.items() if isinstance(value, torch.Tensor) } return VBPTPreparedBatch( inputs=device_inputs, prompts=list(prompts), prompt_style=styles[0] if len(set(styles)) == 1 else "mixed", prompt_token_count=int(device_inputs["input_ids"].shape[1]), ) def decode_new_tokens( self, output_ids: torch.Tensor, *, padded_prompt_tokens: int, ) -> list[str]: """Decode only model continuations, never the padded prompt prefix.""" if output_ids.ndim != 2: raise RuntimeError("Expected generated token IDs with shape [batch, sequence].") if output_ids.shape[1] < padded_prompt_tokens: raise RuntimeError("Generated sequence is shorter than the prepared prompt.") new_ids = output_ids[:, padded_prompt_tokens:] return [ self.tokenizer.decode(row, skip_special_tokens=True).strip() for row in new_ids ] # Public clients can only request keys in this catalog. # Add models here instead of accepting arbitrary Hugging Face repo IDs. # # GGUF quant policy when gguf_quant="auto": # <= 1B -> Q6 first # 1B–4B -> Q4 first # > 4B -> Q3 first # The loader falls back to the next available quantization only for "auto". # Explicit choices such as gguf_quant="q3" remain strict. # # Do not set chat_format unless a particular GGUF has been tested and needs an # explicit override. Modern GGUFs normally include tokenizer.chat_template; # leaving chat_format unset lets llama-cpp-python use that model metadata. MODEL_CATALOG = { "jackrong-qwen35-0.8b-gguf": { "backend": "gguf", "repo": "Jackrong/Qwen3.5-0.8B-Claude-4.6-Opus-Reasoning-Distilled-GGUF", "params_b": 0.8, "label": "Jackrong Qwen3.5 0.8B GGUF (auto Q6)", }, "jackrong-qwen35-9b-gguf": { "backend": "gguf", "repo": "Jackrong/Qwen3.5-9B-Claude-4.6-Opus-Reasoning-Distilled-GGUF", "params_b": 9.0, "label": "Jackrong Qwen3.5 9B GGUF (auto Q3)", }, "qwen3-vl-2b-gguf-text": { "backend": "gguf", "repo": "Qwen/Qwen3-VL-2B-Instruct-GGUF", "params_b": 2.0, "label": "Qwen3-VL 2B GGUF text mode (auto Q4)", "note": ( "Text mode only. Image understanding requires the compatible " "multimodal projector/mmproj and an extra vision endpoint." ), }, "hauhau-qwen35-2b-transformers": { "backend": "transformers", "repo": "HauhauCS/Qwen3.5-2B-Uncensored-HauhauCS-Aggressive", "label": "HauhauCS Qwen3.5 2B Transformers FP32 CPU", }, "mihai-qwen3-0.6b-transformers": { "backend": "transformers", "repo": "MihaiPopa-1/Qwen-3-0.6B-Claude-4.7-Opus-Distilled", "label": "MihaiPopa Qwen3 0.6B Transformers FP32 CPU", }, "bachvnju-vbpt-1-0.5B": { "backend": "transformers", "repo": "bachvnju/vbpt-1-0.5B", "loader": "vbpt", "label": "bachvnju/vbpt-1-0.5B Transformers F32 CPU", "note": ( "Uses the dedicated VBPT dataloader: native tokenizer chat template " "when valid, otherwise benchmark-compatible raw prompt batching." ), }, # Example: enable only after replacing the repo with a real pre-exported # ONNX causal language model repository. # # "my-onnx-model": { # "backend": "onnx", # "repo": "YOUR-ONNX-EXPORT-REPO", # "label": "ONNX Runtime CPU model", # }, } if DEFAULT_MODEL not in MODEL_CATALOG: raise RuntimeError( f"DEFAULT_MODEL={DEFAULT_MODEL!r} is not in MODEL_CATALOG: " f"{', '.join(MODEL_CATALOG)}" ) model = None tokenizer = None loaded_model_key: Optional[str] = None loaded_backend: Optional[str] = None loaded_context_tokens: Optional[int] = None # Non-None only for model families with a dedicated inference dataloader. model_dataloader: Optional[VBPTDataLoader] = None loaded_prompt_adapter: Optional[str] = None loaded_vbpt_runtime_backend: Optional[str] = None loaded_vbpt_device: Optional[str] = None loaded_vbpt_dtype: Optional[str] = None loaded_vbpt_revision: Optional[str] = None # CPU inference is serialized. Multiple concurrent generation jobs typically make # a small CPU Space slower and can trigger OOM. engine_lock = threading.RLock() class ChatMessage(BaseModel): role: Literal["user", "assistant"] content: str = Field(..., min_length=1, max_length=12000) def _json_depth(value: Any, depth: int = 0) -> int: if isinstance(value, dict): if not value: return depth + 1 return max(_json_depth(item, depth + 1) for item in value.values()) if isinstance(value, list): if not value: return depth + 1 return max(_json_depth(item, depth + 1) for item in value) return depth def validate_response_schema(schema: dict[str, Any]) -> None: """Reject pathological client schemas before llama.cpp compiles a grammar.""" try: encoded = json.dumps(schema, ensure_ascii=False, separators=(",", ":")) except (TypeError, ValueError) as exc: raise ValueError(f"response_schema must be JSON-serializable: {exc}") from exc if len(encoded.encode("utf-8")) > MAX_JSON_SCHEMA_BYTES: raise ValueError( f"response_schema exceeds MAX_JSON_SCHEMA_BYTES={MAX_JSON_SCHEMA_BYTES}" ) if _json_depth(schema) > MAX_JSON_SCHEMA_DEPTH: raise ValueError( f"response_schema nesting exceeds MAX_JSON_SCHEMA_DEPTH={MAX_JSON_SCHEMA_DEPTH}" ) try: Draft202012Validator.check_schema(schema) except SchemaError as exc: raise ValueError(f"Invalid Draft 2020-12 JSON Schema: {exc.message}") from exc class GenerateRequest(BaseModel): model: str = DEFAULT_MODEL # Either `prompt` or `messages` is required. prompt: Optional[str] = Field(default=None, max_length=12000) messages: list[ChatMessage] = Field(default_factory=list, max_length=40) system: str = Field( default="Bạn là trợ lý AI hữu ích. Trả lời rõ ràng bằng tiếng Việt.", max_length=4000, ) # Context budget for this request. GGUF models reload when this changes, # because n_ctx is set during llama.cpp model initialization. context_tokens: int = Field(default=2048, ge=256, le=MAX_CONTEXT_TOKENS) max_new_tokens: int = Field(default=256, ge=1, le=MAX_NEW_TOKENS) temperature: float = Field(default=0.6, ge=0.0, le=2.0) top_p: float = Field(default=0.95, ge=0.05, le=1.0) top_k: int = Field(default=20, ge=0, le=200) repetition_penalty: float = Field(default=1.12, ge=1.0, le=2.0) use_cache: bool = True seed: Optional[int] = Field(default=None, ge=0, le=2_147_483_647) max_time: Optional[float] = Field(default=300.0, ge=1.0, le=900.0) # JSON output controls. response_schema enables llama.cpp grammar-based # JSON Schema constraints and a second server-side validation pass. json_mode: bool = False response_schema: Optional[dict[str, Any]] = None # GGUF only. "auto" => Q6 for <=1B, Q4 for 1B–4B, Q3 for >4B. # Exact choice depends on which files actually exist in the repository. gguf_quant: Literal["auto", "q3", "q4", "q5", "q6", "q8"] = "auto" @model_validator(mode="after") def validate_input(self): if not self.prompt and not self.messages: raise ValueError("Provide either prompt or messages.") if self.response_schema is not None: validate_response_schema(self.response_schema) return self def unload_model() -> None: global model, tokenizer, loaded_model_key, loaded_backend, loaded_context_tokens global model_dataloader, loaded_prompt_adapter global loaded_vbpt_runtime_backend, loaded_vbpt_device, loaded_vbpt_dtype, loaded_vbpt_revision model = None tokenizer = None model_dataloader = None loaded_prompt_adapter = None loaded_model_key = None loaded_backend = None loaded_context_tokens = None loaded_vbpt_runtime_backend = None loaded_vbpt_device = None loaded_vbpt_dtype = None loaded_vbpt_revision = None gc.collect() def quant_priorities(params_b: float, requested_quant: str) -> list[str]: groups = { "q3": ["q3_k_l", "q3_k_m", "q3_k_s", "q3_0"], "q4": ["q4_k_m", "q4_k_s", "q4_0"], "q5": ["q5_k_m", "q5_k_s", "q5_0"], "q6": ["q6_k", "q6_k_l", "q6_k_m"], "q8": ["q8_0"], } if requested_quant != "auto": return groups[requested_quant] if params_b <= 1.0: # Q3 is deliberately excluded for models at or below 4B. return groups["q6"] + groups["q5"] + groups["q4"] + groups["q8"] if params_b <= 4.0: # Q3 is deliberately excluded for models at or below 4B. return groups["q4"] + groups["q5"] + groups["q6"] + groups["q8"] return groups["q3"] + groups["q4"] + groups["q5"] + groups["q6"] + groups["q8"] def choose_gguf_file(repo_id: str, params_b: float, requested_quant: str) -> str: files = HfApi().list_repo_files(repo_id, repo_type="model") candidates = [ item for item in files if item.lower().endswith(".gguf") and "mmproj" not in item.lower() and "projector" not in item.lower() ] if not candidates: raise RuntimeError(f"No usable GGUF files found in {repo_id}") # Prefer a normal model file, not split model segments, when both exist. candidates.sort(key=lambda name: ("-00001-of-" in name.lower(), len(name), name.lower())) for needle in quant_priorities(params_b, requested_quant): for filename in candidates: if needle in filename.lower(): return filename available = ", ".join(candidates[:20]) raise RuntimeError( f"No GGUF matching requested quantization '{requested_quant}' in " f"{repo_id}. Available: {available}" ) def describe_chat_template(loaded: Any) -> str: metadata = getattr(loaded, "metadata", {}) or {} if metadata.get("tokenizer.chat_template"): return "gguf tokenizer.chat_template" return "llama-cpp-python default/fallback (GGUF has no tokenizer.chat_template)" def load_model( model_key: str, context_tokens: int, gguf_quant: str, ) -> None: global model, tokenizer, loaded_model_key, loaded_backend, loaded_context_tokens global model_dataloader, loaded_prompt_adapter global loaded_vbpt_runtime_backend, loaded_vbpt_device, loaded_vbpt_dtype, loaded_vbpt_revision if model_key not in MODEL_CATALOG: raise HTTPException( status_code=422, detail={ "error": "Unknown model key", "available_models": list(MODEL_CATALOG), }, ) spec = MODEL_CATALOG[model_key] backend = spec["backend"] if ( backend == "gguf" and gguf_quant == "q3" and float(spec["params_b"]) <= 4.0 ): raise HTTPException( status_code=422, detail="Q3 is reserved for GGUF models larger than 4B parameters.", ) # A llama.cpp model needs reload if n_ctx or quantization policy changes. same_gguf_options = ( backend != "gguf" or ( loaded_context_tokens == context_tokens and getattr(model, "_selected_quant_request", None) == gguf_quant ) ) if model is not None and loaded_model_key == model_key and same_gguf_options: return unload_model() try: if backend == "gguf": if Llama is None: raise RuntimeError( "GGUF backend requires llama-cpp-python. Install it or select " "the default bachvnju-vbpt-1-0.5B Transformers model." ) filename = choose_gguf_file( repo_id=spec["repo"], params_b=float(spec["params_b"]), requested_quant=gguf_quant, ) local_path = hf_hub_download( repo_id=spec["repo"], filename=filename, repo_type="model", ) llama_options = { "model_path": local_path, "n_ctx": context_tokens, "n_threads": CPU_THREADS, "n_threads_batch": CPU_THREADS, "n_batch": 512, "n_gpu_layers": 0, "verbose": False, } # Only catalog entries that explicitly provide a tested override use it. if spec.get("chat_format") is not None: llama_options["chat_format"] = spec["chat_format"] loaded = Llama(**llama_options) # Store runtime details without relying on private llama-cpp internals. loaded._selected_quant_request = gguf_quant loaded._selected_gguf_filename = filename loaded._chat_template_source = describe_chat_template(loaded) model = loaded tokenizer = None elif backend == "transformers": if spec.get("loader") == "vbpt": # VBPT uses its own loader instead of the generic Transformers path. # It follows the supplied benchmark's dtype compatibility fallback, # pad-token repair, template handling, and left-padded batching. ( loaded, loaded_tokenizer, loaded_dataloader, runtime_backend, runtime_device, runtime_dtype, runtime_revision, ) = _load_vbpt_with_embedded_runtime(spec["repo"]) model = loaded tokenizer = loaded_tokenizer model_dataloader = loaded_dataloader loaded_prompt_adapter = "vbpt" loaded_vbpt_runtime_backend = runtime_backend loaded_vbpt_device = runtime_device loaded_vbpt_dtype = runtime_dtype loaded_vbpt_revision = runtime_revision else: loaded_tokenizer = AutoTokenizer.from_pretrained( spec["repo"], trust_remote_code=True, ) if loaded_tokenizer.pad_token_id is None: if loaded_tokenizer.eos_token is None: raise RuntimeError( f"Tokenizer for {spec['repo']} has neither pad_token nor eos_token." ) loaded_tokenizer.pad_token = loaded_tokenizer.eos_token loaded_tokenizer.truncation_side = "left" loaded = AutoModelForCausalLM.from_pretrained( spec["repo"], torch_dtype=torch.float32, low_cpu_mem_usage=True, trust_remote_code=True, ) loaded.eval() loaded.generation_config.use_cache = True if loaded.generation_config.pad_token_id is None: loaded.generation_config.pad_token_id = loaded_tokenizer.pad_token_id model = loaded tokenizer = loaded_tokenizer model_dataloader = None loaded_prompt_adapter = None loaded_vbpt_runtime_backend = None loaded_vbpt_device = None loaded_vbpt_dtype = None loaded_vbpt_revision = None elif backend == "onnx": if ORTModelForCausalLM is None: raise RuntimeError( "ONNX backend requires optimum[onnxruntime]. Install it or select " "the bachvnju-vbpt-1-0.5B Transformers model." ) loaded_tokenizer = AutoTokenizer.from_pretrained( spec["repo"], trust_remote_code=True, ) if loaded_tokenizer.pad_token_id is None: loaded_tokenizer.pad_token = loaded_tokenizer.eos_token loaded_tokenizer.truncation_side = "left" loaded = ORTModelForCausalLM.from_pretrained( spec["repo"], provider="CPUExecutionProvider", ) model = loaded tokenizer = loaded_tokenizer model_dataloader = None loaded_prompt_adapter = None loaded_vbpt_runtime_backend = None loaded_vbpt_device = None loaded_vbpt_dtype = None loaded_vbpt_revision = None else: raise RuntimeError(f"Unsupported backend: {backend}") loaded_model_key = model_key loaded_backend = backend loaded_context_tokens = context_tokens except HTTPException: unload_model() raise except Exception as exc: unload_model() raise HTTPException( status_code=500, detail=f"Could not load {model_key}: {type(exc).__name__}: {exc}", ) def make_messages(req: GenerateRequest) -> list[dict[str, str]]: system = req.system if req.response_schema is not None: system += ( "\nTrả về đúng một JSON document khớp JSON Schema được yêu cầu. " "Không thêm markdown, giải thích, hay văn bản ngoài JSON." ) elif req.json_mode: system += "\nTrả về đúng một JSON object hợp lệ, không thêm markdown hay văn bản ngoài JSON." result: list[dict[str, str]] = [{"role": "system", "content": system}] if req.messages: result.extend(item.model_dump() for item in req.messages) else: result.append({"role": "user", "content": req.prompt or ""}) return result def make_text_prompt(messages: list[dict[str, str]]) -> str: if getattr(tokenizer, "chat_template", None): return tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True, ) lines = [] for item in messages: if item["role"] == "system": lines.append(item["content"]) elif item["role"] == "user": lines.append(f"User: {item['content']}") else: lines.append(f"Assistant: {item['content']}") lines.append("Assistant:") return "\n".join(lines) def response_format_for(req: GenerateRequest) -> Optional[dict[str, Any]]: if req.response_schema is not None: return {"type": "json_object", "schema": req.response_schema} if req.json_mode: return {"type": "json_object"} return None def parse_and_validate_structured_response( answer: str, schema: Optional[dict[str, Any]], ) -> Any: try: parsed = json.loads(answer) except json.JSONDecodeError as exc: raise HTTPException( status_code=502, detail=( "GGUF backend returned non-JSON despite JSON mode. " f"Parser error: {exc.msg}" ), ) from exc if schema is not None: try: Draft202012Validator(schema).validate(parsed) except ValidationError as exc: path = "/".join(str(item) for item in exc.path) or "" raise HTTPException( status_code=502, detail=( "GGUF backend violated the requested JSON Schema after " f"grammar enforcement at {path}: {exc.message}" ), ) from exc return parsed def generate_gguf(req: GenerateRequest, messages: list[dict[str, str]]) -> tuple[str, str]: kwargs: dict[str, Any] = { "messages": messages, "temperature": req.temperature, "top_p": req.top_p, "top_k": req.top_k, "repeat_penalty": req.repetition_penalty, "max_tokens": req.max_new_tokens, "seed": req.seed, } response_format = response_format_for(req) if response_format is not None: kwargs["response_format"] = response_format result = model.create_chat_completion(**kwargs) answer = (result["choices"][0]["message"].get("content") or "").strip() return answer, getattr(model, "_selected_gguf_filename", "unknown") def generate_transformers_or_onnx( req: GenerateRequest, messages: list[dict[str, str]], ) -> tuple[str, int, int, Optional[str]]: if req.json_mode or req.response_schema is not None: raise HTTPException( status_code=422, detail=( "json_mode and response_schema are currently enforced only by the " "GGUF/llama.cpp backend; select a GGUF model." ), ) do_sample = req.temperature > 0.0 kwargs: dict[str, Any] = { "max_new_tokens": req.max_new_tokens, "do_sample": do_sample, "repetition_penalty": req.repetition_penalty, "use_cache": req.use_cache, "pad_token_id": tokenizer.pad_token_id, "eos_token_id": tokenizer.eos_token_id, } if req.max_time is not None: kwargs["max_time"] = req.max_time if do_sample: kwargs.update( temperature=req.temperature, top_p=req.top_p, top_k=req.top_k, ) if req.seed is not None: torch.manual_seed(req.seed) # VBPT does not pass through the generic `User:/Assistant:` formatter. # Its dedicated dataloader first tries the repository tokenizer template and # then deliberately falls back to the benchmark's raw-prompt behavior. if loaded_prompt_adapter == "vbpt": if model_dataloader is None: raise RuntimeError("VBPT model is loaded without its dedicated dataloader.") prepared = model_dataloader.prepare_batch( [messages], context_tokens=req.context_tokens, ) with torch.inference_mode(): output_ids = model.generate(**prepared.inputs, **kwargs) answer = model_dataloader.decode_new_tokens( output_ids, padded_prompt_tokens=prepared.prompt_token_count, )[0] new_tokens = int(output_ids.shape[1] - prepared.prompt_token_count) return answer, prepared.prompt_token_count, new_tokens, prepared.prompt_style prompt = make_text_prompt(messages) inputs = tokenizer( prompt, return_tensors="pt", truncation=True, max_length=req.context_tokens, ) with torch.inference_mode(): output_ids = model.generate(**inputs, **kwargs) input_tokens = int(inputs["input_ids"].shape[1]) new_tokens = output_ids[0][input_tokens:] answer = tokenizer.decode(new_tokens, skip_special_tokens=True).strip() return answer, input_tokens, int(new_tokens.shape[0]), None @asynccontextmanager async def lifespan(app: FastAPI): print("[vbpt-api] BUILD=space-ready-v5.7; VBPT uses a local embedded-runtime overlay, not the repository Python runtime.", flush=True) torch.set_num_threads(CPU_THREADS) try: torch.set_num_interop_threads(1) except RuntimeError: pass # Warm-load the default model. with engine_lock: load_model(DEFAULT_MODEL, context_tokens=2048, gguf_quant="auto") yield with engine_lock: unload_model() app = FastAPI( title="CPU Multi-Backend AI API (all-in-one VBPT adapter)", version="5.7.0-space-ready-vbpt-embedded-runtime", lifespan=lifespan, ) @app.get("/") def root(): return { "status": "ok", "endpoints": ["/health", "/system", "/models", "/generate", "/docs"], } @app.get("/health") def health(): details: dict[str, Any] = { "api_build": "space-ready-v5.7", "status": "healthy" if model is not None else "loading", "loaded_model": loaded_model_key, "backend": loaded_backend, "cpu_threads": CPU_THREADS, "context_tokens": loaded_context_tokens, "max_context_tokens": MAX_CONTEXT_TOKENS, "max_new_tokens": MAX_NEW_TOKENS, } if loaded_backend == "gguf" and model is not None: details.update( gguf_file=getattr(model, "_selected_gguf_filename", None), gguf_quant_request=getattr(model, "_selected_quant_request", None), chat_template_source=getattr(model, "_chat_template_source", None), ) if loaded_prompt_adapter is not None: details["prompt_adapter"] = loaded_prompt_adapter if loaded_prompt_adapter == "vbpt": details.update( vbpt_runtime=VBPT_RUNTIME_VERSION, vbpt_runtime_backend=loaded_vbpt_runtime_backend, vbpt_device=loaded_vbpt_device, vbpt_dtype=loaded_vbpt_dtype, vbpt_revision=loaded_vbpt_revision, ) return details @app.get("/system") def system(): try: lscpu = subprocess.check_output(["lscpu"], text=True, timeout=3) except Exception as exc: lscpu = f"lscpu unavailable: {exc}" return { "processor": platform.processor(), "logical_cores": os.cpu_count(), "configured_cpu_threads": CPU_THREADS, "lscpu": lscpu, } @app.get("/models") def list_models(): return { "loaded_model": loaded_model_key, "models": [ { "key": key, "backend": spec["backend"], "repo": spec["repo"], "label": spec["label"], "loader": spec.get("loader"), "note": spec.get("note"), } for key, spec in MODEL_CATALOG.items() ], "notes": [ "Only one model is loaded at once.", "GGUF auto chooses Q6 for <=1B, Q4 for 1B–4B, and Q3 for >4B when available.", "No catalog entry forces chat_format; GGUF tokenizer.chat_template is preferred.", "JSON Schema is grammar-constrained by llama.cpp and validated again server-side.", "Generic Transformers 4/8-bit bitsandbytes is not enabled on CPU.", "ONNX backend needs a repository containing a pre-exported ONNX model.", "VBPT uses a dedicated tokenizer/prompt/batch dataloader rather than the generic Transformer formatter.", "VBPT uses the embedded verified GDN runtime: auto selects CUDA+FLA when available, otherwise CPU reference mode.", "Set VBPT_RUNTIME_BACKEND=fla|reference, VBPT_DEVICE=cuda|cpu, and optional VBPT_REVISION to control VBPT startup.", ], } @app.post("/generate") def generate(req: GenerateRequest): with engine_lock: load_model( model_key=req.model, context_tokens=req.context_tokens, gguf_quant=req.gguf_quant, ) messages = make_messages(req) try: if loaded_backend == "gguf": answer, gguf_file = generate_gguf(req, messages) payload: dict[str, Any] = { "response": answer, "model": loaded_model_key, "backend": loaded_backend, "gguf_file": gguf_file, "context_tokens": req.context_tokens, } if req.json_mode or req.response_schema is not None: payload["response_json"] = parse_and_validate_structured_response( answer, req.response_schema, ) return payload answer, input_tokens, output_tokens, prompt_style = generate_transformers_or_onnx( req, messages, ) payload = { "response": answer, "model": loaded_model_key, "backend": loaded_backend, "input_tokens": input_tokens, "output_tokens": output_tokens, } if prompt_style is not None: payload["prompt_style"] = prompt_style return payload except HTTPException: raise except Exception as exc: raise HTTPException( status_code=500, detail=f"Generation failed: {type(exc).__name__}: {exc}", ) def _run_vbpt_loader_self_test() -> None: """Offline test for the embedded VBPT adapter; never downloads model weights.""" class _Config: pad_token_id = None eos_token_id = None class _GenerationConfig: pad_token_id = None eos_token_id = None use_cache = False class _FakeModel: def __init__(self) -> None: self.config = _Config() self.generation_config = _GenerationConfig() class _FakeTokenizer: def __init__(self, *, has_template: bool) -> None: self.pad_token_id = None self.eos_token_id = 2 self.eos_token = "" self.pad_token = None self.padding_side = "right" self.truncation_side = "right" self.chat_template = "placeholder" if has_template else None def __setattr__(self, name: str, value: Any) -> None: object.__setattr__(self, name, value) if name == "pad_token" and value == "": object.__setattr__(self, "pad_token_id", 2) def apply_chat_template( self, messages: Sequence[Mapping[str, str]], *, tokenize: bool, add_generation_prompt: bool, ) -> str: assert tokenize is False assert add_generation_prompt is True return "|".join(f"{item['role']}={item['content']}" for item in messages) + "|assistant=" def __call__( self, prompts: str | Sequence[str], *, return_tensors: str, padding: bool, truncation: bool, max_length: int, return_attention_mask: bool, ) -> dict[str, torch.Tensor]: if isinstance(prompts, str): prompts = [prompts] rows = [[(ord(ch) % 80) + 3 for ch in prompt][-max_length:] for prompt in prompts] width = max(len(row) for row in rows) padded = [[self.pad_token_id] * (width - len(row)) + row for row in rows] masks = [[0] * (width - len(row)) + [1] * len(row) for row in rows] return { "input_ids": torch.tensor(padded), "attention_mask": torch.tensor(masks), } def decode(self, ids: Any, *, skip_special_tokens: bool = True) -> str: values = ids.tolist() if isinstance(ids, torch.Tensor) else list(ids) return "".join("Z" if value == 99 else f"<{value}>" for value in values if value != 2) raw_tokenizer = _FakeTokenizer(has_template=False) loader = VBPTDataLoader(raw_tokenizer) fake_model = _FakeModel() loader.apply_generation_token_ids(fake_model) batch = loader.prepare_batch( [ [{"role": "system", "content": "S"}, {"role": "user", "content": "Xin chào"}], [{"role": "user", "content": "B"}], ], context_tokens=32, ) assert batch.prompt_style == "raw_benchmark_compatible" assert raw_tokenizer.padding_side == "left" assert raw_tokenizer.truncation_side == "left" assert fake_model.generation_config.pad_token_id == 2 output = torch.cat([batch.inputs["input_ids"], torch.tensor([[99], [99]])], dim=1) assert loader.decode_new_tokens(output, padded_prompt_tokens=batch.prompt_token_count) == ["Z", "Z"] templated_loader = VBPTDataLoader(_FakeTokenizer(has_template=True)) _, style = templated_loader.render_messages([{"role": "user", "content": "hello"}]) assert style == "native_chat_template" print("Embedded VBPT dataloader self-test: PASS") if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="All-in-one multi-backend API with conditional VBPT loader") parser.add_argument("--self-test-vbpt-loader", action="store_true", help="run the offline embedded VBPT loader self-test") parser.add_argument("--host", default=os.getenv("HOST", "0.0.0.0")) parser.add_argument("--port", type=int, default=int(os.getenv("PORT", "8000"))) args = parser.parse_args() if args.self_test_vbpt_loader: _run_vbpt_loader_self_test() else: import uvicorn uvicorn.run(app, host=args.host, port=args.port)