diff --git a/delphi/polismath/conversation/conversation.py b/delphi/polismath/conversation/conversation.py index a837a446f..fa497e180 100644 --- a/delphi/polismath/conversation/conversation.py +++ b/delphi/polismath/conversation/conversation.py @@ -26,6 +26,7 @@ ) from polismath.pca_kmeans_rep.repness import conv_repness from polismath.pca_kmeans_rep.corr import compute_correlation +from polismath.pca_kmeans_rep.group_k_smoother import group_k_smoother_update from polismath.utils.engine_mode import resolve_engine_mode, ENGINE_MODE_LEGACY @@ -847,8 +848,31 @@ def _compute_clusters(self, logger.info(f"Selected k={best_k} with silhouette={best_score:.4f}") - # Use the best clustering - group_labels, group_centers, group_member_lists, _ = group_clusterings[best_k] + # Engine-mode K selection (PR-D). 'improved' (default) keeps best_k + # exactly as computed above — bit-for-bit unchanged, including its + # strict-'>' tie-break (LOWER k wins ties). 'clojure-legacy' instead + # runs the group-k-smoother (conversation.clj:454-478): it damps K + # flicker (K only switches after :group-k-buffer=4 consecutive ticks + # agree) and uses Clojure's max-key HIGHER-k-wins tie-break, threading + # {last_k, last_k_count, smoothed_k} plus the per-k clusterings across + # ticks on the conv. These threaded fields are NOT persisted (matching + # conv_man.clj:52-74) — they live in-memory across update_votes only. + if resolve_engine_mode() == ENGINE_MODE_LEGACY: + silhouettes_by_k = {k: group_clusterings[k][3] for k in group_clusterings} + new_smoother_state, selected_k = group_k_smoother_update( + prev_group_k_smoother or {}, silhouettes_by_k) + self.group_clusterings = group_clusterings + self.group_k_smoother = new_smoother_state + logger.info(f"Legacy group-k-smoother: best_k={best_k} " + f"smoothed_k={selected_k} state={new_smoother_state}") + else: + selected_k = best_k + + # Use the selected clustering (best_k in improved mode, smoothed_k in + # legacy mode). The smoother's clamp guarantees selected_k is a key of + # group_clusterings, so this never KeyErrors and group_clusters is never + # None. + group_labels, group_centers, group_member_lists, _ = group_clusterings[selected_k] # Convert to dictionary format with base cluster IDs as members group_clusters = [] diff --git a/delphi/polismath/pca_kmeans_rep/group_k_smoother.py b/delphi/polismath/pca_kmeans_rep/group_k_smoother.py new file mode 100644 index 000000000..d92a37887 --- /dev/null +++ b/delphi/polismath/pca_kmeans_rep/group_k_smoother.py @@ -0,0 +1,129 @@ +""" +Group-K smoother: port of Clojure :group-k-smoother (conversation.clj:454-478). + +Damps flicker in K (the number of opinion groups). Left un-smoothed, K would +jump every tick to whatever k currently maximizes the silhouette; the smoother +only lets K switch to a new best value after `:group-k-buffer` (= 4, +conversation.clj:154) consecutive ticks agree on it. + +The state {last_k, last_k_count, smoothed_k} is threaded ON THE CONV across +conv-update ticks (conversation.clj:457) and is NOT persisted (Clojure's +math_main whitelist omits it, conv_man.clj:52-74). This module is a pure +function so it can be unit-tested in isolation and reused by +Conversation._compute_clusters. + +NOTE: Python has no subgroups (conversation.py hardcodes subgroup_clusters = +{}), so ONLY the top-level group-k-smoother is ported here — the parallel +subgroup smoother (conversation.clj:520-560) is intentionally not. +""" + +from typing import Any, Dict, Mapping, Optional, Tuple + +# Clojure :group-k-buffer default (conversation.clj:154): switch K only after +# this many consecutive ticks agree on a new best-K. +GROUP_K_BUFFER = 4 + + +def _argmax_silhouette_higher_k_wins(silhouettes_by_k: Mapping[int, float]) -> Optional[int]: + """ + argmax_k silhouette[k], breaking ties toward the HIGHER k. + + Clojure computes this as + (apply max-key group-clusterings-silhouettes (keys group-clusterings)) + (conversation.clj:461). Clojure's `max-key` returns the LAST argument among + equal-valued maxima, and the group-clusterings map (built by + `plmb/map-from-keys` over `(range 2 (inc max-k))`) iterates its keys in + ASCENDING order for the small array-maps used here. So on a silhouette tie + the HIGHER k is kept. We reproduce that by scanning k ascending and + replacing the incumbent on `>=` (not strict `>`). + + Returns None only if `silhouettes_by_k` is empty (the caller guarantees at + least k=2 is present, conversation.py group loop over range(2, max_k+1) + with max_k >= 2). + """ + best_k: Optional[int] = None + best_score: Optional[float] = None + for k in sorted(silhouettes_by_k): + score = silhouettes_by_k[k] + if best_score is None or score >= best_score: + best_k = k + best_score = score + return best_k + + +def group_k_smoother_update( + prev_state: Optional[Mapping[str, Any]], + silhouettes_by_k: Mapping[int, float], + buffer: int = GROUP_K_BUFFER, +) -> Tuple[Dict[str, Any], int]: + """ + Advance the group-K smoother by one tick. Pure function. + + Port of Clojure :group-k-smoother (conversation.clj:454-478). + + State carried on the conv across ticks (conversation.clj:457): + last_k - the best-K from the previous tick (None on the first tick) + last_k_count - consecutive ticks this_k has equalled last_k + (Clojure `:or {last-k-count 0}`, conversation.clj:457) + smoothed_k - the K actually used last tick (None on the first tick) + + Update rule (conversation.clj:461-478): + this_k = argmax_k silhouette (ties -> higher k; see helper) + same = last_k is not None and this_k == last_k + this_k_count = last_k_count + 1 if same else 1 + smoothed_k = this_k if this_k_count >= buffer + else (prev smoothed_k if not None else this_k) + clamp (#2536, conversation.clj:469-478): if smoothed_k is not among the + current clusterings' k-values, fall back to this_k. + + First tick (smoothed_k is None): accepts this_k immediately — the + cold-start invariant that makes 'improved' and 'clojure-legacy' coincide on + tick 1. + + Args: + prev_state: previous {last_k, last_k_count, smoothed_k}; None/{} on the + first tick. + silhouettes_by_k: {k: silhouette} for THIS tick's clusterings. Its keys + are the valid k-values used by the clamp. + buffer: consecutive-agreement threshold before switching K + (Clojure :group-k-buffer, default 4). + + Returns: + (new_state, smoothed_k). `smoothed_k` is guaranteed to be a key of + `silhouettes_by_k`, so `clusterings[smoothed_k]` never KeyErrors. + + Raises: + ValueError: if `silhouettes_by_k` is empty — the membership guarantee + above would be impossible to honor. + """ + if not silhouettes_by_k: + raise ValueError( + "group_k_smoother_update requires a non-empty silhouettes_by_k: " + "smoothed_k must be one of its keys") + state = prev_state or {} + last_k = state.get('last_k') # None if absent + last_k_count = state.get('last_k_count', 0) # Clojure :or {last-k-count 0} + prev_smoothed_k = state.get('smoothed_k') # None if absent + + this_k = _argmax_silhouette_higher_k_wins(silhouettes_by_k) + + same = last_k is not None and this_k == last_k + this_k_count = last_k_count + 1 if same else 1 + + if this_k_count >= buffer: + smoothed_k = this_k + else: + smoothed_k = prev_smoothed_k if prev_smoothed_k is not None else this_k + + # Clamp (#2536, conversation.clj:469-478): a carried smoothed_k that no + # longer exists this tick (e.g. the base-cluster count shrank so max-k + # dropped) falls back to the current best available k. + if smoothed_k not in silhouettes_by_k: + smoothed_k = this_k + + new_state = { + 'last_k': this_k, + 'last_k_count': this_k_count, + 'smoothed_k': smoothed_k, + } + return new_state, smoothed_k diff --git a/delphi/tests/test_group_k_smoother.py b/delphi/tests/test_group_k_smoother.py new file mode 100644 index 000000000..992288587 --- /dev/null +++ b/delphi/tests/test_group_k_smoother.py @@ -0,0 +1,206 @@ +#!/usr/bin/env python3 +""" +Tests for the group-K smoother (PR-D smoother part). + +Port of Clojure :group-k-smoother (conversation.clj:454-478), which damps the +number-of-opinion-groups (K) so it only switches to a new best value after +`:group-k-buffer` (4, conversation.clj:154) consecutive ticks agree on it. + +Two layers: + 1. Pure-function unit tests (buffer counting, reset-on-change, clamp, + first-tick, higher-k tie-break) — fast, deterministic. + 2. A chained-update_votes integration test proving the smoother is threaded + across ticks in 'clojure-legacy' mode (no flicker on brief alternation, + switch after 4 consecutive) and is inert in 'improved' mode. +""" + +import os +import sys + +import pytest + +sys.path.append(os.path.abspath(os.path.dirname(__file__))) + +from polismath.pca_kmeans_rep.group_k_smoother import ( + group_k_smoother_update, + GROUP_K_BUFFER, +) +from polismath.utils.engine_mode import ENGINE_MODE_ENV_VAR +from polismath.pca_kmeans_rep.pca import PCA_IMPL_ENV_VAR +import polismath.conversation.conversation as conv_mod +from polismath.conversation.conversation import Conversation + + +# --------------------------------------------------------------------------- +# Pure-function unit tests +# --------------------------------------------------------------------------- + +def _sils(chosen, ks=(2, 3)): + """Silhouettes that make `chosen` the unique argmax over ks.""" + return {k: (1.0 if k == chosen else 0.0) for k in ks} + + +def _drive(this_k_seq, ks=(2, 3), buffer=GROUP_K_BUFFER): + """Chain the smoother over a sequence of desired this_k values, returning + the list of smoothed_k emitted at each tick.""" + state = {} + out = [] + for chosen in this_k_seq: + state, sm = group_k_smoother_update(state, _sils(chosen, ks), buffer=buffer) + out.append(sm) + return out + + +class TestGroupKSmootherPure: + + def test_default_buffer_is_four(self): + assert GROUP_K_BUFFER == 4 # Clojure :group-k-buffer, conversation.clj:154 + + def test_first_tick_accepts_best_k(self): + state, sm = group_k_smoother_update({}, {2: 0.1, 3: 0.9}) + assert sm == 3 + assert state == {'last_k': 3, 'last_k_count': 1, 'smoothed_k': 3} + + def test_first_tick_accepts_best_k_none_state(self): + _, sm = group_k_smoother_update(None, {2: 0.9, 3: 0.1}) + assert sm == 2 + + def test_switch_only_after_four_consecutive(self): + # smoothed_k established at 2, then this_k flips to 3 and must wait 4 + # consecutive ticks before smoothed_k follows. + out = _drive([2, 2, 2, 2, 3, 3, 3, 3]) + assert out == [2, 2, 2, 2, 2, 2, 2, 3] + + def test_reset_on_change(self): + # A single interrupting this_k=2 (tick index 6) resets the 3-streak, so + # the switch is delayed until 4 fresh consecutive 3's accumulate. + out = _drive([2, 2, 2, 2, 3, 3, 2, 3, 3, 3, 3]) + assert out == [2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 3] + # Interrupt really did reset the counter (would have switched at index 7 + # without the reset). + assert out[7] == 2 + + def test_brief_alternation_does_not_flicker(self): + # this_k alternates but never reaches 4-in-a-row -> smoothed_k pinned. + out = _drive([2, 3, 2, 3, 2, 3]) + assert out == [2, 2, 2, 2, 2, 2] + + def test_clamp_missing_carried_smoothed_falls_back_to_this_k(self): + # Carried smoothed_k=5 no longer exists among {2,3} -> fall back to + # this_k, never KeyError (Clojure clamp #2536, conversation.clj:469-478). + prev = {'last_k': 5, 'last_k_count': 10, 'smoothed_k': 5} + state, sm = group_k_smoother_update(prev, {2: 0.9, 3: 0.1}) + assert sm == 2 # this_k + assert state['smoothed_k'] == 2 + + def test_clamp_present_carried_smoothed_is_kept(self): + prev = {'last_k': 3, 'last_k_count': 1, 'smoothed_k': 3} + # this_k=2 but not yet 4-in-a-row, so smoothed_k should stay carried 3 + # (which IS present) rather than flip. + _, sm = group_k_smoother_update(prev, {2: 0.9, 3: 0.1}) + assert sm == 3 + + def test_empty_silhouettes_raises(self): + """The contract guarantees smoothed_k is a key of silhouettes_by_k — + impossible for an empty dict, so fail fast instead of returning None.""" + with pytest.raises(ValueError, match="non-empty"): + group_k_smoother_update({}, {}) + + def test_tie_break_higher_k_wins(self): + # Clojure max-key returns the LAST maximal arg; keys iterate ascending + # -> higher k wins ties (conversation.clj:461). + _, sm = group_k_smoother_update({}, {2: 0.5, 3: 0.5}) + assert sm == 3 + + def test_tie_break_higher_k_wins_among_partial_ties(self): + _, sm = group_k_smoother_update({}, {2: 0.5, 3: 0.5, 4: 0.2}) + assert sm == 3 + _, sm = group_k_smoother_update({}, {2: 0.2, 3: 0.5, 4: 0.5}) + assert sm == 4 + + +# --------------------------------------------------------------------------- +# Pipeline integration: smoother threaded across chained update_votes +# --------------------------------------------------------------------------- + +class _SilStub: + """Deterministic silhouette stub keyed on CALL INDEX, not label counts, so + it is robust to empty clusters. The group loop calls silhouette once per k + in ascending order (k=2 then k=3) every tick, so call 2*t is the k=2 call + of tick t and call 2*t+1 is the k=3 call. Returns 1.0 for the tick's + preferred k and 0.0 otherwise.""" + + def __init__(self, prefs): + self.prefs = prefs + self.n = 0 + + def __call__(self, X, labels): + idx = self.n + self.n += 1 + tick = idx // 2 + this_call_k = 2 if (idx % 2 == 0) else 3 + pref = self.prefs[min(tick, len(self.prefs) - 1)] + return 1.0 if this_call_k == pref else 0.0 + + +def _many_ptpt_votes(n_ptpts=18, n_cmnts=8): + """18 DISTINCT ternary vote rows over 8 comments (3 group signatures + 5 + unique bits), so base k-means yields 18 singleton base clusters and + max-k = min(5, 2 + 18//12) = 3 -> group clusterings for k in {2, 3}.""" + votes = [] + for i in range(n_ptpts): + g = i % 3 + for j in range(n_cmnts): + if j < 3: + v = 1.0 if j == g else -1.0 + else: + v = 1.0 if ((i >> (j - 3)) & 1) else -1.0 + votes.append({'pid': f'p{i}', 'tid': f'c{j}', 'vote': v}) + return {'votes': votes} + + +# A repeat of p0's c0 vote (g=0 -> j==0 -> +1): unchanged matrix, but still +# triggers a recompute tick. +_REPEAT_VOTE = {'votes': [{'pid': 'p0', 'tid': 'c0', 'vote': 1.0}]} + + +class TestSmootherPipeline: + + def _setup(self, monkeypatch, mode, prefs): + monkeypatch.delenv(PCA_IMPL_ENV_VAR, raising=False) # default powerit + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, mode) + stub = _SilStub(prefs) + monkeypatch.setattr(conv_mod, 'calculate_silhouette_sklearn', stub) + return stub + + def test_legacy_no_flicker_then_switch_after_four(self, monkeypatch): + # this_k schedule: 2,2,3,2,3,3,3,3,3 (the isolated 3 at tick 2 is brief + # noise; 3 becomes stable from tick 4 -> switch on the 4th consecutive). + prefs = [2, 2, 3, 2, 3, 3, 3, 3, 3] + self._setup(monkeypatch, 'clojure-legacy', prefs) + + conv = Conversation('smooth').update_votes(_many_ptpt_votes()) + # Sanity: the two-tick group clusterings really exist for k in {2,3}. + assert set(conv.group_clusterings.keys()) == {2, 3} + + smoothed = [conv.group_k_smoother['smoothed_k']] + for _ in range(1, len(prefs)): + conv = conv.update_votes(_REPEAT_VOTE) + smoothed.append(conv.group_k_smoother['smoothed_k']) + + assert smoothed == [2, 2, 2, 2, 2, 2, 2, 3, 3], smoothed + # Brief alternation (tick 2) did not flip; switch happened only on the + # 4th consecutive this_k=3 (tick 7), not the 3rd (tick 6). + assert smoothed[6] == 2 and smoothed[7] == 3 + # group_clusters is picked from the smoothed k and is never None/empty. + assert conv.group_clusters, "group_clusters must be populated" + + def test_improved_mode_leaves_smoother_inert(self, monkeypatch): + prefs = [3, 3, 3, 3] + self._setup(monkeypatch, 'improved', prefs) + conv = Conversation('smooth').update_votes(_many_ptpt_votes()) + # Improved mode never touches the smoother/clusterings state. + assert conv.group_k_smoother == {} + assert conv.group_clusterings == {} + # But still produces group clusters via the untouched best_k path. + assert conv.group_clusters, "group_clusters must be populated"