| """MiniCPM adapter for the pinned Apache-2.0 Kev pointer implementation. |
| |
| No Qwen token IDs are reused. Structural markers are five existing MiniCPM |
| unused tokens. All user angle brackets are escaped before tokenization. |
| Gold labels enter only the returned training targets, never the input payload. |
| """ |
| from copy import deepcopy |
|
|
| from decision_data import api_payload, gold_key, question_keys |
| from kev.api import SystemOneRequest, to_record |
| from kev.model import SPECIAL, DecisionModel, encode |
|
|
| MARKERS = ['<unused_token_' + str(i) + '>' for i in range(5)] |
| |
| |
| |
| MARKER_SETS = {'minicpm': MARKERS, 'kev': list(SPECIAL)} |
|
|
|
|
| def choose_markers(tokenizer): |
| """(family, markers, ids): the first family whose five markers exist as distinct single tokens.""" |
| for family, markers in MARKER_SETS.items(): |
| ids = [tokenizer.convert_tokens_to_ids(t) for t in markers] |
| if len(set(ids)) != 5 or any(i is None or i == tokenizer.unk_token_id for i in ids): |
| continue |
| if all(tokenizer(t, add_special_tokens=False).input_ids == [i] for t, i in zip(markers, ids)): |
| return family, list(markers), ids |
| raise ValueError('No vocabulary family with five distinct existing single-token structural markers') |
|
|
|
|
| class StructuralTokenizer: |
| def __init__(self, tokenizer): |
| self.tokenizer = tokenizer |
| self.family, self.markers, self.marker_ids = choose_markers(tokenizer) |
| self.mapping = dict(zip(SPECIAL, self.markers)) |
| self.pad_token_id = tokenizer.pad_token_id |
|
|
| def convert_tokens_to_ids(self, text): |
| return self.tokenizer.convert_tokens_to_ids(self.mapping.get(text, text)) |
|
|
| def __call__(self, text, **kwargs): |
| |
| return self.tokenizer(text.replace('<', '＜').replace('>', '＞'), **kwargs) |
|
|
|
|
| def encode_record(tokenizer, record, max_tokens=2048, option_isolation=False): |
| wire = api_payload(record) |
| rec, _ = to_record(SystemOneRequest.model_validate(wire)) |
| result = encode(tokenizer, rec, strict=True, max_state=max_tokens, |
| max_branch=max_tokens, option_isolation=option_isolation) |
| if len(result['ids']) > max_tokens: |
| raise ValueError('Packed input exceeds the declared context cap') |
| result['labels'] = [question_keys(q).index(gold_key(q)) for q in record['questions'].values()] |
| result['question_ids'] = list(record['questions']) |
| result['soft'] = [soft_target(q) for q in record['questions'].values()] |
| result['ordinal'] = [q['type'] == 'score' for q in record['questions'].values()] |
| |
| result['option_keys'] = [question_keys(q) for q in record['questions'].values()] |
| return result |
|
|
|
|
| def ranked_probability_score(probabilities, gold): |
| """Mean squared error between the predicted and the observed cumulative distribution. |
| |
| This is the objective the pinned Apache-2.0 Kev recipe records as `ranked_probability_score` for Score |
| questions. Unlike cross entropy it charges for how far the predicted mass sits from the true level, which |
| is what our error map needs: on the five-level replay questions 115 of our 142 errors are off by exactly |
| one level, so the ordering is nearly right and only the distance is unpriced. |
| """ |
| if len(probabilities) < 2 or not 0 <= gold < len(probabilities): |
| raise ValueError('Invalid ordered scale') |
| total, running = 0., 0. |
| for i in range(len(probabilities) - 1): |
| running += probabilities[i] |
| total += (running - (1. if i >= gold else 0.)) ** 2 |
| return total / (len(probabilities) - 1) |
|
|
|
|
| def spherical_score(probabilities, target): |
| """Spherical scoring rule: the target-weighted probability divided by the vector's own length. |
| |
| Taken from the open Laya implementation of this contract, whose reward is |
| `log score + 0.75 * spherical - 1.0 * ranked probability score`. It is strictly proper like log loss but |
| bounded, so it keeps rewarding a confident correct answer where log loss has already saturated, and it |
| punishes spreading mass thinly across options. Returned as a score to maximise; the trainer subtracts it. |
| """ |
| if len(probabilities) != len(target) or not probabilities: |
| raise ValueError('Distribution and target must share a length') |
| norm = sum(p * p for p in probabilities) ** .5 |
| if norm <= 0: |
| raise ValueError('Degenerate distribution') |
| return sum(t * p for t, p in zip(target, probabilities)) / norm |
|
|
|
|
| def ordinal_target(levels, gold, smoothing): |
| """Neighbour-smoothed target for an ordered scale. |
| |
| A plain one-hot target treats every wrong level as equally wrong, which is false for Score: |
| predicting 4 when the answer is 5 is not the same mistake as predicting 1. The smoothing mass is |
| split between the immediate neighbours, so the ordering is expressed in the target itself and the |
| loss stays an ordinary log loss. |
| """ |
| if not 0 <= smoothing < 1: |
| raise ValueError('Ordinal smoothing must be in [0, 1)') |
| if levels < 2 or not 0 <= gold < levels: |
| raise ValueError('Invalid ordered scale') |
| if smoothing == 0: |
| return None |
| target = [0.] * levels |
| target[gold] = 1 - smoothing |
| neighbours = [i for i in (gold - 1, gold + 1) if 0 <= i < levels] |
| for i in neighbours: |
| target[i] = smoothing / len(neighbours) |
| return target |
|
|
|
|
| def soft_target(question): |
| """Exact target distribution for a question that carries one, aligned with the option order. |
| |
| Returns None when the question has no distribution, or when augmentation changed the option set, |
| so such a question keeps the ordinary hard cross entropy. |
| """ |
| given = question.get('probabilities') |
| if not given: |
| return None |
| keys = question_keys(question) |
| if set(keys) != set(given): |
| return None |
| vector = [float(given[k]) for k in keys] |
| total = sum(vector) |
| if abs(total - 1) > 1e-6 or any(v < 0 for v in vector): |
| raise ValueError('Question probabilities are not a distribution') |
| return vector |
|
|
|
|
| def permute_choices(record, rng): |
| result = deepcopy(record) |
| for question in result['questions'].values(): |
| |
| if question['type'] == 'choice': |
| items = list(question['criteria'].items()) |
| rng.shuffle(items) |
| question['criteria'] = dict(items) |
| return result |
|
|
|
|
| def permuted_pair(record, rng): |
| """A copy with every choice question's options reordered, plus the mapping back to the original order. |
| |
| perms[q][j] is the original index of the option that now sits at position j, or None when the question |
| was left alone. Labels are option keys, so they stay correct without translation; only the logits need |
| realigning before the two predictions can be compared. |
| """ |
| result = deepcopy(record) |
| perms = [] |
| for question in result['questions'].values(): |
| if question['type'] == 'choice' and len(question['criteria']) >= 3: |
| items = list(question['criteria'].items()) |
| order = list(range(len(items))) |
| rng.shuffle(order) |
| question['criteria'] = dict(items[i] for i in order) |
| perms.append(order) |
| else: |
| perms.append(None) |
| return result, perms |
|
|
|
|
| def load_pointer(path, tokenizer, precision='bf16', option_isolation=False, lora_targets='all', lora_rank=16, |
| revision=None, train_markers=True): |
| """The pointer model. `train_markers=False` keeps the five structural-token embedding rows frozen (Kev's |
| default); every checkpoint's recipe records the choice as `train_structural_embeddings`, and the loaders |
| pass it back here so the adapter keys match what was saved.""" |
| import torch |
| if precision not in ['bf16', 'fp32']: |
| raise ValueError('Unsupported precision') |
| model = DecisionModel(path, tokenizer, 'cuda', lora=lora_rank, head_dim=256, revision=revision, |
| dtype=torch.bfloat16 if precision == 'bf16' else torch.float32, |
| attn='sdpa', special_embeddings=train_markers, option_isolation=option_isolation, |
| lora_targets=lora_targets) |
| if not train_markers: |
| model.lm.config.use_cache = False |
| model.lm.gradient_checkpointing_enable(gradient_checkpointing_kwargs={'use_reentrant': False}) |
| return model |
| |
| |
| |
| words = ['state', 'question', 'option', 'end', 'decision'] |
| with torch.no_grad(): |
| embedding = model.lm.get_input_embeddings() |
| rows = [] |
| for word in words: |
| ids = tokenizer.tokenizer(word, add_special_tokens=False).input_ids |
| rows.append(embedding(torch.tensor(ids, device='cuda')).mean(0)) |
| found = [(n, p) for n, p in model.named_parameters() if n.endswith('trainable_tokens_delta.default')] |
| if len(found) != 1 or found[0][1].shape != (5, model.lm.config.hidden_size): |
| raise ValueError('Unexpected PEFT structural-embedding state') |
| found[0][1].copy_(torch.stack(rows).to(found[0][1].dtype)) |
| model.lm.config.use_cache = False |
| model.lm.gradient_checkpointing_enable(gradient_checkpointing_kwargs={'use_reentrant': False}) |
| return model |
|
|
