fixed_nemotron_notebook.ipynb

{
 "nbformat": 4,
 "nbformat_minor": 5,
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "name": "python",
   "version": "3.12.0"
  },
  "kaggle": {
   "accelerator": "nvidiaTeslaT4x2",
   "isInternetEnabled": true,
   "isGpuEnabled": true,
   "language": "python",
   "sourceType": "notebook",
   "gpuMemoryLimit": "30GB"
  }
 },
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": "# Latent Dynamical Control of Low-Rank Adaptation in Frozen Transformer Models\n**Locked In Safety STL Inc / Lightfoot Technology LLC**\n\nCopyright 2026. Apache 2.0 License. Credit required on any use or distribution.\n\n---\n\n## Abstract\n\nWe introduce a temporally-driven adapter modulation mechanism where low-rank updates are\ncontrolled by a latent dynamical system, enabling non-stationary adaptation of frozen\ntransformer representations.\n\nThe base model remains fully frozen. A latent operator field modulates LoRA adapter\nscaling through a bounded, history-dependent control field.",
   "id": "m_title"
  },
  {
   "cell_type": "code",
   "metadata": {},
   "source": "import sys, subprocess\nsubprocess.run([sys.executable,'-m','pip','install','-q',\n    'z3-solver','sympy','peft','transformers','accelerate','bitsandbytes',\n    'datasets','trl','scipy'], check=False)\nprint(\"Installs done\")",
   "id": "c_install",
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "code",
   "metadata": {},
   "source": "import os, re, math, collections, zipfile, time\nimport numpy as np\nimport torch\nfrom datasets import load_dataset\nfrom transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig, TrainingArguments\nfrom peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training, TaskType\nfrom trl import SFTTrainer\nfrom kaggle_secrets import UserSecretsClient\nfrom scipy.linalg import eigvalsh\nprint(\"Imports done\")",
   "id": "c_imports",
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "code",
   "metadata": {},
   "source": "_secrets = UserSecretsClient()\nhf_token = None\nfor _name in [\"k2\", \"K2\", \"hf_token\", \"HF_TOKEN\"]:\n    try:\n        hf_token = _secrets.get_secret(_name)\n        print(f\"Token loaded: '{_name}'\")\n        break\n    except Exception:\n        continue\nif hf_token is None:\n    raise ValueError(\"No HuggingFace token found. Add a secret named 'k2'.\")",
   "id": "c_token",
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "code",
   "metadata": {},
   "source": "bnb_config = BitsAndBytesConfig(\n    load_in_4bit=True,\n    bnb_4bit_quant_type=\"nf4\",\n    bnb_4bit_compute_dtype=torch.bfloat16,\n    bnb_4bit_use_double_quant=True\n)\n\nmodel_name = \"nvidia/NVIDIA-Nemotron-3-Nano-30B-A3B-BF16\"\ntokenizer = AutoTokenizer.from_pretrained(model_name, token=hf_token, trust_remote_code=True)\nif tokenizer.pad_token is None:\n    tokenizer.pad_token = tokenizer.eos_token\n\nmodel = AutoModelForCausalLM.from_pretrained(\n    model_name, token=hf_token,\n    quantization_config=bnb_config,\n    device_map=\"auto\", trust_remote_code=True,\n    torch_dtype=torch.bfloat16,\n    attn_implementation=\"eager\"\n)\nprint(\"Model loaded:\", model_name)",
   "id": "c_model",
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "code",
   "metadata": {},
   "source": "model = prepare_model_for_kbit_training(model)\n\nmodel = get_peft_model(model, LoraConfig(\n    r=32, lora_alpha=64,\n    target_modules=[\"q_proj\",\"k_proj\",\"v_proj\",\"o_proj\",\"gate_proj\",\"up_proj\",\"down_proj\"],\n    lora_dropout=0.05, bias=\"none\",\n    task_type=TaskType.CAUSAL_LM\n))\nmodel.print_trainable_parameters()\nprint(\"LoRA adapter attached (improved config)\")",
   "id": "c_lora",
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "code",
   "metadata": {},
   "source": "# Proprietary verification layer — internal structure not disclosed.\n\ndef _build_verification_manifold():\n    n, dim = 8, 256\n    H = np.zeros((dim, dim), dtype=complex)\n    I2 = np.eye(2)\n    X = np.array([[0,1],[1,0]], dtype=complex)\n    Y = np.array([[0,-1j],[1j,0]], dtype=complex)\n    Z = np.array([[1,0],[0,-1]], dtype=complex)\n    def _k1(op, s):\n        ops=[I2]*n; ops[s]=op; out=ops[0]\n        for o in ops[1:]: out=np.kron(out,o)\n        return out\n    def _k2(o1,s1,o2,s2):\n        ops=[I2]*n; ops[s1]=o1; ops[s2]=o2; out=ops[0]\n        for o in ops[1:]: out=np.kron(out,o)\n        return out\n    _a,_b,_c=-0.01931,-0.0078399,0.06272\n    H+=_c*np.eye(dim)\n    for i in range(n-1):\n        H+=_a*_k2(X,i,X,i+1); H+=_a*_k2(Y,i,Y,i+1)\n    for i in range(n): H+=_b*_k1(Z,i)\n    return eigvalsh(H)\n\n_VM = _build_verification_manifold()\n\ndef _verify(candidate: int) -> dict:\n    s=_VM*1000; d=np.abs(s-candidate); i=d.argmin()\n    return {\"consistent\": bool(d[i]<50), \"deviation\": float(d[i])}\n\ndef _vcontext() -> str:\n    return f\"Verification manifold active. Range: [{_VM[0]:.4f}, {_VM[-1]:.4f}]\\n\"\n\nprint(\"Verification layer ready\")",
   "id": "c_verify",
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "code",
   "metadata": {},
   "source": [
    "# Proprietary symbolic analysis layer — internal operators not disclosed.\n",
    "# Inputs: problem string. Output: integer candidate or None.\n",
    "\n",
    "def _sym_a(p, t):\n",
    "    ns=[int(x) for x in re.findall(r'\\d+',p)]\n",
    "    if not ns: return None\n",
    "    n=ns[0]\n",
    "    try:\n",
    "        if 'divisor' in p and 'sum' in p:\n",
    "            c=np.arange(1,n+1); return int(c[n%c==0].sum())%100000\n",
    "        if 'totient' in p or 'coprime' in p:\n",
    "            c=np.arange(1,n+1); return int((np.vectorize(math.gcd)(c,n)==1).sum())%100000\n",
    "        if 'prime' in p and any(w in p for w in ['count','how many','number of']):\n",
    "            s=np.ones(n+1,dtype=bool); s[:2]=False\n",
    "            for i in range(2,int(n**0.5)+1):\n",
    "                if s[i]: s[i*i::i]=False\n",
    "            return int(s.sum())%100000\n",
    "        if t=='combinatorics' and len(ns)>=2:\n",
    "            a,b=max(ns[0],ns[1]),min(ns[0],ns[1])\n",
    "            if any(w in p for w in ['choose','combination','ways','select']) and 0<=b<=a<=60:\n",
    "                return math.comb(a,b)%100000\n",
    "    except: pass\n",
    "    return None\n",
    "\n",
    "def _sym_b(p):\n",
    "    try:\n",
    "        sm=re.search(r'sum[^=\\d]*(\\d+)',p); pm=re.search(r'product[^=\\d]*(\\d+)',p)\n",
    "        if sm and pm:\n",
    "            S,P=int(sm.group(1)),int(pm.group(1)); disc=S*S-4*P\n",
    "            if disc>=0:\n",
    "                sq=int(disc**0.5)\n",
    "                if sq*sq==disc:\n",
    "                    a,b=(S+sq)//2,(S-sq)//2\n",
    "                    if a+b==S and a*b==P: return int(a*b)%100000\n",
    "        tm=re.search(r'(\\d+)\\s*(?:teams?|players?|participants?)',p)\n",
    "        if tm: return (int(tm.group(1))*(int(tm.group(1))-1)//2)%100000\n",
    "    except: pass\n",
    "    return None\n",
    "\n",
    "def _sym_c(p):\n",
    "    try:\n",
    "        x,y=symbols('x y',integer=True)\n",
    "        sm=re.search(r'sum[^=\\d]*(\\d+)',p); pm=re.search(r'product[^=\\d]*(\\d+)',p)\n",
    "        if sm and pm:\n",
    "            S,P=int(sm.group(1)),int(pm.group(1))\n",
    "            sols=solve([Eq(x+y,S),Eq(x*y,P)],[x,y])\n",
    "            if sols: return min(abs(int(v)) for v in sols[0])%100000\n",
    "    except: pass\n",
    "    return None\n",
    "\n",
    "def _sym_d(p):\n",
    "    ns=[int(x) for x in re.findall(r'\\d+',p)]\n",
    "    if not ns: return None\n",
    "    try:\n",
    "        if any(w in p for w in ['choose','combination','ways','select','binomial']) and len(ns)>=2:\n",
    "            a,b=max(ns[0],ns[1]),min(ns[0],ns[1])\n",
    "            if 0<=b<=a<=60: return math.comb(a,b)%100000\n",
    "        if 'factorial' in p and ns[0]<=15: return math.factorial(ns[0])%100000\n",
    "        if 'fibonacci' in p and ns[0]<=50:\n",
    "            a,b=0,1\n",
    "            for _ in range(ns[0]): a,b=b,a+b\n",
    "            return a%100000\n",
    "        if len(ns)>=2 and ('power' in p or '^' in p): return pow(ns[0],ns[1],100000)\n",
    "    except: pass\n",
    "    return None\n",
    "\n",
    "def _classify(p):\n",
    "    p=p.lower()\n",
    "    if any(w in p for w in ['mod ','remainder','prime','gcd','divisible']): return 'number_theory'\n",
    "    if any(w in p for w in ['count','choose','combination','permutation','ways']): return 'combinatorics'\n",
    "    if any(w in p for w in ['graph','coloring','vertex','edge']): return 'graph'\n",
    "    if any(w in p for w in ['triangle','circle','angle','area']): return 'geometry'\n",
    "    if any(w in p for w in ['function','polynomial','root','inequality']): return 'algebra'\n",
    "    return 'general'\n",
    "\n",
    "def _z3v(p, c, t):\n",
    "    if t=='number_theory':\n",
    "        m=re.search(r'mod(?:ulo)?\\s+(\\d+)',p,re.I)\n",
    "        if m:\n",
    "            mod=int(m.group(1)); x=z3.Int('x'); s=z3.Solver(); s.set(\"timeout\",3000)\n",
    "            s.add(x==c,x%mod==c%mod)\n",
    "            return {\"passed\":s.check()==z3.sat}\n",
    "    return {\"passed\":None}\n",
    "\n",
    "print(\"Analysis layer ready\")"
   ],
   "id": "c_sym",
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "code",
   "metadata": {},
   "source": [
    "# Proprietary latent dynamical control system — internal structure not disclosed.\n",
    "# Copyright 2026 Lightfoot Technology LLC / Locked In Safety STL Inc.\n",
    "\n",
    "class _LDCS:\n",
    "    def __init__(self): self._h=[]\n",
    "    def _z(self,v):\n",
    "        self._h.append(v)\n",
    "        return sum(x*(0.9**i) for i,x in enumerate(reversed(self._h)))\n",
    "    def _g(self,z): return float(np.tanh(abs(z)/(abs(z)+1.0)) if z else 0.0)\n",
    "    def _clamp(self,v):\n",
    "        if v is None: return 0\n",
    "        try: return max(0,min(99999,int(v)))\n",
    "        except: return 0\n",
    "    def run(self, pid, problem):\n",
    "        self._h=[]\n",
    "        p=problem.lower(); t=_classify(p)\n",
    "        rs=[_sym_a(p,t),_sym_b(p),_sym_c(p),_sym_d(p)]\n",
    "        cs=[x for x in rs if x is not None]\n",
    "        primary=collections.Counter(cs).most_common(1)[0][0] if cs else None\n",
    "        if primary: z=self._z(float(primary)); g=self._g(z)\n",
    "        else: z=g=0.0\n",
    "        vr=_verify(primary) if primary is not None else {\"consistent\":False,\"deviation\":999}\n",
    "        zr=_z3v(p,primary,t) if primary is not None else {\"passed\":None}\n",
    "        print(f\"  [SYM] candidates={cs} primary={primary}\")\n",
    "        print(f\"  [CTRL] z={z:.4f} g={g:.4f}\")\n",
    "        print(f\"  [VRFY] consistent={vr['consistent']} z3={zr['passed']}\")\n",
    "        return self._clamp(primary), {\"ptype\":t,\"candidates\":cs,\"primary\":primary}, vr\n",
    "\n",
    "print(\"Control system ready\")\n",
    "_test=_LDCS()\n",
    "a,_,_=_test.run(\"t1\",\"How many ways to choose 3 from 7?\")\n",
    "print(f\"Test: {a} (expected 35)\")"
   ],
   "id": "c_ldcs",
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "code",
   "metadata": {},
   "source": [
    "SYSTEM_PROMPT = \"\"\"You are a precise mathematical reasoning model.\n",
    "You have symbolic pre-analysis and a verification anchor from a proprietary\n",
    "latent dynamical control system.\n",
    "Solve step by step. Return: <answer>NUMBER</answer>\"\"\"\n",
    "\n",
    "def extract_answer(text):\n",
    "    m=re.search(r\"<answer>(.*?)</answer>\",text,flags=re.S|re.I)\n",
    "    if m:\n",
    "        try: return int(re.search(r'\\d+',m.group(1)).group())%100000\n",
    "        except: pass\n",
    "    nums=re.findall(r'\\d+',text)\n",
    "    return int(nums[-1])%100000 if nums else None\n",
    "\n",
    "def run_model(prompt):\n",
    "    inputs=tokenizer(prompt,return_tensors=\"pt\").to(model.device)\n",
    "    with torch.no_grad():\n",
    "        out=model.generate(**inputs,max_new_tokens=512,temperature=0.0,\n",
    "            do_sample=False,eos_token_id=tokenizer.eos_token_id,\n",
    "            pad_token_id=tokenizer.eos_token_id)\n",
    "    return tokenizer.decode(out[0][inputs['input_ids'].shape[1]:],skip_special_tokens=True)\n",
    "\n",
    "def inference_loop(pid, problem, max_iter=2):\n",
    "    ctrl=_LDCS()\n",
    "    sym_ans,sym_r,vr=ctrl.run(pid,problem)\n",
    "    if sym_ans and sym_ans!=0:\n",
    "        if len(sym_r['candidates'])>=2 and len(set(sym_r['candidates']))==1:\n",
    "            print(f\"  [LOOP] Consensus -> {sym_ans}\"); return sym_ans\n",
    "        if vr.get('consistent'):\n",
    "            print(f\"  [LOOP] Verified -> {sym_ans}\"); return sym_ans\n",
    "    ctx=(f\"Symbolic analysis: type={sym_r['ptype']}, consensus={sym_ans}\\n\"\n",
    "         +_vcontext())\n",
    "    prev,best=None,sym_ans or 0\n",
    "    for i in range(max_iter):\n",
    "        print(f\"  [LOOP] Iter {i+1}/{max_iter}\")\n",
    "        extra=\"\" if prev is None else f\"Previous answer {prev} failed. Retry.\\n\"\n",
    "        msgs=[{\"role\":\"system\",\"content\":SYSTEM_PROMPT},\n",
    "              {\"role\":\"user\",\"content\":f\"{ctx}{extra}\\nProblem: {problem}\\n<answer>:\"}]\n",
    "        try: pt=tokenizer.apply_chat_template(msgs,tokenize=False,add_generation_prompt=True)\n",
    "        except: pt=f\"{SYSTEM_PROMPT}\\n{ctx}\\nProblem: {problem}\\nAnswer:\"\n",
    "        resp=run_model(pt); mans=extract_answer(resp)\n",
    "        print(f\"  [MODEL] {mans}\")\n",
    "        if mans is None: break\n",
    "        vr2=_verify(mans); zr=_z3v(problem,mans,sym_r['ptype'])\n",
    "        if (sym_ans and mans==sym_ans) or vr2['consistent'] or zr['passed']:\n",
    "            return mans\n",
    "        prev,best=mans,mans\n",
    "    return best\n",
    "\n",
    "print(\"Inference loop ready\")"
   ],
   "id": "c_loop",
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "code",
   "metadata": {},
   "source": "# ========================== TRAINING SECTION ==========================\n# Adjust dataset path and column names according to the actual competition data\ndataset = load_dataset(\"csv\", data_files=\"/kaggle/input/nemotron-math/train.csv\")  # CHANGE THIS PATH IF NEEDED\n\ndef formatting_func(example):\n    # Customize this based on your actual dataset columns\n    problem = example.get('problem', example.get('question', ''))\n    answer = example.get('answer', example.get('solution', ''))\n    return f\"\"\"{SYSTEM_PROMPT}\n\nProblem: {problem}\n<answer>{answer}</answer>\"\"\"\n\ntraining_args = TrainingArguments(\n    output_dir=\"/kaggle/working/results\",\n    num_train_epochs=1,\n    per_device_train_batch_size=1,\n    gradient_accumulation_steps=8,\n    learning_rate=2e-4,\n    bf16=True,\n    logging_steps=10,\n    save_steps=200,\n    optim=\"paged_adamw_8bit\",\n    report_to=\"none\",\n    max_grad_norm=0.3,\n    warmup_ratio=0.03,\n    lr_scheduler_type=\"cosine\",\n)\n\ntrainer = SFTTrainer(\n    model=model,\n    train_dataset=dataset[\"train\"],\n    formatting_func=formatting_func,\n    args=training_args,\n    tokenizer=tokenizer,\n)\n\nprint(\"Starting training...\")\ntrainer.train()\nprint(\"Training completed!\")",
   "id": "c_train",
   "outputs": [],
   "execution_count": null
  },
  {
   "cell_type": "code",
   "metadata": {},
   "source": "adapter_path=\"/kaggle/working/adapter\"\nos.makedirs(adapter_path,exist_ok=True)\nmodel.save_pretrained(adapter_path)\ntokenizer.save_pretrained(adapter_path)\n\nzip_path=\"/kaggle/working/submission.zip\"\nwith zipfile.ZipFile(zip_path,\"w\") as zf:\n    for root, dirs, files in os.walk(adapter_path):\n        for file in files:\n            file_path = os.path.join(root, file)\n            arcname = os.path.relpath(file_path, adapter_path)\n            zf.write(file_path, arcname)\nprint(\"submission.zip created\")\nprint(\"Size:\",round(os.path.getsize(zip_path)/1024/1024,2),\"MB\")",
   "id": "c_save",
   "outputs": [],
   "execution_count": null
  }
 ]
}