Spaces:
Paused
Paused
| """ | |
| Novelty Detector — determines if the sensory state has changed enough | |
| to warrant LLM attention. | |
| If nothing changed, don't bother the LLM. | |
| """ | |
| import math | |
| from .stream_state import SessionState | |
| def compute_novelty(state: SessionState) -> float: | |
| """Compute novelty score from recent frames and audio. | |
| Returns 0.0-1.0 indicating how much the state has changed. | |
| """ | |
| frames = list(state.frames) | |
| if len(frames) < 2: | |
| return 1.0 if frames else 0.0 | |
| # Motion novelty: average frame delta over recent frames | |
| recent = frames[-min(8, len(frames)):] | |
| motion_avg = sum(f.motion_score for f in recent) / len(recent) | |
| # Entropy novelty: variance in entropy indicates scene change | |
| entropies = [f.entropy for f in recent] | |
| if len(entropies) > 1: | |
| entropy_mean = sum(entropies) / len(entropies) | |
| entropy_var = sum((e - entropy_mean) ** 2 for e in entropies) / len(entropies) | |
| entropy_novelty = min(1.0, math.sqrt(entropy_var) / 2.0) | |
| else: | |
| entropy_novelty = 0.0 | |
| # Audio novelty: new chunks since last observer run | |
| new_audio = sum(1 for c in state.audio_chunks if c.ts > state.last_observer_ts) | |
| audio_novelty = min(1.0, new_audio / 5.0) | |
| # Combined | |
| novelty = 0.4 * motion_avg + 0.3 * entropy_novelty + 0.3 * audio_novelty | |
| return min(1.0, novelty) | |
| def should_observe(state: SessionState, novelty: float, min_interval: float = 1.5) -> bool: | |
| """Decide if observer LLM should run. | |
| Called on every frame. Throttled to min_interval seconds. | |
| Runs if there's any novelty OR new audio since last observation. | |
| """ | |
| elapsed = __import__("time").time() - state.last_observer_ts | |
| if elapsed < min_interval: | |
| return False | |
| # Run if there's any motion, scene change, or new audio | |
| new_audio = sum(1 for c in state.audio_chunks if c.ts > state.last_observer_ts) | |
| return novelty > 0.02 or new_audio > 0 | |
| def compute_qvd(state: SessionState, novelty: float) -> float: | |
| """Quality Value Density — should we generate code? | |
| QVD_t = (ΔU + ΔC + ΔE) / (MB + λ·seconds) · Conf | |
| Simplified heuristic version. | |
| """ | |
| # ΔU: new user intent (instruction changed) | |
| delta_u = 1.0 if state.user_instruction and not state.observer_state.get("last_instruction") else 0.0 | |
| # ΔC: new code-relevant context (frames + audio) | |
| delta_c = novelty * 0.5 | |
| # ΔE: new external evidence (placeholder — would be retrieval) | |
| delta_e = 0.0 | |
| # Cost: MB processed + time elapsed | |
| frames = list(state.frames) | |
| mb = sum(f.width * f.height * 3 for f in frames) / (1024 * 1024) if frames else 0.1 | |
| seconds = max(1.0, __import__("time").time() - state.last_builder_ts) | |
| # Confidence: based on speaker attribution + vision clarity | |
| conf = 0.5 | |
| if state.speakers.get("user", {}).get("confidence", 0) > 0.5: | |
| conf += 0.2 | |
| if frames and frames[-1].entropy > 5.0: | |
| conf += 0.15 | |
| conf = min(1.0, conf) | |
| numerator = delta_u + delta_c + delta_e | |
| denominator = mb + 0.1 * seconds | |
| if denominator < 0.01: | |
| denominator = 0.01 | |
| return (numerator / denominator) * conf | |