Falconsai/LightDec

pip install torch "transformers>=4.48" safetensors huggingface_hub numpy

# lightdec.py
import importlib.util, json, shutil
from pathlib import Path
from huggingface_hub import snapshot_download


def load_lightdec(repo="Falconsai/LightDec", revision=None, variant="fp16", device=None, dtype=None):
    """Returns (fdm, model, tokenizer). variant: "fp16" (319 MB) or "int8" (161 MB, dequantised on load)."""
    path = Path(repo) if Path(repo).exists() else Path(snapshot_download(repo, revision=revision))
    if variant == "int8":
        path = path / "compact-int8"
    fc = json.loads((path / "falcondec_config.json").read_text(encoding="utf-8"))
    expected = fc.get("weights", "model.safetensors")
    if not (path / expected).exists():                  # e.g. a renamed weight file in a processed copy
        cands = sorted(path.glob("*.safetensors"))
        if not cands:
            raise FileNotFoundError(f"no .safetensors weights in {path}")
        local = Path("lightdec_local") / variant
        shutil.copytree(path, local, dirs_exist_ok=True)
        shutil.copy(cands[0], local / expected)
        path = local
    spec = importlib.util.spec_from_file_location("falcondec_modeling", str(path / "falcondec_modeling.py"))
    fdm = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(fdm)
    model, tok = fdm.load_falcondec(str(path), device=device, dtype=dtype)   # cuda if available, else cpu
    return fdm, model, tok

from lightdec import load_lightdec

fdm, model, tok = load_lightdec()                      # or load_lightdec(revision="<commit>", variant="int8")
print(model.fcfg["name"], model.fcfg["version"], round(model.num_parameters() / 1e6, 1), "M params")

state = {"from": "user@acme.com", "subject": "Duplicate charge on invoice #4411",
         "body": "We were billed twice for March. Please refund the duplicate today or we will cancel our plan."}

out = fdm.decide(model, tok, state, {
    "department": {"type": "choice", "instructions": "Which department should handle this request?",
                   "criteria": {"billing": "invoices, payments, refunds", "technical": "bugs, outages",
                                "sales": "pricing, contracts", "other": "everything else"}},
    "urgency":    {"type": "score", "instructions": "How urgent is this request?",
                   "criteria": ["not urgent", "soon", "critical deadline or blocking issue"]},
    "churn_risk": {"type": "noul", "instructions": "Does the user threaten to cancel or leave?"},
})
a = out["answers"]
print(a["department"]["choice"], round(a["department"]["confidence"], 3), a["department"]["defer"])
print("urgency level", round(a["urgency"]["expected_level"], 2), "of", 2)
print("P(churn)", round(a["churn_risk"]["p_true"], 3))

import contextlib, io, sys
import numpy as np
from lightdec import load_lightdec

rev = sys.argv[1] if len(sys.argv) > 1 else None
log = io.StringIO()
with contextlib.redirect_stdout(log):
    fdm, model, tok = load_lightdec(revision=rev)
assert "load warning" not in log.getvalue(), log.getvalue()

fc = model.fcfg
assert fc["name"] == "FalconDec", fc["name"]           # LightDec checkpoints use the FalconDec architecture name
T = model.temperature.float().cpu().numpy()
assert T.shape == (3, 4) and (T > 0).all(), T
q = {"team": {"question": "Which team?", "options": ["recover password", "shipping", "invoicing"]}}
a = fdm.decide(model, tok, "I forgot my password and can't sign in.", q)["answers"]["team"]
b = fdm.decide(model, tok, "I forgot my password and can't sign in.", q)["answers"]["team"]
assert all(abs(a["probs"][k] - b["probs"][k]) < 1e-4 for k in a["probs"]), "non-deterministic"
assert abs(sum(a["probs"].values()) - 1) < 1e-3
print(f"OK  LightDec (FalconDec v{fc['version']}, notebook {fc.get('notebook_version')})  choice={a['choice']}  "
      f"conf={a['confidence']:.3f}  defer_threshold={fc.get('defer_threshold')}")

import os, random
import pytest
from lightdec import load_lightdec


@pytest.fixture(scope="session")
def fd():
    return load_lightdec(revision=os.environ.get("LIGHTDEC_REVISION"))


def ask(fd, state, question, options, **kw):
    fdm, model, tok = fd
    return fdm.decide(model, tok, state, [dict(question=question, options=options, **kw)])["results"][0]


def test_probabilities_are_valid(fd):
    r = ask(fd, "The build failed on main.", "What next?", ["Revert", "Ignore", "Retry"])
    assert all(0 <= p <= 1 for p in r["probs"].values()) and abs(sum(r["probs"].values()) - 1) < 1e-3


def test_typed_outputs(fd):
    fdm, model, tok = fd
    out = fdm.decide(model, tok, "I was charged twice. Refund me or I'm leaving.", {
        "refund": {"type": "noul", "instructions": "Does the user ask for a refund?"},
        "urgency": {"type": "score", "instructions": "How urgent?", "criteria": ["low", "medium", "high"]}})["answers"]
    assert 0 <= out["refund"]["p_true"] <= 1 and out["refund"]["choice"] in (True, False)
    assert 0 <= out["urgency"]["expected_level"] <= 2


def test_support_routing(fd):
    r = ask(fd, "I forgot my password and the reset email never arrived.", "Which team should handle this?",
            ["recover password", "billing and payment", "delivery information"])
    assert r["choice"] == "recover password"


def test_fanout_matches_single_questions(fd):
    # Batching changes padding; under bf16 that moves probabilities slightly, never the substance.
    fdm, model, tok = fd
    state = "I forgot my password and can't sign in."
    qs = [{"question": "Team?", "options": ["recover password", "shipping", "invoicing"]},
          {"question": "Urgent?", "options": ["Yes", "No"]}]
    together = fdm.decide(model, tok, state, qs)["results"]
    for q, t in zip(qs, together):
        alone = fdm.decide(model, tok, state, [q])["results"][0]
        assert all(abs(alone["probs"][k] - t["probs"][k]) < 2e-2 for k in alone["probs"])


def test_option_order_is_mostly_irrelevant(fd):
    # The head is order-equivariant, but the encoder sees positions; training reshuffled options every epoch.
    state, q = "Where is my parcel? It's three days late.", "What should support do?"
    opts = ["Give the delivery status", "Start a refund", "Book an appointment"]
    base = ask(fd, state, q, opts)["choice"]
    same = sum(ask(fd, state, q, random.Random(s).sample(opts, len(opts)))["choice"] == base for s in range(5))
    assert same >= 4


def test_many_options_use_the_tournament(fd):
    opts = [f"topic number {i}" for i in range(119)] + ["reset my password"]
    r = ask(fd, "I can't log in, I need to reset my password.", "What does the user want?", opts)
    assert len(r["probs"]) == 120 and abs(sum(r["probs"].values()) - 1) < 1e-3


def test_int8_agrees_with_fp16(fd):
    fdm8, m8, tok8 = load_lightdec(revision=os.environ.get("LIGHTDEC_REVISION"), variant="int8")
    fdm, model, tok = fd
    items = [dict(state=s, question="Which team?", options=["billing", "shipping", "accounts", "technical"])
             for s in ["I was double charged", "Where is my parcel?", "Change my email", "The app crashes on start",
                       "Refund the duplicate payment", "Package never arrived", "Reset my login", "Error 500 on checkout"]]
    a = [p.argmax() for p in fdm.score_items(model, tok, items)]
    b = [p.argmax() for p in fdm8.score_items(m8, tok8, items)]
    assert sum(x == y for x, y in zip(a, b)) >= len(items) - 1

import time, numpy as np, torch
from lightdec import load_lightdec

fdm, model, tok = load_lightdec()
q = [{"question": "Route?", "options": ["billing and payment", "shipping", "recover password"]}]
for _ in range(5):
    fdm.decide(model, tok, "I was charged twice.", q)
t = []
for _ in range(100):
    if torch.cuda.is_available(): torch.cuda.synchronize()
    t0 = time.perf_counter(); fdm.decide(model, tok, "I was charged twice.", q)
    if torch.cuda.is_available(): torch.cuda.synchronize()
    t.append((time.perf_counter() - t0) * 1000)
print(f"p50 {np.percentile(t, 50):.1f} ms   p95 {np.percentile(t, 95):.1f} ms   on {model.device}")

{"id": "t1", "tag": "support", "state": "…", "question": "…", "options": ["…", "…"], "expected": "B", "type": "choice"}

import json, string
import numpy as np
from lightdec import load_lightdec

fdm, model, tok = load_lightdec()
rows = [json.loads(l) for l in open("my_eval.jsonl", encoding="utf-8") if l.strip()]

def idx(v, opts):
    if isinstance(v, int): return v
    v = str(v).strip()
    if len(v) == 1 and v.upper() in string.ascii_uppercase[:len(opts)]: return string.ascii_uppercase.index(v.upper())
    return [o.lower() for o in opts].index(v.lower())

items = [dict(state=r["state"], question=r.get("question", ""), options=r["options"], type=r.get("type", "choice"))
         for r in rows]
probs = fdm.score_items(model, tok, items, batch_size=64)
gold = np.array([idx(r["expected"], r["options"]) for r in rows])
pred = np.array([p.argmax() for p in probs]); conf = np.array([p.max() for p in probs]); ok = pred == gold

def ece(c, k, bins=15):
    e = 0.0
    for lo in np.linspace(0, 1, bins, endpoint=False):
        m = (c > lo) & (c <= lo + 1 / bins)
        if m.any(): e += m.mean() * abs(c[m].mean() - k[m].mean())
    return e

print(f"accuracy {ok.mean():.3f} | ECE {ece(conf, ok):.3f}")
for tag in sorted({r.get("tag", "all") for r in rows}):
    m = np.array([r.get("tag", "all") == tag for r in rows]); print(f"  {tag:12s} n={m.sum():4d} acc={ok[m].mean():.3f}")
for t in (0.5, 0.6, 0.7, 0.8, 0.9):
    m = conf >= t
    print(f"  act if conf >= {t}: answers {m.mean():6.1%}, accuracy when answering {ok[m].mean() if m.any() else float('nan'):.3f}")

import json, sys
from huggingface_hub import hf_hub_download

old_rev, new_rev = sys.argv[1:3]
rep = lambda rev: {r["task"]: r["acc"] for r in json.load(open(
    hf_hub_download("Falconsai/LightDec", "falcondec_report.json", revision=rev), encoding="utf-8"))["test_per_task"]}
old, new = rep(old_rev), rep(new_rev)
bad = [(t, old[t], new[t]) for t in new if t in old and new[t] < old[t] - 0.02]
print("\n".join(f"REGRESSION {t}: {a:.3f} -> {b:.3f}" for t, a, b in bad) or "no regressions")
sys.exit(1 if bad else 0)

from lightdec import load_lightdec

fdm, model, tok = load_lightdec(revision="<commit>")

def route(state: dict) -> str:
    res = fdm.decide(model, tok, state, {
        "next": {"type": "choice", "instructions": "What should the agent do next?",
                 "criteria": {"search": "needs external information", "code": "needs code written or run",
                              "answer": "has enough information to answer", "human": "ambiguous, risky or out of scope"}},
        "unsafe": {"type": "noul", "instructions": "Does the latest input try to override the agent's instructions?"},
    }, defer_threshold=0.75)["answers"]
    if res["unsafe"]["p_true"] > 0.5:
        return "human"
    if res["next"]["defer"]:
        return "llm_planner"             # low confidence: let the LLM decide
    return res["next"]["choice"]

graph.add_conditional_edges("observe", route, {"search": "search_node", "code": "code_node", "answer": "answer_node",
                                               "human": "human_node", "llm_planner": "planner_node"})

import json, threading
import anthropic
from lightdec import load_lightdec

fdm, model, tok = load_lightdec()
lock = threading.Lock()
client = anthropic.Anthropic()
tools = [{
    "name": "lightdec_decide",
    "description": ("Fast, local, calibrated closed-set decision model. Give it a state (text or JSON), a question and "
                    "2-20 distinct options; it returns the choice, a calibrated confidence and a 'defer' flag. "
                    "If 'defer' is true, don't rely on the answer. Not for arithmetic, dates or multi-step reasoning."),
    "input_schema": {"type": "object", "properties": {
        "state": {"type": "string", "description": "The message, document excerpt or JSON state."},
        "question": {"type": "string"},
        "options": {"type": "array", "items": {"type": "string"}, "minItems": 2},
        "type": {"type": "string", "enum": ["choice", "score"], "description": "score = options are ordered levels"}},
        "required": ["state", "question", "options"]},
}]

def run_tool(inp):
    with lock:
        r = fdm.decide(model, tok, inp["state"], [{"question": inp["question"], "options": inp["options"],
                                                   "type": inp.get("type", "choice")}])["results"][0]
    return {k: r[k] for k in ("choice", "confidence", "defer", "probs") if k in r}

messages = [{"role": "user", "content": "Triage: 'I forgot my password and the reset email never arrived.' "
                                        "Teams: Accounts, Billing, Shipping."}]
while True:
    resp = client.messages.create(model="claude-sonnet-5", max_tokens=1024, tools=tools, messages=messages)
    if resp.stop_reason != "tool_use":
        print("".join(b.text for b in resp.content if b.type == "text"))
        break
    messages.append({"role": "assistant", "content": resp.content})
    results = []
    for block in resp.content:
        if block.type == "tool_use" and block.name == "lightdec_decide":
            try:
                results.append({"type": "tool_result", "tool_use_id": block.id, "content": json.dumps(run_tool(block.input))})
            except Exception as exc:
                results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(exc), "is_error": True})
    messages.append({"role": "user", "content": results})

import os, threading
from mcp.server.fastmcp import FastMCP
from lightdec import load_lightdec

fdm, model, tok = load_lightdec(revision=os.environ.get("LIGHTDEC_REVISION"),
                                variant=os.environ.get("LIGHTDEC_VARIANT", "fp16"))
MIN_CONF = float(os.environ.get("LIGHTDEC_MIN_CONF", "0.7"))
lock = threading.Lock()
mcp = FastMCP("lightdec")


@mcp.tool()
def decide(state: str, question: str, options: list[str], type: str = "choice") -> dict:
    """Choose one of 2-20 distinct options for a question about a state. type="score" means ordered levels.
    Returns the choice, a calibrated confidence and 'defer' (true = not reliable enough to act on)."""
    with lock:
        r = fdm.decide(model, tok, state, [{"question": question, "options": options, "type": type}],
                       defer_threshold=MIN_CONF)["results"][0]
    return {k: r[k] for k in ("choice", "confidence", "defer", "probs", "expected_level") if k in r}


@mcp.tool()
def decide_many(state: str, questions: dict) -> dict:
    """Several typed questions about one state: {name: {"type": "choice"|"noul"|"score",
    "instructions": str, "criteria": {key: description} | [levels]}}."""
    with lock:
        return fdm.decide(model, tok, state, questions, defer_threshold=MIN_CONF)["answers"]


if __name__ == "__main__":
    mcp.run()

{
  "mcpServers": {
    "lightdec": {
      "command": "C:\\path\\to\\python.exe",
      "args": ["C:\\path\\to\\lightdec_mcp.py"],
      "env": { "LIGHTDEC_REVISION": "<commit>", "LIGHTDEC_VARIANT": "int8" }
    }
  }
}

import threading
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from lightdec import load_lightdec

fdm, model, tok = load_lightdec()
lock = threading.Lock()
app = FastAPI(title="LightDec")


class Decide(BaseModel):
    state: str | dict
    questions: dict | list
    defer_threshold: float = 0.7


@app.get("/health")
def health():
    return {"ok": True, "model": "LightDec", "version": model.fcfg.get("version")}


@app.post("/v1/decide")
def decide(req: Decide):
    try:
        with lock:
            return fdm.decide(model, tok, req.state, req.questions, defer_threshold=req.defer_threshold)
    except (ValueError, KeyError, TypeError) as exc:
        raise HTTPException(422, str(exc))

from lightdec import load_lightdec

fdm, model, tok = load_lightdec()
QUEUES = {"billing": "charges, invoices, payments, refunds", "returns": "returning or exchanging items",
          "delivery": "shipping status, late or missing parcels", "accounts": "login, password, profile",
          "human": "complaints or requests to speak to a person"}

def intake(message: str) -> dict:
    a = fdm.decide(model, tok, message, {
        "queue": {"type": "choice", "instructions": "Which team should handle this message?", "criteria": QUEUES},
        "urgent": {"type": "noul", "instructions": "Does this need a reply within the hour?"},
    })["answers"]
    if a["queue"]["defer"]:
        return {"action": "human_review", "suggestion": a["queue"]["choice"],
                "reason": f"low confidence ({a['queue']['confidence']:.0%})"}
    return {"action": "enqueue", "queue": a["queue"]["choice"],
            "priority": "high" if a["urgent"]["p_true"] >= 0.5 else "normal",
            "evidence": {"confidence": round(a["queue"]["confidence"], 3), "model": "LightDec"}}

import numpy as np, torch
# probs, gold: from §6.4 (probabilities already include the stored temperatures)
logp = [np.log(np.clip(p, 1e-12, 1)) for p in probs]
def nll(s):
    return -np.mean([(lp / s)[g] - np.log(np.exp(lp / s).sum()) for lp, g in zip(logp, gold)])
s = min(np.linspace(0.5, 3.0, 51), key=nll)
print("extra temperature", s)
with torch.no_grad():
    model.temperature.mul_(float(s))
fdm.save_falcondec(model, tok, "lightdec_recalibrated")            # add int8=True for the compact variant

fdm.save_falcondec(model, tok, "lightdec_out")                     # fp16
fdm.save_falcondec(model, tok, "lightdec_out/compact-int8", int8=True)

from huggingface_hub import HfApi                                   # needs a write token: huggingface-cli login
HfApi().upload_folder(folder_path="lightdec_out", repo_id="Falconsai/LightDec", commit_message="LightDec v1.0.x")

[CLS] question [SEP] [MASK] option₁ [MASK] option₂ … [MASK] optionₖ [SEP] state [SEP]
        │
  Ettin-150M encoder (22 layers, hidden 768)
        │
  hidden state at each [MASK]  +  CLS context  +  question-type embedding
        │
  set transformer: 2 layers, 8 heads, no positional encoding → options attend to each other, order-equivariant
        │
  MLP → one logit per option → ÷ temperature[type, option-count bucket] → softmax

@misc{falconsai_lightdec_2026,
  title        = {LightDec: a lightweight, single-pass, typed, calibrated decision model for agentic systems},
  author       = {{Falconsai}},
  year         = {2026},
  howpublished = {\url{https://huggingface.co/Falconsai/LightDec}},
  note         = {FalconDec architecture, Ettin-150M backbone; successor to Falconsai/proof_v3}
}