| """Classify one subject/body JSON record using Letterweave's seven-purpose contract.""" |
| import argparse |
| import json |
| import math |
| from pathlib import Path |
|
|
| def main(): |
| parser=argparse.ArgumentParser(description=__doc__) |
| parser.add_argument('--checkpoint', required=True) |
| parser.add_argument('--revision', help='Pin a published model commit for reproducibility') |
| parser.add_argument('--email', type=Path, required=True, help='UTF-8 JSON file with subject and body strings') |
| parser.add_argument("--max-input-tokens", type=int, default=16384, help="Explicit serving budget up to 16384 tokens; overflow is rejected") |
| args=parser.parse_args() |
| if not 0 < args.max_input_tokens <= 16384: |
| raise ValueError("Serving budget must be between 1 and 16384 tokens") |
| state=json.loads(args.email.read_text()) |
| if set(state)!={'subject','body'} or not all(isinstance(v,str) for v in state.values()): |
| raise ValueError('Expected exactly subject and body strings') |
| import unsloth |
| import torch |
| from huggingface_hub import snapshot_download |
| from unsloth import FastDecisionModel |
| from unsloth.models.clef import encode_record |
| if not torch.cuda.is_available(): |
| raise RuntimeError('This release was qualified on CUDA, not CPU inference') |
| folder=Path(args.checkpoint) |
| if not folder.is_dir(): |
| folder=Path(snapshot_download(args.checkpoint,revision=args.revision)) |
| model,tokenizer=FastDecisionModel.from_pretrained(str(folder),max_seq_length=args.max_input_tokens,load_in_4bit=True,dtype=torch.bfloat16,offload_embedding=True) |
| embeddings={id(m.weight) for m in model.encoder.modules() if isinstance(m,torch.nn.Embedding)} |
| if not all(v.device.type=='cuda' or (v.device.type=='cpu' and id(v) in embeddings) for v in model.parameters()): |
| raise RuntimeError('Only native embedding CPU offload is qualified; decoder and decision head require CUDA') |
| questions=json.loads((folder/'questions.json').read_text()) |
| mapping=json.loads((folder/'label-map.json').read_text()) |
| text_tokenizer=getattr(tokenizer,'tokenizer',tokenizer) |
| record=encode_record(text_tokenizer,{'state':state,'questions':questions},max_length=1000000) |
| if len(record.input_ids)>args.max_input_tokens or len(record.questions)!=7: |
| raise ValueError('Complete input exceeds the configured token budget or seven-question contract') |
| FastDecisionModel.for_inference(model) |
| answers=FastDecisionModel.predict(model,tokenizer,state,questions) |
| if set(answers)!=set(mapping.values()): |
| raise ValueError('Unexpected answer IDs') |
| probabilities={label:float(answers[qid]['noul']) for label,qid in mapping.items()} |
| if not all(math.isfinite(p) and 0<=p<=1 for p in probabilities.values()): |
| raise ValueError('Invalid purpose probability') |
| print(json.dumps({'labels':[label for label,p in probabilities.items() if p>.5],'probabilities':probabilities,'threshold':0.5,'threshold_policy':'strictly_greater_than','input_tokens':len(record.input_ids)},ensure_ascii=False)) |
|
|
| if __name__=='__main__': |
| main() |
|
|
