File size: 27,704 Bytes
3e363e2
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
"""Typed path readout: answer from node types and structure, never relation names.

The unseen-ontology holdout renames every relation to an opaque label, so a readout
that matches the question to relation names has nothing to match.  Node types keep
their names, and an oracle that follows only node types reaches 0.99 exact match on
that split, so the information is in the input.  This readout enumerates the paths
from the start entities and scores each path by its node types alone.

Two complementary scorers, blended with a weight chosen on internal holdouts:

* :class:`TypeSequenceScorer` reads the whole question and predicts the hop count and
  the type at each hop.  Robust to new wording of known question types, but it learns
  whole type sequences, so it fails on new *combinations* (0.01 on held-out intents).
* :class:`TypeBagScorer` reads the question token by token: each token names a type
  or nothing, and a path is scored by how likely its multiset of types is under those
  token probabilities, plus the answer type.  Order is left to the graph.  Because a
  word such as "supplier" names the same type in every intent, this composes (0.65 on
  held-out intents), at some cost on new wording.

Types are scored by name embedding, so type names never seen in training work as
long as the name carries meaning.  A small classifier over the readout's own scores
decides whether to answer or abstain, and confidence is isotonic-calibrated per
predicted hop count.
"""

from __future__ import annotations

import math
import re
from collections import Counter
from collections.abc import Callable, Sequence
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Any

import torch
from torch import Tensor, nn
from torch.nn import functional

from graph.graph import GraphStore
from graph.training.calibration import IsotonicCalibrator
from graph.types import (
    EntityAnswer,
    GraphPath,
    GraphQuery,
    ReasoningResult,
    ReasoningStatus,
)

MAX_HOPS = 4
#: Largest per-type count the bag scorer models along one path.
MAX_COUNT = 4
DEFAULT_ENCODER = "sentence-transformers/all-mpnet-base-v2"
#: Features read by the abstain head: v1 has 5, v2 adds novelty and agreement, v3 adds
#: the gate's P(sequence scorer is right).
FEATURES_V1 = 5
FEATURES_V2 = 7
FEATURES_V3 = 8
NUM_FEATURES = FEATURES_V1
#: Index of the template-novelty feature.  It predicts errors on held-out *intents* but
#: not on new wording of known intents, so the abstain head must not read it: with it,
#: the v3 head abstained on 71% of answerable unseen-ontology questions.
NOVELTY_FEATURE = 5
#: Inputs to the gate; see :func:`gate_features`.  Gates trained with the first 5 only
#: still load: :meth:`TypedReadout.gate_probability` slices to the gate's input size.
GATE_FEATURES = 7
#: Hard cap on enumerated paths per query, so a dense graph cannot stall inference.
MAX_PATHS = 200_000
_ENTITY = re.compile(r"\be\d+_\d+\b")

Encoder = Callable[[Sequence[str]], Tensor]
TokenEncoder = Callable[[Sequence[str]], list[Tensor]]


class TypeSequenceScorer(nn.Module):
    """Question embedding -> P(hop count) and P(node type at hop i | hop count)."""

    def __init__(self, dim: int, max_hops: int = MAX_HOPS) -> None:
        super().__init__()
        self.max_hops = max_hops
        self.hop = nn.Sequential(nn.Linear(dim, 256), nn.GELU(), nn.Linear(256, max_hops))
        self.slot = nn.Parameter(torch.randn(max_hops, max_hops, 64) * 0.1)
        self.mix = nn.Sequential(nn.Linear(dim + 64, 512), nn.GELU(), nn.Linear(512, dim))
        self.scale = nn.Parameter(torch.tensor(10.0))

    def hop_logp(self, question: Tensor) -> Tensor:
        return functional.log_softmax(self.hop(question), -1)

    def type_logits(
        self, question: Tensor, hops: int, position: int, type_embeddings: Tensor
    ) -> Tensor:
        """Logits over ``type_embeddings`` rows for the node at ``position``."""
        slot = self.slot[hops - 1, position].expand(question.shape[0], -1)
        query = functional.normalize(self.mix(torch.cat([question, slot], -1)), dim=-1)
        return self.scale * (query @ type_embeddings.T)


class TypeBagScorer(nn.Module):
    """Token-level type mentions -> likelihood of a path's type multiset."""

    def __init__(self, dim: int, max_count: int = MAX_COUNT) -> None:
        super().__init__()
        self.max_count = max_count
        self.tok = nn.Sequential(nn.Linear(dim, dim), nn.GELU(), nn.Linear(dim, dim))
        self.none = nn.Parameter(torch.tensor(2.0))
        self.scale = nn.Parameter(torch.tensor(10.0))
        self.ans = nn.Sequential(nn.Linear(dim, 512), nn.GELU(), nn.Linear(512, dim))
        self.ascale = nn.Parameter(torch.tensor(10.0))

    def token_probs(self, tokens: Tensor, type_embeddings: Tensor) -> Tensor:
        """Per token, P(token names type k); the remainder is P(names no type)."""
        z = functional.normalize(self.tok(tokens), dim=-1)
        logits = self.scale * (z @ type_embeddings.T)
        logits = torch.cat([logits, self.none.expand(logits.shape[0], 1)], -1)
        return torch.softmax(logits, -1)[:, :-1]

    def count_logp(self, tokens: Tensor, type_embeddings: Tensor) -> Tensor:
        """[types, max_count + 1]: log P(type k is named c times), Poisson-binomial."""
        p = self.token_probs(tokens, type_embeddings).clamp(1e-6, 1 - 1e-6)
        k = p.shape[1]
        dist = torch.zeros(k, self.max_count + 1, device=p.device)
        dist[:, 0] = 1.0
        for row in p:
            prob = row[:, None]
            shifted = torch.cat([torch.zeros(k, 1, device=p.device), dist[:, :-1]], 1)
            dist = dist * (1 - prob) + shifted * prob
        return torch.log(dist.clamp_min(1e-12))

    def answer_logp(self, question: Tensor, type_embeddings: Tensor) -> Tensor:
        z = functional.normalize(self.ans(question), dim=-1)
        return functional.log_softmax(self.ascale * (z @ type_embeddings.T), -1)


class GateHead(nn.Module):
    """P(the sequence scorer's answer is right rather than the bag scorer's)."""

    def __init__(self, features: int = GATE_FEATURES) -> None:
        super().__init__()
        self.net = nn.Sequential(nn.Linear(features, 16), nn.GELU(), nn.Linear(16, 1))

    def forward(self, features: Tensor) -> Tensor:
        return self.net(features).squeeze(-1)


class AbstainHead(nn.Module):
    """P(answering is correct) from the readout's own scores."""

    def __init__(self, features: int = NUM_FEATURES) -> None:
        super().__init__()
        self.net = nn.Sequential(nn.Linear(features, 32), nn.GELU(), nn.Linear(32, 1))

    def forward(self, features: Tensor) -> Tensor:
        return self.net(features).squeeze(-1)


@dataclass(frozen=True)
class ReadoutScore:
    """Everything the readout computes for one query before deciding."""

    answers: frozenset[str]
    features: tuple[float, ...]
    best_paths: dict[str, GraphPath]
    endpoint_scores: dict[str, float]
    #: argmax of the hop-count head; known at inference, used to pick a calibrator.
    predicted_hops: int = 0


class TypedReadout:
    """Neuro-symbolic readout over a :class:`GraphStore`; see the module docstring."""

    def __init__(
        self,
        scorer: TypeSequenceScorer,
        abstain: AbstainHead,
        encode: Encoder,
        *,
        feature_mean: Tensor,
        feature_std: Tensor,
        threshold: float,
        calibration: tuple[tuple[float, ...], tuple[float, ...]] | None = None,
        hop_calibration: dict[int, tuple[tuple[float, ...], tuple[float, ...]]] | None = None,
        bag: TypeBagScorer | None = None,
        encode_tokens: TokenEncoder | None = None,
        blend_weight: float = 1.0,
        template_bank: Tensor | None = None,
        gate: GateHead | None = None,
        gate_mean: Tensor | None = None,
        gate_std: Tensor | None = None,
        feature_mask: Sequence[int] | None = None,
        decision_source: str = "chosen",
        model_version: str = "typed-readout-v1",
        device: str | torch.device = "cpu",
    ) -> None:
        if bag is not None and encode_tokens is None:
            raise ValueError("a bag scorer needs a token encoder")
        self.device = torch.device(device)
        self.scorer = scorer.to(self.device).eval()
        self.bag = bag.to(self.device).eval() if bag is not None else None
        self.abstain = abstain.to(self.device).eval()
        self.encode = encode
        self.encode_tokens = encode_tokens
        self.blend_weight = float(blend_weight)
        self.template_bank = (
            functional.normalize(template_bank.to(self.device).float(), dim=-1)
            if template_bank is not None
            else None
        )
        self.gate = gate.to(self.device).eval() if gate is not None else None
        self.gate_mean = (gate_mean if gate_mean is not None else torch.zeros(GATE_FEATURES)).to(self.device)
        self.gate_std = (
            (gate_std if gate_std is not None else torch.ones(GATE_FEATURES)).to(self.device).clamp_min(1e-6)
        )
        #: Indices of score features the abstain head reads (None: all of them).
        self.feature_mask = tuple(feature_mask) if feature_mask is not None else None
        #: Which scores the answer/abstain decision reads: "chosen" (the scorer that
        #: produced the answer) or "sequence" (always the sequence scorer, whose scores
        #: the decision was calibrated on and which transfer under wording shift).
        if decision_source not in ("chosen", "sequence"):
            raise ValueError(f"unknown decision_source {decision_source!r}")
        self.decision_source = decision_source
        self.feature_mean = feature_mean.to(self.device)
        self.feature_std = feature_std.to(self.device).clamp_min(1e-6)
        self.threshold = float(threshold)
        self.calibration = calibration
        self.hop_calibration = dict(hop_calibration or {})
        self.model_version = model_version
        self._type_cache: dict[str, Tensor] = {}

    @property
    def feature_count(self) -> int:
        if self.gate is not None:
            return FEATURES_V3
        return FEATURES_V2 if self.template_bank is not None else FEATURES_V1

    # -- construction -------------------------------------------------------

    @classmethod
    def from_checkpoint(
        cls,
        path: str | Path,
        *,
        device: str | torch.device = "cpu",
        encode: Encoder | None = None,
        encode_tokens: TokenEncoder | None = None,
    ) -> TypedReadout:
        bundle = torch.load(path, map_location="cpu", weights_only=True)
        encoder_name = bundle.get("encoder", DEFAULT_ENCODER)
        if encode is None:
            encode, default_tokens = sentence_encoders(encoder_name, device)
            encode_tokens = encode_tokens or default_tokens
        scorer = TypeSequenceScorer(bundle["dim"], bundle.get("max_hops", MAX_HOPS))
        scorer.load_state_dict(bundle["scorer"])
        bag = None
        if bundle.get("bag") is not None:
            bag = TypeBagScorer(bundle["dim"], bundle.get("max_count", MAX_COUNT))
            bag.load_state_dict(bundle["bag"])
        gate = None
        if bundle.get("gate") is not None:
            gate = GateHead(int(bundle["gate"]["net.0.weight"].shape[1]))
            gate.load_state_dict(bundle["gate"])
        # The head's input size is read from its weights, not ``num_features``: an
        # internal readout saved before its gate existed recorded the wrong count.
        abstain = AbstainHead(int(bundle["abstain"]["net.0.weight"].shape[1]))
        abstain.load_state_dict(bundle["abstain"])
        calibration = bundle.get("calibration")
        hop_calibration = {
            int(hops): (tuple(pair[0]), tuple(pair[1]))
            for hops, pair in (bundle.get("hop_calibration") or {}).items()
        }
        return cls(
            scorer,
            abstain,
            encode,
            feature_mean=bundle["feature_mean"],
            feature_std=bundle["feature_std"],
            threshold=bundle["threshold"],
            calibration=tuple(map(tuple, calibration)) if calibration else None,
            hop_calibration=hop_calibration,
            bag=bag,
            encode_tokens=encode_tokens,
            blend_weight=bundle.get("blend_weight", 1.0),
            template_bank=bundle.get("template_bank"),
            gate=gate,
            gate_mean=bundle.get("gate_mean"),
            gate_std=bundle.get("gate_std"),
            feature_mask=bundle.get("feature_mask"),
            decision_source=bundle.get("decision_source", "chosen"),
            model_version=bundle.get("model_version", "typed-readout-v1"),
            device=device,
        )

    def state(self, *, dim: int, encoder: str) -> dict[str, Any]:
        """Serialisable bundle for :meth:`from_checkpoint`."""
        return {
            "scorer": self.scorer.state_dict(),
            "bag": self.bag.state_dict() if self.bag is not None else None,
            "abstain": self.abstain.state_dict(),
            "feature_mean": self.feature_mean.cpu(),
            "feature_std": self.feature_std.cpu(),
            "threshold": self.threshold,
            "calibration": self.calibration,
            "hop_calibration": self.hop_calibration,
            "blend_weight": self.blend_weight,
            "template_bank": self.template_bank.half().cpu()
            if self.template_bank is not None
            else None,
            "gate": self.gate.state_dict() if self.gate is not None else None,
            "gate_mean": self.gate_mean.cpu() if self.gate is not None else None,
            "gate_std": self.gate_std.cpu() if self.gate is not None else None,
            "feature_mask": list(self.feature_mask) if self.feature_mask is not None else None,
            "decision_source": self.decision_source,
            "dim": dim,
            "max_hops": self.scorer.max_hops,
            "max_count": self.bag.max_count if self.bag is not None else MAX_COUNT,
            "num_features": int(self.abstain.net[0].in_features),
            "encoder": encoder,
            "model_version": self.model_version,
        }

    # -- scoring --------------------------------------------------------------

    def _type_embeddings(self, names: Sequence[str]) -> Tensor:
        missing = [name for name in names if name not in self._type_cache]
        if missing:
            vectors = self.encode([name.replace("_", " ") for name in missing])
            for name, vector in zip(missing, vectors, strict=True):
                self._type_cache[name] = vector.to(self.device).float()
        return torch.stack([self._type_cache[name] for name in names])

    def encode_query(self, query: GraphQuery) -> tuple[Tensor, Tensor | None, Tensor | None]:
        """Sentence vector, token vectors (bag) and masked-template vector (novelty)."""
        text = question_text(query)
        texts = [text, template_text(text)] if self.template_bank is not None else [text]
        vectors = self.encode(texts)
        tokens = self.encode_tokens([text])[0] if self.bag is not None else None
        masked = vectors[1] if self.template_bank is not None else None
        return vectors[0], tokens, masked

    @torch.no_grad()
    def dual(
        self,
        graph: GraphStore,
        query: GraphQuery,
        question: Tensor | None = None,
        tokens: Tensor | None = None,
        masked: Tensor | None = None,
    ) -> dict[str, Any]:
        """Per-endpoint best scores under each scorer, and the shared signals."""
        if question is None:
            question, tokens, masked = self.encode_query(query)
        question = question.to(self.device).float().reshape(1, -1)
        type_names = sorted({name for name in graph.nodes.values() if name})
        type_index = {name: i for i, name in enumerate(type_names)}
        type_embeddings = self._type_embeddings(type_names)
        hop_logp = self.scorer.hop_logp(question)[0]
        hop_table = hop_logp.tolist()
        use_bag = self.bag is not None and tokens is not None
        if use_bag:
            tokens = functional.normalize(tokens.to(self.device).float(), dim=-1)
            count_logp = self.bag.count_logp(tokens, type_embeddings).cpu()
            answer_logp = self.bag.answer_logp(question, type_embeddings)[0].cpu()
            # Every type contributes log P(count = 0) unless the path contains it, so
            # sum that once and adjust only the types on the path: O(hops) per path.
            zero_total = float(count_logp[:, 0].sum())
            count_table = count_logp.tolist()
        starts = set(query.start_entity_ids) or set(query.start_entities)
        per_hop: dict[int, list[Tensor]] = {}
        tables: dict[str, tuple[dict[str, float], dict[str, GraphPath]]] = {
            "sequence": ({}, {}), "bag": ({}, {}), "blend": ({}, {}),
        }

        def keep(name: str, end: str, value: float, path: GraphPath) -> None:
            best, paths = tables[name]
            if value > best.get(end, -math.inf):
                best[end] = value
                paths[end] = path

        max_hops = min(query.max_hops, self.scorer.max_hops)
        for count, path in enumerate(graph.iter_paths(sorted(starts), max_hops)):
            if count >= MAX_PATHS:
                break
            hops = len(path.steps)
            if not hops or hops < query.min_hops:
                continue
            end = path.steps[-1].target_id
            if end in starts:
                continue
            types = [graph.nodes[step.target_id] for step in path.steps]
            if hops not in per_hop:
                per_hop[hops] = [
                    functional.log_softmax(
                        self.scorer.type_logits(question, hops, i, type_embeddings)[0], -1
                    ).tolist()
                    for i in range(hops)
                ]
            sequence = hop_table[hops - 1] + sum(
                per_hop[hops][i][type_index[name]] for i, name in enumerate(types)
            )
            keep("sequence", end, sequence, path)
            if use_bag:
                bag_value = float(answer_logp[type_index[types[-1]]]) + zero_total
                for name, count in Counter(types).items():
                    row = count_table[type_index[name]]
                    bag_value += row[min(count, self.bag.max_count)] - row[0]
                keep("bag", end, bag_value, path)
                keep("blend", end, self.blend_weight * sequence + bag_value, path)
        novelty = 0.0
        if self.template_bank is not None and masked is not None:
            vector = functional.normalize(masked.to(self.device).float(), dim=-1)
            novelty = float((self.template_bank @ vector).max())
        return {"tables": tables, "hop_logp": hop_logp, "novelty": novelty, "use_bag": use_bag}

    @torch.no_grad()
    def gate_probability(self, dual: dict[str, Any]) -> float | None:
        """P(sequence scorer is right), or None without a gate or bag scorer."""
        if self.gate is None or not dual["use_bag"]:
            return None
        values = gate_features(
            dual["tables"]["sequence"][0], dual["tables"]["bag"][0], dual["hop_logp"], dual["novelty"]
        )
        if values is None:
            return None
        width = int(self.gate.net[0].in_features)
        features = torch.tensor(values[:width], device=self.device).reshape(1, -1)
        mean, std = self.gate_mean[:width], self.gate_std[:width]
        return float(torch.sigmoid(self.gate((features - mean) / std))[0])

    @torch.no_grad()
    def score(
        self,
        graph: GraphStore,
        query: GraphQuery,
        question: Tensor | None = None,
        tokens: Tensor | None = None,
        masked: Tensor | None = None,
    ) -> ReadoutScore:
        """Score every endpoint reachable from the query's start entities."""
        dual = self.dual(graph, query, question, tokens, masked)
        tables = dual["tables"]
        gate_p = self.gate_probability(dual)
        if gate_p is not None:
            chosen = "sequence" if gate_p >= 0.5 else "bag"
        elif dual["use_bag"]:
            chosen = "blend"
        else:
            chosen = "sequence"
        best, best_paths = tables[chosen]
        extra: tuple[float, ...] = ()
        if self.template_bank is not None:
            agree = (
                float(_top(tables["sequence"][0]) == _top(tables["bag"][0]))
                if dual["use_bag"]
                else 1.0
            )
            extra = (dual["novelty"], agree)
            if self.gate is not None:
                extra = (*extra, gate_p if gate_p is not None else 1.0)
        result = _finalise(best, best_paths, dual["hop_logp"], extra)
        if self.decision_source == "sequence" and chosen != "sequence":
            sequence_view = _finalise(*tables["sequence"], dual["hop_logp"], extra)
            result = replace(result, features=sequence_view.features)
        return result

    @torch.no_grad()
    def probability(self, score: ReadoutScore) -> float:
        """P(answering is correct), from the abstain head."""
        values = score.features
        if self.feature_mask is not None:
            values = tuple(values[i] for i in self.feature_mask)
        features = torch.tensor(values, device=self.device).reshape(1, -1)
        logit = self.abstain((features - self.feature_mean) / self.feature_std)
        return float(torch.sigmoid(logit)[0])

    def calibrate(self, confidence: float, hops: int | None = None) -> float:
        """Isotonic calibration, per predicted hop count when fitted for it."""
        table = self.hop_calibration.get(hops) if hops is not None else None
        table = table or self.calibration
        if not table:
            return confidence
        calibrator = IsotonicCalibrator(*table)
        return float(calibrator.transform(torch.tensor([confidence]))[0])

    # -- decision -------------------------------------------------------------

    def reason(self, graph: GraphStore, query: GraphQuery) -> ReasoningResult:
        score = self.score(graph, query)
        probability = self.probability(score)
        answer = bool(score.answers) and probability >= self.threshold
        # Confidence describes the decision made: for an abstention, the chance
        # that abstaining was right is approximated by 1 - P(answer correct).
        raw = probability if answer else 1.0 - probability
        confidence = self.calibrate(raw, score.predicted_hops)
        answers = (
            tuple(
                EntityAnswer(entity, confidence)
                for entity in sorted(score.answers, key=lambda e: -score.endpoint_scores[e])
            )
            if answer
            else ()
        )
        status = ReasoningStatus.SUPPORTED if answer else ReasoningStatus.INSUFFICIENT_EVIDENCE
        boolean_answer = None
        aggregate_value = None
        if query.answer_mode == "count":
            aggregate_value = len(answers)
        elif query.answer_mode == "boolean":
            targets = set(query.target_entity_ids)
            boolean_answer = bool(targets & {a.entity_id for a in answers}) if targets else answer
        return ReasoningResult(
            status=status,
            answers=answers,
            supporting_paths=tuple(score.best_paths[a.entity_id] for a in answers)
            if query.return_paths
            else (),
            conflicting_paths=(),
            confidence=confidence,
            calibrated=bool(self.calibration or self.hop_calibration),
            model_version=self.model_version,
            graph_version=graph.graph_version,
            reasoning_steps=max((len(p.steps) for p in score.best_paths.values()), default=0),
            metadata={"readout": "typed", "answer_probability": probability},
            boolean_answer=boolean_answer,
            aggregate_value=aggregate_value,
        )

    __call__ = reason


def gate_features(
    sequence: dict[str, float], bag: dict[str, float], hop_logp: Tensor, novelty: float
) -> tuple[float, ...] | None:
    """Gate inputs: how much each scorer dislikes the other's answer, hop certainty,
    template novelty, whether the two agree, and each scorer's best score.

    The best scores separate the two shifts: on new wording of a known intent the
    sequence scorer still finds a path that fits its predicted types (high best score);
    on a new combination its memorised type sequence rarely exists from the start
    entity (low best score), which is when the bag scorer should be trusted."""
    if not sequence or not bag:
        return None
    seq_top, bag_top = _top(sequence), _top(bag)
    seq_max, bag_max = max(sequence.values()), max(bag.values())
    bag_on_seq = max(bag[e] for e in seq_top) - bag_max
    seq_on_bag = max(sequence[e] for e in bag_top) - seq_max
    return (
        max(bag_on_seq, -50.0),
        max(seq_on_bag, -50.0),
        float(hop_logp.max().exp()),
        novelty,
        float(seq_top == bag_top),
        max(seq_max, -100.0),
        max(bag_max, -100.0),
    )


def _top(scores: dict[str, float]) -> frozenset[str]:
    if not scores:
        return frozenset()
    top = max(scores.values())
    return frozenset(e for e, s in scores.items() if s >= top - 1e-6)


def _finalise(
    best: dict[str, float],
    best_paths: dict[str, GraphPath],
    hop_logp: Tensor,
    extra: tuple[float, ...] = (),
) -> ReadoutScore:
    hops = int(hop_logp.argmax()) + 1
    if not best:
        return ReadoutScore(
            answers=frozenset(),
            features=(-30.0, 0.0, 0.0, float(hop_logp.max()), 0.0, *extra),
            best_paths={},
            endpoint_scores={},
            predicted_hops=hops,
        )
    ranked = sorted(best.values(), reverse=True)
    top = ranked[0]
    answers = frozenset(e for e, s in best.items() if s >= top - 1e-6)
    runner_up = next((s for s in ranked if s < top - 1e-6), top - 30.0)
    features = (
        top,
        top - runner_up,
        float(len(answers)),
        float(hop_logp.max()),
        math.log1p(len(best)),
        *extra,
    )
    return ReadoutScore(answers, features, best_paths, dict(best), hops)


def question_text(query: GraphQuery) -> str:
    return query.text or query.canonical_text() or ""


def template_text(text: str) -> str:
    """Question with entity identifiers masked, for template-novelty lookups."""
    return _ENTITY.sub("X", text).lower().strip()


def sentence_encoders(
    name: str, device: str | torch.device = "cpu"
) -> tuple[Encoder, TokenEncoder]:
    """Frozen sentence-transformer: pooled unit vectors and per-token vectors."""
    from sentence_transformers import SentenceTransformer

    model = SentenceTransformer(name, device=str(device))

    def encode(texts: Sequence[str]) -> Tensor:
        with torch.no_grad():
            return (
                model.encode(
                    list(texts),
                    normalize_embeddings=True,
                    convert_to_tensor=True,
                    batch_size=256,
                )
                .float()
                .clone()
            )

    def encode_tokens(texts: Sequence[str]) -> list[Tensor]:
        with torch.no_grad():
            rows = model.encode(list(texts), output_value="token_embeddings", batch_size=256)
        return [functional.normalize(row.float().clone(), dim=-1) for row in rows]

    return encode, encode_tokens


def sentence_encoder(name: str, device: str | torch.device = "cpu") -> Encoder:
    """Pooled encoder only (the v1 interface)."""
    return sentence_encoders(name, device)[0]