Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import math | |
| from collections import Counter | |
| from typing import Dict, List, Optional | |
| from viral_script_engine.memory.creator_history import CreatorHistoryBuffer, EpisodeMemory | |
| _REWARD_KEYS = [ | |
| "r1_hook_strength", | |
| "r2_coherence", | |
| "r3_cultural_alignment", | |
| "r4_debate_resolution", | |
| "r5_defender_preservation", | |
| "r6_safety", | |
| "r7_originality", | |
| "r8_persona_fit", | |
| "r9_platform_pacing", | |
| ] | |
| _DELTA_THRESHOLD = 0.05 | |
| class MemoryCompressor: | |
| """ | |
| Compresses a completed episode into a structured EpisodeMemory. | |
| Called at the end of every episode, before the next reset(). | |
| Zero LLM calls — all compression is rule-based. | |
| """ | |
| def compress(self, episode_log: dict, episode_number: int) -> EpisodeMemory: | |
| """ | |
| episode_log fields expected: | |
| episode_id, niche, platform, first_critique_claims, | |
| actions_taken, initial_reward_components, final_reward_components, | |
| final_total_reward | |
| """ | |
| episode_id = episode_log.get("episode_id", "unknown") | |
| niche = episode_log.get("niche", "unknown") | |
| platform = episode_log.get("platform", "unknown") | |
| actions_taken: List[str] = episode_log.get("actions_taken", []) | |
| initial_rc: dict = episode_log.get("initial_reward_components", {}) | |
| final_rc: dict = episode_log.get("final_reward_components", {}) | |
| final_total = episode_log.get("final_total_reward", 0.0) | |
| # 1. dominant_flaw: most common critique_class from first-step claims | |
| first_claims = episode_log.get("first_critique_claims", []) | |
| if first_claims: | |
| counts = Counter( | |
| c.get("critique_class", "unknown") for c in first_claims | |
| ) | |
| dominant_flaw = counts.most_common(1)[0][0] | |
| else: | |
| dominant_flaw = "hook_weakness" | |
| # 2. what_worked / what_didnt — reward components with significant delta | |
| what_worked: List[str] = [] | |
| what_didnt: List[str] = [] | |
| for key in _REWARD_KEYS: | |
| init_val = initial_rc.get(key) | |
| final_val = final_rc.get(key) | |
| if init_val is None or final_val is None: | |
| continue | |
| delta = final_val - init_val | |
| if delta > _DELTA_THRESHOLD: | |
| what_worked.append(key) | |
| elif delta < -_DELTA_THRESHOLD: | |
| what_didnt.append(key) | |
| # 3. key_learning — rule-based template | |
| most_used_action = ( | |
| Counter(actions_taken).most_common(1)[0][0] if actions_taken else "no_action" | |
| ) | |
| worked_str = what_worked[0] if what_worked else "no component" | |
| didnt_str = what_didnt[0] if what_didnt else "no regressions" | |
| key_learning = ( | |
| f"Fixed {dominant_flaw} using {most_used_action}. " | |
| f"{worked_str} improved, {didnt_str}." | |
| ) | |
| return EpisodeMemory( | |
| episode_id=episode_id, | |
| episode_number=episode_number, | |
| script_niche=niche, | |
| platform=platform, | |
| dominant_flaw=dominant_flaw, | |
| actions_taken=actions_taken, | |
| what_worked=what_worked, | |
| what_didnt=what_didnt, | |
| final_total_reward=final_total, | |
| key_learning=key_learning, | |
| ) | |
| def update_buffer( | |
| self, | |
| existing_buffer: Optional[CreatorHistoryBuffer], | |
| new_memory: EpisodeMemory, | |
| creator_id: str, | |
| ) -> CreatorHistoryBuffer: | |
| """ | |
| Adds new_memory to the buffer, maintaining a sliding window of 5. | |
| Recomputes all aggregate stats. | |
| """ | |
| if existing_buffer is None: | |
| episodes: List[EpisodeMemory] = [] | |
| total = 0 | |
| else: | |
| episodes = list(existing_buffer.recent_episodes) | |
| total = existing_buffer.total_episodes | |
| episodes.append(new_memory) | |
| if len(episodes) > 5: | |
| episodes = episodes[-5:] # keep last 5 | |
| total += 1 | |
| # recurring_weak_points: dominant_flaw in >= 3 of last 5 | |
| flaw_counts = Counter(ep.dominant_flaw for ep in episodes) | |
| recurring_weak_points = [ | |
| flaw for flaw, cnt in flaw_counts.items() if cnt >= 3 | |
| ] | |
| # recurring_strong_points: reward component >= 0.7 in >= 4 of last 5 | |
| recurring_strong_points = self._compute_strong_points(episodes) | |
| # most_effective_action: action_type with highest avg final_total_reward | |
| most_effective_action = self._compute_most_effective_action(episodes) | |
| # voice_stability_score: 1 - std_dev of r3 across episodes (inverted, clamped) | |
| voice_stability_score = self._compute_voice_stability(episodes) | |
| # improvement_trend: slope of final_total_reward | |
| improvement_trend = self._compute_trend(episodes) | |
| return CreatorHistoryBuffer( | |
| creator_id=creator_id, | |
| total_episodes=total, | |
| recent_episodes=episodes, | |
| recurring_weak_points=recurring_weak_points, | |
| recurring_strong_points=recurring_strong_points, | |
| most_effective_action=most_effective_action, | |
| voice_stability_score=voice_stability_score, | |
| improvement_trend=improvement_trend, | |
| ) | |
| # ------------------------------------------------------------------ | |
| # Private helpers | |
| # ------------------------------------------------------------------ | |
| def _compute_strong_points(self, episodes: List[EpisodeMemory]) -> List[str]: | |
| """Reward components consistently >= 0.7 in >= 4 of last 5 episodes.""" | |
| if not episodes: | |
| return [] | |
| # We only know what_worked from EpisodeMemory — approximate by checking | |
| # which components appear in what_worked across >= 4 episodes | |
| counts: Dict[str, int] = {} | |
| for ep in episodes: | |
| for comp in ep.what_worked: | |
| counts[comp] = counts.get(comp, 0) + 1 | |
| threshold = max(4, len(episodes) - 1) if len(episodes) >= 4 else len(episodes) | |
| return [comp for comp, cnt in counts.items() if cnt >= threshold] | |
| def _compute_most_effective_action(self, episodes: List[EpisodeMemory]) -> Optional[str]: | |
| """Action type with highest average final_total_reward across episodes it appeared in.""" | |
| if not episodes: | |
| return None | |
| action_rewards: Dict[str, List[float]] = {} | |
| for ep in episodes: | |
| for action in set(ep.actions_taken): | |
| action_rewards.setdefault(action, []).append(ep.final_total_reward) | |
| if not action_rewards: | |
| return None | |
| return max(action_rewards, key=lambda a: sum(action_rewards[a]) / len(action_rewards[a])) | |
| def _compute_voice_stability(self, episodes: List[EpisodeMemory]) -> float: | |
| """Stability of R3 inferred from whether r3_cultural_alignment was in what_didnt. | |
| A proxy: episodes where R3 did NOT regress count toward stability.""" | |
| if not episodes: | |
| return 1.0 | |
| stable_count = sum( | |
| 1 for ep in episodes if "r3_cultural_alignment" not in ep.what_didnt | |
| ) | |
| return stable_count / len(episodes) | |
| def _compute_trend(self, episodes: List[EpisodeMemory]) -> str: | |
| """Slope of final_total_reward across the episode window.""" | |
| if len(episodes) < 2: | |
| return "plateauing" | |
| rewards = [ep.final_total_reward for ep in episodes] | |
| n = len(rewards) | |
| x_mean = (n - 1) / 2.0 | |
| y_mean = sum(rewards) / n | |
| numerator = sum((i - x_mean) * (rewards[i] - y_mean) for i in range(n)) | |
| denominator = sum((i - x_mean) ** 2 for i in range(n)) | |
| if denominator == 0: | |
| return "plateauing" | |
| slope = numerator / denominator | |
| if slope > 0.02: | |
| return "improving" | |
| elif slope < -0.02: | |
| return "declining" | |
| return "plateauing" | |