"""Minimal loader for the Fairleap v1 CLM Qwen3.5-4B adapter. Three things about this model will look like broken weights if you meet them cold, so they are handled here rather than left as surprises: 1. `Qwen/Qwen3.5-4B` is a **vision-language** checkpoint (`Qwen3_5ForConditionalGeneration`). `from_pretrained` returns a `Qwen3VLProcessor`, not a tokenizer, and its `__call__` reads the first positional argument as an *image source* -- so `processor("halo")` raises `Incorrect image source`. The inner text tokenizer is what you want. 2. This repository holds a **LoRA adapter only**. There are no base weights and no `config.json`, so `AutoModelForCausalLM.from_pretrained(".")` fails and PEFT falls back to treating the local path as a Hub repo id. 3. The model is useless without the **stuffed system prompt**. The service it was trained for is stateless and identity-blind: every fact about a driver arrives in the request. `build_system_prompt()` assembles the shape the model was trained on. python load_model.py # loads and runs one grounded question """ import json from pathlib import Path REPO_ID = "fairleap-ai/fairleap-v1-clm-qwen3.5-4b-adapter" BASE_MODEL = "Qwen/Qwen3.5-4B" ADAPTER_DIR = str(Path(__file__).parent) MAX_SEQ = 4096 # The only callable tool. Advice topics are prompt blocks, not tools -- calling # an LLM endpoint from an LLM is an extra hop and a second inference bill. PREDICT_EARNINGS_TOOL = { "type": "function", "function": { "name": "predict_earnings", "description": ( "Prediksi penghasilan harian dan jam kerja untuk rentang tanggal di masa " "depan menggunakan model XGBoost. Riwayat harian driver ditambahkan " "otomatis oleh sistem, jangan sertakan riwayat dalam argumen." ), "parameters": { "type": "object", "properties": { "start": {"type": "string", "description": "Tanggal awal, YYYY-MM-DD."}, "end": {"type": "string", "description": "Tanggal akhir, YYYY-MM-DD."}, "wellness_score": {"type": "integer", "minimum": 1, "maximum": 100}, }, "required": ["start", "end", "wellness_score"], }, }, } PERSONA = """Kamu adalah Fairleap AI Assistant, asisten untuk mitra driver Gojek/GOTO di Indonesia. Jawab hangat, ringkas, dan konkret dalam Bahasa Indonesia. Jangan pernah mengarang angka.""" def build_system_prompt(today, city, vehicle, risk, wellness_score, period, totals, daily=None): """Assemble the stuffed driver context the model was trained to read. daily is an optional list of (date, earnings, orders) rows. Training used seven-day summaries plus up to fourteen daily rows; the model reads these figures literally, so anything absent here it should decline to state. """ lines = [ PERSONA, "", "=== KONTEKS DRIVER (disediakan sistem, bukan hasil tool) ===", f"Tanggal hari ini: {today}", f"Kota: {city}", f"Kendaraan: {vehicle}", f"Toleransi risiko: {risk}", f"Skor kesehatan terakhir: {wellness_score}/100", f"Ringkasan 7 hari terakhir ({period}):", ] lines += [f"- {k}: {v}" for k, v in totals.items()] if daily: lines.append("Rincian harian:") # Indonesian thousands separator is a dot, so convert the number alone -- # doing it on the whole line turns the comma before "order" into a dot. lines += [f"- {d}: Rp{e:,}".replace(",", ".") + f", {o} order" for d, e, o in daily] return "\n".join(lines) def get_tokenizer(processor): """Extract the text tokenizer from a VL processor. See note 1 above.""" return getattr(processor, "tokenizer", processor) def load(adapter_dir=ADAPTER_DIR, load_in_4bit=True): """Return (model, tokenizer). Prefers unsloth, falls back to peft.""" # Without this the loaders report `Unrecognized model in . Should have # a model_type key in its config.json`, which sends you looking for a # corrupt checkpoint instead of a wrong path. ADAPTER_DIR defaults to this # file's own directory, so moving load_model.py away from the weights # breaks it. if not Path(adapter_dir, "adapter_config.json").is_file(): raise SystemExit( f"no adapter_config.json in {adapter_dir!r} -- pass adapter_dir= " f"pointing at the directory holding adapter_model.safetensors" ) try: from unsloth import FastLanguageModel model, processor = FastLanguageModel.from_pretrained( model_name=adapter_dir, max_seq_length=MAX_SEQ, dtype=None, load_in_4bit=load_in_4bit) FastLanguageModel.for_inference(model) return model, get_tokenizer(processor) except ImportError: pass import torch from peft import PeftModel from transformers import AutoModelForImageTextToText, AutoProcessor processor = AutoProcessor.from_pretrained(BASE_MODEL) base = AutoModelForImageTextToText.from_pretrained( BASE_MODEL, torch_dtype="auto", device_map="auto") model = PeftModel.from_pretrained(base, adapter_dir) model.eval() return model, get_tokenizer(processor) def chat(model, tokenizer, messages, tools=None, max_new_tokens=400, temperature=0.7, top_p=0.9): """One turn. Pass tools only when a forecast is plausibly needed. Offering the tool on every turn is not the distribution this model was trained on -- under 10% of training conversations carried it -- and it provokes forecast calls on questions about fatigue or traffic. See the model card, Limitations. """ try: text = tokenizer.apply_chat_template( messages, tools=tools, tokenize=False, add_generation_prompt=True, enable_thinking=False) except TypeError: text = tokenizer.apply_chat_template( messages, tools=tools, tokenize=False, add_generation_prompt=True) encoded = tokenizer(text, return_tensors="pt") encoded = {k: v.to(model.device) for k, v in encoded.items() if k in ("input_ids", "attention_mask")} import torch with torch.no_grad(): out = model.generate( **encoded, max_new_tokens=max_new_tokens, temperature=temperature, top_p=top_p, do_sample=True, # Without both of these generation runs past the tool call and # invents the result it was about to ask for. eos_token_id=tokenizer.eos_token_id, pad_token_id=tokenizer.pad_token_id or tokenizer.eos_token_id, ) return tokenizer.decode(out[0][encoded["input_ids"].shape[-1]:], skip_special_tokens=True) def parse_tool_call(text): """Recover a tool call. The model emits Qwen's XML block, not JSON.""" import re fn = re.search(r"", text, re.I) if not fn: return None args = {} for key, raw in re.findall(r"\s*(.*?)\s*", text, re.I | re.S): args[key] = int(raw) if raw.lstrip("-").isdigit() else raw return fn.group(1), args if __name__ == "__main__": system = build_system_prompt( today="2026-08-24", city="Bekasi", vehicle="motor", risk="sedang", wellness_score=62, period="2026-08-18 s/d 2026-08-24", totals={ "Total penghasilan": "Rp1.482.000", "Total order": "88", "Hari kerja": "6 dari 7 hari", "Rata-rata per hari kerja": "Rp247.000", "Hari terbaik": "2026-08-21 (Rp281.000, 16 order)", }, ) model, tokenizer = load() question = "berapa penghasilan saya minggu ini?" reply = chat(model, tokenizer, [{"role": "system", "content": system}, {"role": "user", "content": question}]) print(f">>> {question}\n{reply.strip()}\n") question = "kira-kira minggu depan saya bisa dapat berapa?" reply = chat(model, tokenizer, [{"role": "system", "content": system}, {"role": "user", "content": question}], tools=[PREDICT_EARNINGS_TOOL]) print(f">>> {question}\n{reply.strip()}") print(f"\nparsed tool call: {json.dumps(parse_tool_call(reply), ensure_ascii=False)}")