| | import numpy as np
|
| | from loguru import logger
|
| | from typing import List, Optional, Tuple
|
| |
|
| | from .binary_hdv import BinaryHDV, majority_bundle
|
| | from .config import ContextConfig
|
| |
|
| | class TopicTracker:
|
| | """
|
| | Phase 12.2: Contextual Awareness
|
| | Tracks the rolling conversational context using an HDV moving average.
|
| | Detects topic shifts and resets the context when appropriate.
|
| | """
|
| | def __init__(self, config: ContextConfig, dimension: int):
|
| | self.config = config
|
| | self.dimension = dimension
|
| | self.context_vector: Optional[BinaryHDV] = None
|
| | self.history: List[BinaryHDV] = []
|
| |
|
| | def add_query(self, query_hdv: BinaryHDV) -> Tuple[bool, float]:
|
| | """
|
| | Adds a new query to the tracker.
|
| | Returns (is_shift, similarity).
|
| | If is_shift is True, it means a topic boundary was detected.
|
| | """
|
| | if not self.config.enabled:
|
| | return False, 1.0
|
| |
|
| | if self.context_vector is None:
|
| | self.context_vector = query_hdv
|
| | self.history = [query_hdv]
|
| | return False, 1.0
|
| |
|
| | similarity = self.context_vector.similarity(query_hdv)
|
| |
|
| |
|
| | if similarity < self.config.shift_threshold:
|
| | logger.info(f"Topic shift detected! Similarity {similarity:.3f} < {self.config.shift_threshold}")
|
| | self.reset(query_hdv)
|
| | return True, similarity
|
| |
|
| |
|
| | self.history.append(query_hdv)
|
| | if len(self.history) > self.config.rolling_window_size:
|
| | self.history.pop(0)
|
| |
|
| |
|
| | self.context_vector = majority_bundle(self.history)
|
| | return False, similarity
|
| |
|
| | def reset(self, new_context: Optional[BinaryHDV] = None):
|
| | """Resets the topic tracker, optionally seeding it with a new query."""
|
| | if new_context:
|
| | self.context_vector = new_context
|
| | self.history = [new_context]
|
| | else:
|
| | self.context_vector = None
|
| | self.history = []
|
| |
|
| | def get_context(self) -> Optional[BinaryHDV]:
|
| | """Returns the current topic context vector."""
|
| | return self.context_vector
|
| |
|
