PhysiQuanty/Binary-LLM-POC

1#!/usr/bin/env python32# infer.py3# ============================================================4# HF inference (CausalLM) en base-25# - Encode le --prompt en bits (MSB->LSB) comme llmTalk6# - Prompt = [BOS] + bits + [EOS] + [BOS]  (reset latent)7# - Boucle manuelle token-par-token (pas model.generate)8# - Décodage FINAL via decode_base2_digits_strict9# - indentation AVEC TABULATIONS (comme ton fichier actuel)10# ============================================================11 12import sys13import os14import argparse15import random16import codecs17from typing import List, Dict18from collections import Counter19 20import torch21from transformers import AutoModelForCausalLM22 23 24def decode_base2_digits_strict(digits: List[int], *, encoding: str = "utf-8", errors: str = "replace") -> str:25	# Filtre minimal: ne garder que 0/1 (au cas où)26	bits: List[int] = []27	for d in digits:28		di = int(d)29		if di == 0 or di == 1:30			bits.append(di)31 32	n_full_bytes = len(bits) // 833	if n_full_bytes <= 0:34		return ""35 36	out = bytearray(n_full_bytes)37 38	j = 039	for i in range(n_full_bytes):40		# MSB -> LSB (bits[j] est le bit de poids fort)41		b = 042		b = (b << 1) | bits[j + 0]43		b = (b << 1) | bits[j + 1]44		b = (b << 1) | bits[j + 2]45		b = (b << 1) | bits[j + 3]46		b = (b << 1) | bits[j + 4]47		b = (b << 1) | bits[j + 5]48		b = (b << 1) | bits[j + 6]49		b = (b << 1) | bits[j + 7]50		out[i] = b51		j += 852 53	bb = bytes(out)54 55	# Décodage robuste UTF-8 (gère proprement les séquences multi-octets)56	if encoding.lower() == "utf-8":57		inc = codecs.getincrementaldecoder("utf-8")(errors=errors)58		s = inc.decode(bb, final=False)59		s += inc.decode(b"", final=True)60		return s61 62	return bb.decode(encoding, errors=errors)63 64 65def bytes_to_base2_digits_bytesafe(data: bytes) -> List[int]:66	digits: List[int] = []67	for b in data:68		for i in range(7, -1, -1):69			digits.append((b >> i) & 1)70	return digits71 72 73def text_to_base2_digits(text: str) -> List[int]:74	# Même logique que llmTalk: UTF-8 -> bits MSB->LSB75	return bytes_to_base2_digits_bytesafe(text.encode("utf-8"))76 77 78def wrap_base2_sequence_2(ids: List[int], bos_id: int, eos_id: int) -> List[int]:79	return [int(bos_id), *ids, int(eos_id)]80 81 82def apply_repetition_penalty_(logits: torch.Tensor, token_ids: List[int], penalty: float) -> None:83	if penalty is None or penalty == 1.0 or penalty <= 0:84		return85	for t in set(token_ids):86		val = logits[0, t]87		logits[0, t] = val * penalty if val < 0 else val / penalty88 89 90def apply_presence_frequency_penalties_(logits: torch.Tensor, token_ids: List[int], presence_penalty: float, frequency_penalty: float) -> None:91	counts = Counter(token_ids)92	if presence_penalty:93		for t in counts:94			logits[0, t] -= presence_penalty95	if frequency_penalty:96		for t, c in counts.items():97			logits[0, t] -= frequency_penalty * c98 99 100def get_banned_tokens_no_repeat_ngram(seq: List[int], n: int) -> set:101	if n <= 0 or len(seq) < n - 1:102		return set()103 104	prefix_len = n - 1105	ngrams: Dict[tuple, set] = {}106	for i in range(len(seq) - n + 1):107		prefix = tuple(seq[i:i + prefix_len])108		nxt = seq[i + prefix_len]109		ngrams.setdefault(prefix, set()).add(nxt)110 111	return ngrams.get(tuple(seq[-prefix_len:]), set())112 113 114def mask_banned_tokens_(logits: torch.Tensor, banned: set) -> None:115	if banned:116		logits[0, list(banned)] = float("-inf")117 118 119def _maybe_hf_token() -> str:120	tok = os.environ.get("HF_TOKEN")121	if tok:122		return tok123	tok = os.environ.get("HUGGINGFACE_HUB_TOKEN")124	if tok:125		return tok126	return ""127 128 129def main() -> None:130	parser = argparse.ArgumentParser()131 132	parser.add_argument("--repo", type=str, required=True, help="chemin dossier HF local (./hf_binaryllm_repo) ou repo_id")133	parser.add_argument("--device", type=str, default="cuda", choices=["cpu", "cuda"])134	parser.add_argument("--seed", type=int, default=-1)135 136	# Base-2 avec 2 spéciaux => vocab_size=4 attendu: 0,1 + BOS=2 + EOS=3137	parser.add_argument("--bos", type=int, default=2, help="BOS id (base2: BOS=2)")138	parser.add_argument("--eos", type=int, default=3, help="EOS id (base2: EOS=3)")139	parser.add_argument("--prompt", type=str, required=True, help="texte à encoder en base2 (UTF-8 -> bits MSB->LSB)")140 141	parser.add_argument("--max_new_tokens", type=int, default=800)142	parser.add_argument("--temperature", type=float, default=0.7)143	parser.add_argument("--top_k", type=int, default=50)144 145	parser.add_argument("--repetition_penalty", type=float, default=1.0)146	parser.add_argument("--presence_penalty", type=float, default=0.0)147	parser.add_argument("--frequency_penalty", type=float, default=0.0)148	parser.add_argument("--no_repeat_ngram_size", type=int, default=0)149 150	parser.add_argument("--decode_encoding", type=str, default="utf-8")151	parser.add_argument("--decode_errors", type=str, default="replace")152	parser.add_argument("--print_ids", action="store_true")153	parser.add_argument("--stream", action="store_true", help="stream strict (réaffiche decode strict à chaque step)")154 155	args = parser.parse_args()156 157	seed = args.seed if args.seed >= 0 else random.randint(0, 2**31 - 1)158	print(f"[Seed] {seed}")159	torch.manual_seed(seed)160	if torch.cuda.is_available():161		torch.cuda.manual_seed_all(seed)162 163	device = torch.device("cuda" if (args.device == "cuda" and torch.cuda.is_available()) else "cpu")164	print(f"[Device] {device}")165 166	# --------- Load HF model ---------167	hf_token = _maybe_hf_token()168	if hf_token:169		m = AutoModelForCausalLM.from_pretrained(args.repo, trust_remote_code=True, token=hf_token)170	else:171		m = AutoModelForCausalLM.from_pretrained(args.repo, trust_remote_code=True)172 173	m.to(device)174	m.eval()175 176	# IMPORTANT: pas de KV-cache (train-like)177	if hasattr(m, "config") and m.config is not None:178		m.config.use_cache = True179 180	# --------- Encode prompt EXACTEMENT comme llmTalk (base=2) ---------181	def encode_prompt(text: str) -> List[int]:182		ids = text_to_base2_digits(text)                      # 0/1 bits (MSB->LSB)183		ids = wrap_base2_sequence_2(ids, args.bos, args.eos) # [BOS] bits [EOS]184		ids = ids + [int(args.bos)]                          # reset latent: ...[EOS][BOS]185		print("[+] PROMPT IDS = ", ids)                             186		return ids187 188	prompt_ids = encode_prompt(args.prompt)189 190	tokens = torch.tensor([prompt_ids], dtype=torch.long, device=device)191	generated: List[int] = []192	last_text_len = 0193 194	print("\n[Prompt]\n", args.prompt)195	print(f"\n[Prompt IDs] len={len(prompt_ids)} | BOS={args.bos} EOS={args.eos}")196	print("\n[Stream]" if args.stream else "\n[Output]")197 198	with torch.no_grad():199		for _ in range(int(args.max_new_tokens)):200			# full forward sur toute la séquence, sans cache201			out = m(input_ids=tokens, use_cache=True)202			logits = out.logits[:, -1, :]203 204			full_seq = tokens[0].tolist()205 206			apply_repetition_penalty_(logits, full_seq, float(args.repetition_penalty))207			apply_presence_frequency_penalties_(logits, full_seq, float(args.presence_penalty), float(args.frequency_penalty))208 209			if int(args.no_repeat_ngram_size) > 0:210				banned = get_banned_tokens_no_repeat_ngram(full_seq, int(args.no_repeat_ngram_size))211				mask_banned_tokens_(logits, banned)212 213			logits = logits / max(float(args.temperature), 1e-6)214 215			if 0 < int(args.top_k) < logits.size(-1):216				v, _ = torch.topk(logits, int(args.top_k))217				logits[logits < v[:, [-1]]] = float("-inf")218 219			probs = torch.softmax(logits, dim=-1)220			next_token = torch.multinomial(probs, 1)221			tok_id = int(next_token.item())222 223			if tok_id == int(args.eos):224				break225 226			tokens = torch.cat([tokens, next_token], dim=1)227			generated.append(tok_id)228 229			if args.stream:230				text = decode_base2_digits_strict(generated, encoding=args.decode_encoding, errors=args.decode_errors)231				if len(text) > last_text_len:232					sys.stdout.write(text[last_text_len:])233					sys.stdout.flush()234					last_text_len = len(text)235 236	if args.stream:237		print()238 239	print("\n[Final Output]\n")240	print(decode_base2_digits_strict(generated, encoding=args.decode_encoding, errors=args.decode_errors))241 242	if args.print_ids:243		print("\n[Generated IDs]\n")244		print(generated)245 246 247if __name__ == "__main__":248	main()249