Spaces:
Paused
Paused
File size: 3,151 Bytes
68b18cf 05f68f1 68b18cf 05f68f1 68b18cf | 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 | """
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
|