SPEV / decision_layout.py
spragian's picture
Release isolated single-pass SPEV decision head
2e0601e verified
Raw History Blame Contribute Delete
4.43 kB
"""Token layout for an evidence prefix and independent decision candidates."""
from dataclasses import dataclass
from .candidate_tree import CandidateTree
from .decision_utils import MARKER, literal, locate_markers
@dataclass(frozen=True)
class DecisionLayout:
tree: CandidateTree
anchors: tuple[int, ...]
question_positions: tuple[tuple[int, ...], ...]
lexical_ids: tuple[tuple[int, ...], ...]
candidate_questions: tuple[int, ...]
question_types: tuple[int, ...]
def to_dict(self) -> dict:
return {
"prefix_length": self.tree.prefix_length,
"branch_lengths": self.tree.branch_lengths,
"anchors": self.anchors,
"question_positions": self.question_positions,
"lexical_ids": self.lexical_ids,
"candidate_questions": self.candidate_questions,
"question_types": self.question_types,
}
@classmethod
def from_dict(cls, data: dict):
return cls(
CandidateTree(data["prefix_length"], tuple(data["branch_lengths"])),
tuple(data["anchors"]), tuple(tuple(row) for row in data["question_positions"]),
tuple(tuple(row) for row in data["lexical_ids"]),
tuple(data["candidate_questions"]), tuple(data["question_types"]),
)
def validate(self, length: int):
count = len(self.tree.branch_lengths)
if self.tree.length != length or not (len(self.anchors) == len(self.lexical_ids) == len(self.candidate_questions) == count):
raise ValueError("Decision layout does not match the packed sequence")
if not self.question_types or len(self.question_positions) != len(self.question_types):
raise ValueError("Decision question metadata is incomplete")
if any(kind not in (0, 1, 2) for kind in self.question_types):
raise ValueError("Invalid decision question type")
if set(self.candidate_questions) != set(range(len(self.question_types))):
raise ValueError("Every question requires candidate branches")
start = self.tree.prefix_length
for index, size in enumerate(self.tree.branch_lengths):
if not start <= self.anchors[index] < start + size or not self.lexical_ids[index]:
raise ValueError("Invalid decision candidate metadata")
start += size
for index, positions in enumerate(self.question_positions):
branch = self.candidate_questions.index(index)
start = self.tree.prefix_length + sum(self.tree.branch_lengths[:branch])
end = start + self.tree.branch_lengths[branch]
if not positions or any(position < start or position >= end for position in positions):
raise ValueError("Question positions must belong to its first candidate branch")
def pack_candidates(tokenizer, prefix: list[int], questions: dict) -> tuple[list[int], DecisionLayout]:
ids = list(prefix)
lengths, anchors, positions, lexical, owners, kinds = [], [], [], [], [], []
marker = tokenizer.encode(MARKER.lstrip("\n"), add_special_tokens=False)
for question_index, question in enumerate(questions.values()):
kinds.append({"predicate": 0, "choice": 1, "score": 2}[question["type"]])
for candidate_index, description in enumerate(question["criteria"].values()):
text = literal({"question": question["instructions"], "criterion": description}) + MARKER
encoded = tokenizer(text, add_special_tokens=False, return_offsets_mapping=True)
start = len(ids)
branch = encoded["input_ids"]
ids.extend(branch)
lengths.append(len(branch))
anchors.append(start + locate_markers(branch, marker, 1)[0])
owners.append(question_index)
lexical.append(tuple(tokenizer.encode(literal(description), add_special_tokens=False)))
if candidate_index == 0:
phrase = literal(question["instructions"])
begin = len('{"question": ')
end = begin + len(phrase)
positions.append(tuple(start + i for i, (a, b) in enumerate(encoded["offset_mapping"]) if a < end and b > begin))
layout = DecisionLayout(CandidateTree(len(prefix), tuple(lengths)), tuple(anchors), tuple(positions), tuple(lexical), tuple(owners), tuple(kinds))
layout.validate(len(ids))
return ids, layout