frankmorales2020/glm-4.6-topo-2026

CODE: https://github.com/frank-morales2020/AST/blob/main/TOPO2026_GLM_DEMO_CORRECTED.ipynb

INFERENCE

# ============================================================================
# INFERENCE (CORRECTED): reload the TOPO-2026 classifier and run predictions
# ============================================================================
# IMPORTANT: what was trained is a 3-task TEXT CLASSIFIER on top of a FROZEN
# GLM base. It cannot "chat" / generate free text -- that capability was never
# trained. So the correct test is classification, which is what this cell does.
#
# Requires the same classes/utilities defined in the training cell:
#   TaskAwareGLM, get_embed_layer, MODEL_NAME, BitsAndBytesConfig, etc.
# Run the training cell (or at least its class/import definitions) first.

import torch
from transformers import AutoModel, AutoTokenizer, BitsAndBytesConfig
from huggingface_hub import hf_hub_download

REPO_ID = "frankmorales2020/glm-4.6-topo-2026"
device  = torch.device("cuda" if torch.cuda.is_available() else "cpu")

print("📥 Loading stock GLM base in 4-bit (same as training)...")
bnb = BitsAndBytesConfig(load_in_4bit=True, bnb_4bit_quant_type="nf4",
                         bnb_4bit_use_double_quant=True,
                         bnb_4bit_compute_dtype=torch.bfloat16)
base_model = AutoModel.from_pretrained(
    MODEL_NAME, trust_remote_code=True, quantization_config=bnb,
    device_map="auto", torch_dtype=torch.bfloat16)
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME, trust_remote_code=True)
if tokenizer.pad_token is None:
    tokenizer.pad_token = tokenizer.eos_token

print("📥 Rebuilding TaskAwareGLM and loading trained parts...")
model = TaskAwareGLM(base_model).to(device)

ckpt = torch.load(hf_hub_download(REPO_ID, "topo_trained_parts.pt"),
                  map_location="cpu", weights_only=False)
model.classifier_A.load_state_dict({k: v.to(model.classifier_A.weight.dtype)
                                    for k, v in ckpt["classifier_A"].items()})
model.classifier_B.load_state_dict({k: v.to(model.classifier_B.weight.dtype)
                                    for k, v in ckpt["classifier_B"].items()})
model.classifier_C.load_state_dict({k: v.to(model.classifier_C.weight.dtype)
                                    for k, v in ckpt["classifier_C"].items()})

# Restore trained embedding rows (copy into the live embedding table)
with torch.no_grad():
    emb = get_embed_layer(base_model)
    w = ckpt["embed_tokens_weight"].to(emb.weight.dtype).to(emb.weight.device)
    if w.shape == emb.weight.shape:
        emb.weight.copy_(w)
        print("  ✅ embedding weights restored")
    else:
        print(f"  ⚠️ embedding shape mismatch {tuple(w.shape)} vs {tuple(emb.weight.shape)}; skipped")
model.eval()
print("✅ Classifier ready. Anchors:", ckpt["prime_anchors"],
      "| Λ =", round(ckpt["safety_constant"], 6))

TASK_LABELS = {
    "A": ["World", "Sports"],
    "B": ["Business", "Science"],
    "C": ["Politics", "Technology"],
}

@torch.no_grad()
def classify(text, task):
    model.switch_task(task)
    tok = tokenizer([text], return_tensors="pt", padding=True,
                    truncation=True, max_length=ckpt.get("max_len", 64)).to(device)
    logits = model(tok.input_ids, tok.attention_mask)
    probs = torch.softmax(logits, dim=1)[0]
    idx = int(torch.argmax(probs))
    return TASK_LABELS[task][idx], float(probs[idx])

print("\n" + "="*80 + "\n🚀 CLASSIFICATION TESTS\n" + "="*80)
tests = [
    ("World summit on climate policy",        "A"),
    ("The team won the championship match",   "A"),
    ("Quarterly business earnings report",    "B"),
    ("New scientific research breakthrough",  "B"),
    ("Government election and policy debate", "C"),
    ("Latest smartphone technology release",  "C"),
]
for text, task in tests:
    label, conf = classify(text, task)
    print(f"[{task}] {text!r:45s} -> {label:12s} ({conf*100:.1f}%)")

print("\n" + "="*80 + "\n🏁 INFERENCE COMPLETE\n" + "="*80)