60 lines
4.6 KiB
Python
60 lines
4.6 KiB
Python
"""Pinned Qwen3 split/token/request IDs and local DS4 family controls; no inference."""
|
||
import argparse,hashlib,json,os,struct,subprocess,sys
|
||
from pathlib import Path
|
||
parser=argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument('checkout',type=Path)
|
||
parser.add_argument('root',type=Path)
|
||
parser.add_argument('out',type=Path)
|
||
parser.add_argument('--ds4',type=Path)
|
||
parser.add_argument('--glm',type=Path)
|
||
parser.add_argument('--deepseek',type=Path)
|
||
args=parser.parse_args()
|
||
checkout,root,out=args.checkout,args.root,args.out
|
||
assert subprocess.check_output(['git','-C',str(checkout),'rev-parse','HEAD'],text=True).strip()=='9882703f3105363ddc37eca9f97aa09a1d387112'
|
||
sys.path.insert(0,str(checkout))
|
||
os.environ.update(HF_HUB_OFFLINE='1',TRANSFORMERS_OFFLINE='1')
|
||
from tokenizers import Regex,Tokenizer as BackendTokenizer,pre_tokenizers
|
||
from mtplx.runtime import _QWEN3_PRETOKENIZER_SPLIT,_load_tokenizer_resilient
|
||
from mtplx.server import openai as server
|
||
out.mkdir(parents=True,exist_ok=True)
|
||
tokenizer=_load_tokenizer_resilient(root,json.loads((root/'config.json').read_text()))
|
||
component=BackendTokenizer.from_str(tokenizer.backend_tokenizer.to_str())
|
||
component.normalizer=None
|
||
split=pre_tokenizers.Split(Regex(_QWEN3_PRETOKENIZER_SPLIT),behavior='isolated')
|
||
scalars=[chr(i) for i in range(0x110000) if not 0xd800<=i<=0xdfff]
|
||
text=''.join('A'+c+'A ' for c in scalars)
|
||
h=hashlib.sha256();count=0
|
||
for piece,_ in split.pre_tokenize_str(text):
|
||
b=piece.encode(); h.update(struct.pack('<I',len(b)));h.update(b);count+=1
|
||
(out/'scalar-split.json').write_text(json.dumps(dict(scalars=len(scalars),pieces=count,sha256=h.hexdigest(),pattern=_QWEN3_PRETOKENIZER_SPLIT)))
|
||
number_split=pre_tokenizers.Split(Regex(r'\p{N}'),behavior='isolated')
|
||
numbers=[p for p,_ in number_split.pre_tokenize_str('A'.join(scalars)) if len(p)==1 and p!='A']
|
||
texts=[' ①',' Ⅷ',' ²',' 𝟘','Item ①, Ⅷ, ², 𝟘 and 123.','भारत एक विशाल देश है जिसकी संस्कृति बहुत पुरानी है।','ประเทศไทยมีวัฒนธรรมที่เก่าแก่และสวยงามมาก','الْعَرَبِيَّةُ','The quick brown fox jumps over the lazy dog.','中文，标点。','def f(x):\n return x + 123\n',"I'm WE'RE it's he'd I'll you've can't 'ſx",'\u0301a a\u0301 \u0301 १\u0301b','a\r\n b\t\t','a ','a\t \u2003b','a\u0085\u00a0b','x \n \r\n y','😀 ①🙂Ⅷ ²+𝟘']
|
||
texts += [prefix+c+suffix for c in numbers for prefix,suffix in [(' ',''),('a','b'),(' \u0301','\u0301x')]]
|
||
records=[]
|
||
for text in texts:
|
||
full_ids=tokenizer.encode(text,add_special_tokens=False)
|
||
ids=component.encode(text,add_special_tokens=False).ids
|
||
assert component.decode(ids,skip_special_tokens=False)==text,(text,ids)
|
||
records.append(dict(text=text,ids=ids,full_ids=full_ids,pieces=[p for p,_ in split.pre_tokenize_str(text)]))
|
||
(out/'qwen.json').write_text(json.dumps(dict(number_count=len(numbers),cases=records),ensure_ascii=False))
|
||
requests=[]
|
||
for effort in ['direct','low','medium','xhigh']:
|
||
messages=[dict(role='user',content=' ① Ⅷ ² 𝟘 भारत \u0301a\n '),dict(role='assistant',content='One ①.',reasoning_content='' if effort=='direct' else 'Brief ² thought.'),dict(role='user',content='Continue Ⅷ and 𝟘.\n x')]
|
||
ids=server._encode_messages_uncached(tokenizer,[server.ChatMessage(**m) for m in messages],enable_thinking=effort!='direct',reasoning_effort=None if effort=='direct' else effort,preserve_reasoning_history=True,tools=None)
|
||
requests.append(dict(effort=effort,messages=messages,ids=ids,rendered=tokenizer.decode(ids,skip_special_tokens=False,clean_up_tokenization_spaces=False)))
|
||
(out/'requests.json').write_text(json.dumps(requests,ensure_ascii=False))
|
||
print(json.dumps(dict(case='qwen_oracle',number_count=len(numbers),cases=len(records),requests=len(requests),scalar_count=len(scalars),split_count=count,split_sha256=h.hexdigest())),flush=True)
|
||
|
||
# CPU vocabulary loading only; DS4 remains an external behavioral oracle.
|
||
if args.ds4:
|
||
assert args.glm and args.deepseek
|
||
families=[]
|
||
for family,path in [('glm',args.glm),('deepseek',args.deepseek)]:
|
||
cases=[]
|
||
for text in texts[:19]:
|
||
raw=subprocess.check_output([str(args.ds4),'-m',str(path),'--dump-tokens','--raw','-p',text],text=True)
|
||
cases.append(dict(text=text,ids=json.loads(raw.splitlines()[0])))
|
||
families.append(dict(family=family,model=str(path),cases=cases))
|
||
(out/'ds4.json').write_text(json.dumps(dict(revision=subprocess.check_output(['git','-C',str(args.ds4.parent),'rev-parse','HEAD'],text=True).strip(),binary_sha256=hashlib.sha256(args.ds4.read_bytes()).hexdigest(),families=families),ensure_ascii=False))
|