三段式路由管线 深度解析

72def stage1(
73    event,
74    *,
75    now: datetime,
76    policy: Policy,
77    quiet_hours: str = "23:00-08:00",
78    night_whitelist: list[str] | None = None,
79    recent_dedup_keys: frozenset[str] | set[str] = frozenset(),
80    embedder: Embedder = DEFAULT_EMBEDDER,
81) -> RuleHit | None:
82    """Stage-1 hard rules (SPEC §4.4). Returns a forced route, or None to continue."""
83    night_whitelist = night_whitelist or []
84
85    # Drop rules run before the quiet-hours digest rule: noise must die even at
86    # night, or it would resurface in the morning digest (see §4.7 events 1/24).
87    if policy.is_muted(event.topic):
88        return RuleHit("drop", "muted_topic", f"topic {event.topic} muted in POLICY.md")
89
90    if event.dedup_key and event.dedup_key in recent_dedup_keys:
91        return RuleHit("drop", "dedup", "duplicate of an event seen in the last 24h")
92
93    if is_zero_information(event.summary, embedder):
94        return RuleHit("drop", "zero_information", "zero-information template (empty report)")
95
96    if in_quiet_hours(now, quiet_hours) and not _topic_in_whitelist(
97        event.topic, night_whitelist
98    ):
99        return RuleHit("digest", "quiet_hours", f"quiet hours {quiet_hours}, topic not whitelisted")
100
101    rule = policy.route_for(event.topic)
102    if rule:
103        return RuleHit(
104            rule.route, f"policy:{rule.pattern}", f"POLICY.md rule {rule.pattern} -> {rule.route}"
105        )
106
107    return None

123class SimilarityClassifier:
124    """Compares an event against engaged/dismissed historical vectors (ms).
125
126    dismissed-sim > 0.88 with no engaged record → drop;
127    engaged-sim > 0.88 → skip judge, route by historical same-class mean;
128    otherwise → pass through to stage 3.
129    """
130
131    def __init__(self, embedder: Embedder = DEFAULT_EMBEDDER, threshold: float = STAGE2_THRESHOLD):
132        self.embedder = embedder
133        self.threshold = threshold
134        self._engaged: list[tuple[list[float], str]] = []  # (vector, route)
135        self._dismissed: list[list[float]] = []
136
137    def add_engaged(self, text: str, route: str) -> None:
138        self._engaged.append((self.embedder.embed(text), route))
139
140    def add_dismissed(self, text: str) -> None:
141        self._dismissed.append(self.embedder.embed(text))
142
143    def classify(self, text: str) -> Stage2Verdict:
144        vec = self.embedder.embed(text)
145        engaged_hits = [
146            (cosine(vec, evec), route)
147            for evec, route in self._engaged
148            if cosine(vec, evec) > self.threshold
149        ]
150        if engaged_hits:
151            # "route by historical same-class mean": majority route among the
152            # similar engaged records, nearest-first on ties.
153            engaged_hits.sort(reverse=True)
154            routes = [route for _, route in engaged_hits]
155            best = max(set(routes), key=lambda r: (routes.count(r), -routes.index(r)))
156            sim = engaged_hits[0][0]
157            return Stage2Verdict(
158                "route", best, sim, f"engaged-similar {sim:.2f}, historical route {best}"
159            )
160        dismissed_sim = max((cosine(vec, dvec) for dvec in self._dismissed), default=0.0)
161        if dismissed_sim > self.threshold:
162            return Stage2Verdict(
163                "drop", None, dismissed_sim,
164                f"dismissed-similar {dismissed_sim:.2f} with no engaged record",
165            )
166        return Stage2Verdict("pass", None, dismissed_sim, "unfamiliar; ask the judge")

207def score_and_route(
208    result,
209    scene: SceneState,
210    *,
211    topic_weights: dict[str, float] | None = None,
212    scene_cost: float = 0.0,
213    threshold_overrides: dict[str, float] | None = None,
214    threshold_adjust: float = 0.0,
215    memory_hit: bool = False,
216) -> tuple[str, float, dict[str, float], str]:
217    """Compose `score = Σ(w_topic[dim]·comp[dim]) − scene_cost` and route.
218
219    Returns (route, score, components-after-boost, reason).
220    """
221    from context.infer import downgrade_low_confidence, interrupt_threshold
222
223    weights = {**DEFAULT_WEIGHTS, **(topic_weights or {})}
224    comps = {dim: getattr(result, dim) for dim in DIMS}
225    if memory_hit:
226        comps["relevance"] = min(1.0, comps["relevance"] * 1.2)
227
228    score = sum(weights[dim] * comps[dim] for dim in DIMS) - scene_cost
229    threshold = interrupt_threshold(scene.scene, threshold_overrides)
230    if threshold_adjust:
231        # learned global adjustment, clamped (SPEC §4.6 bounds [0.35, 0.95])
232        threshold = min(0.95, max(0.35, threshold + threshold_adjust))
233
234    if score >= threshold:
235        route = "interrupt"
236        reason = f"score {score:.2f} ≥ {scene.scene} threshold {threshold:.2f}"
237    elif score >= DIGEST_FLOOR:
238        route = "digest"
239        reason = f"score {score:.2f} below {scene.scene} threshold {threshold:.2f}"
240    elif result.memorize:
241        route = "curate"
242        reason = f"score {score:.2f} low but worth remembering"
243    else:
244        route = "drop"
245        reason = f"score {score:.2f} with no lasting value"
246
247    route = downgrade_low_confidence(route, scene)
248    if route == "digest" and score >= threshold:
249        reason += f"; downgraded (scene confidence {scene.confidence:.2f} < 0.6)"
250
251    # SPEC §4.4: dispatchable prep work runs first, then delivery ("arrive with a plan")
252    if result.dispatchable and route in ("interrupt", "digest"):
253        route = "dispatch"
254        reason += "; dispatchable prep work available"
255
256    return route, score, comps, reason