diff --git a/src/config.py b/src/config.py index f1a9b16..ea3a8d8 100644 --- a/src/config.py +++ b/src/config.py @@ -33,7 +33,7 @@ }, "optimal_config": { "n_gpu_layers": -1, - "n_batch": 512, + "n_batch": 1024, "n_ctx": 16384, "metal_device": "mps", "main_gpu": 0, diff --git a/src/nfd_three_story_evolve.py b/src/nfd_three_story_evolve.py index 4b743e2..111c30b 100644 --- a/src/nfd_three_story_evolve.py +++ b/src/nfd_three_story_evolve.py @@ -1,3 +1,10 @@ +from __future__ import annotations +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from typing import List, Dict + from .story import Story # Assuming Story is defined in a separate file + import sys from pathlib import Path import logging @@ -24,6 +31,16 @@ logger = logging.getLogger(__name__) +@dataclass +class Memory: + timestamp: float + interaction_type: str + resonance: float + themes: List[str] + emotional_impact: float + partner_id: Optional[str] = None + + @dataclass class StoryState: """Captures the evolving state of a story over time""" @@ -32,16 +49,22 @@ class StoryState: active_themes: List[str] = field(default_factory=list) interaction_count: int = 0 - def update(self, resonance: float, new_themes: List[str]): - self.resonance_level = 0.8 * self.resonance_level + 0.2 * resonance - self.active_themes.extend(new_themes) + def update(self, memory: Memory): + self.resonance_level = 0.8 * self.resonance_level + 0.2 * memory.resonance + self.active_themes.extend(memory.themes) self.interaction_count += 1 -class ThemeRelationshipMap: +class BaseClass: + def __init__(self): + self.logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}") + + +class ThemeRelationshipMap(BaseClass): """Manages theme relationships and their evolution""" def __init__(self): + super().__init__() # Primary theme relationships with resonance values self.primary_relationships = { ("hope", "journey"): 0.7, @@ -64,7 +87,6 @@ def __init__(self): "guidance": {"duty", "guidance", "hope"}, "solitude": {"loneliness", "nature", "subconscious"}, } - self.logger = logging.getLogger(__name__) def get_theme_resonance(self, theme1: str, theme2: str) -> float: """Get resonance between two themes""" @@ -82,15 +104,33 @@ def get_theme_resonance(self, theme1: str, theme2: str) -> float: return 0.3 * shared_groups if shared_groups > 0 else 0.1 -class StoryPerspective: +class ThemeEvolutionEngine(BaseClass): + """Handles theme evolution and perspective shifts""" + + def __init__(self): + super().__init__() + self.theme_resonance = {} # Track theme relationships + + def update_theme_resonance(self, theme: str, resonance: float): + self.theme_resonance[theme] = resonance + + def evolve_themes(self, story: "Story", interaction_history): + new_themes = set() + for interaction in interaction_history[-10:]: # Consider last 10 interactions + if interaction["resonance"] > 0.5: + new_themes.update(interaction["themes_gained"]) + story.themes = list(set(story.themes) | new_themes)[:5] # Keep top 5 themes + + +class StoryPerspective(BaseClass): """Manages a story's evolving perspective""" def __init__(self, initial_filter: np.ndarray): + super().__init__() self.filter = initial_filter.copy() self.shift_history = [] self.theme_influences = {} self.total_shift = 0.0 - self.logger = logging.getLogger(__name__) def update( self, @@ -144,8 +184,9 @@ def update( return shift -class EmotionalState: +class EmotionalState(BaseClass): def __init__(self, description: str, embedding: np.ndarray): + super().__init__() self.description = description self.embedding = embedding @@ -186,7 +227,7 @@ def __str__(self): return self.description -class Story: +class Story(BaseClass): def __init__( self, id: str, @@ -194,30 +235,39 @@ def __init__( embedding: np.ndarray, perspective_filter: np.ndarray, themes: List[str], - field: "NarrativeField", position: np.ndarray = None, velocity: np.ndarray = None, emotional_state: EmotionalState = None, protagonist_name: str = None, + field: Optional["NarrativeField"] = None, **kwargs, ): + super().__init__() self.id = id self.content = content self.embedding = embedding self.perspective_filter = perspective_filter self.themes = themes - self.field = field self.position = position if position is not None else np.random.randn(3) self.velocity = velocity if velocity is not None else np.zeros(3) self.emotional_state = emotional_state self.previous_emotional_state = None - self.memory_layer = [] + self.memory_layer: List[Memory] = [] + self.memory_state = StoryState() self.resonance_history = [] self.total_perspective_shift = 0.0 self.perspective_shifts = [] self.protagonist_name = protagonist_name + self._field = None + if field: + self.set_field(field) + + def set_field(self, field: "NarrativeField"): + self._field = field - self.logger = logging.getLogger(__name__) + @property + def field(self) -> Optional["NarrativeField"]: + return self._field def __str__(self): return f"Story {self.id} ({self.protagonist_name}): {self.content[:50]}..." @@ -276,11 +326,7 @@ def check_emotional_thresholds(self): self.trigger_melancholy_event() def evolve_themes(self, interaction_history): - new_themes = set() - for interaction in interaction_history[-10:]: # Consider last 10 interactions - if interaction["resonance"] > 0.5: - new_themes.update(interaction["themes_gained"]) - self.themes = list(set(self.themes) | new_themes)[:5] # Keep top 5 themes + self.field.theme_manager.evolve_themes(self, interaction_history) def calculate_interaction_strength(self, other_story): base_strength = self.field.detect_resonance(self, other_story) @@ -296,13 +342,14 @@ async def respond_to_environmental_event(self, event: Dict[str, Any]) -> float: await self.emotional_state.update(interaction_strength=intensity * 0.2) # Add a memory of the event - memory = { - "type": "environmental_event", - "event_name": event.get("name", "Unknown Event"), - "intensity": intensity, - "timestamp": self.field.time, - } - self.memory_layer.append(memory) + memory = Memory( + timestamp=self.field.time, + interaction_type="environmental_event", + resonance=intensity, + themes=[], + emotional_impact=intensity * 0.2, + ) + self.add_memory(memory) return intensity @@ -315,14 +362,18 @@ async def update_state(self, avg_resonance: float, avg_shift: float) -> float: ) return shift + def add_memory(self, memory: Memory): + self.memory_layer.append(memory) + self.memory_state.update(memory) + -class NarrativeFieldViz: +class NarrativeFieldViz(BaseClass): """Handles visualization of field state""" def __init__(self, field_size: int = 1024): + super().__init__() self.field_size = field_size self.history: List[Dict] = [] - self.logger = logging.getLogger(__name__) async def capture_state(self, field, timestep: int): """Capture current field state for visualization""" @@ -353,8 +404,9 @@ def _compute_resonance_map(self, field) -> Dict: return resonance_map -class NarrativeField: +class NarrativeField(BaseClass): def __init__(self, dimension: int = 1024): + super().__init__() self.dimension = dimension self.stories: List[Story] = [] self.field_memory = [] @@ -365,7 +417,7 @@ def __init__(self, dimension: int = 1024): self.resonance_threshold = 0.2 # Lower threshold to allow more interactions self.interaction_range = 3.0 # Increased range self.field_potential = np.zeros(dimension) - self.logger = logging.getLogger(__name__) + self.theme_manager = ThemeManager() def detect_resonance(self, story1: Story, story2: Story) -> float: """Calculate resonance between two stories""" @@ -414,7 +466,7 @@ def _calculate_emotional_similarity(self, story1: Story, story2: Story) -> float def add_story(self, story: Story): """Add a new story to the field""" - story.field = self # Add this line + story.set_field(self) self.stories.append(story) self._update_field_potential() self.logger.info(f"Added new story: {story.id}") @@ -435,52 +487,82 @@ async def apply_environmental_event(self, event: Dict[str, Any]): self.field_memory.append(event) -class StoryPhysics: +class StoryPhysics(BaseClass): """Handles physical behavior of stories in the field""" def __init__( self, - damping: float = 0.95, # Increased damping - attraction_strength: float = 0.1, # Reduced strength - max_force: float = 1.0, # Force limiting - max_velocity: float = 0.5, - ): # Velocity limiting + damping: float = 0.95, + attraction_strength: float = 0.2, + repulsion_strength: float = 0.1, + min_distance: float = 0.5, + interaction_range: float = 2.0, + random_force: float = 0.05, + max_force: float = 0.3, + max_velocity: float = 0.2, + target_zone_radius: float = 10.0, + ): + super().__init__() + # Physics parameters self.damping = damping self.attraction_strength = attraction_strength + self.repulsion_strength = repulsion_strength + self.min_distance = min_distance + self.interaction_range = interaction_range + self.random_force = random_force + + # Movement limits self.max_force = max_force self.max_velocity = max_velocity - self.logger = logging.getLogger(__name__) - - def _normalize_force(self, force: np.ndarray) -> np.ndarray: - """Normalize force vector to prevent exponential growth""" - magnitude = np.linalg.norm(force) - if magnitude > self.max_force: - return (force / magnitude) * self.max_force - return force - - def _limit_velocity(self, velocity: np.ndarray) -> np.ndarray: - """Limit velocity magnitude""" - magnitude = np.linalg.norm(velocity) - if magnitude > self.max_velocity: - return (velocity / magnitude) * self.max_velocity - return velocity + self.target_zone_radius = target_zone_radius def update_story_motion(self, story: Story, field: NarrativeField, timestep: int): - """Update story position and velocity based on field forces""" - # Compute net force from other stories + """Update story position and velocity with balanced forces""" net_force = np.zeros(3) + + # Forces from other stories for other in field.stories: if other.id != story.id: - # Compute force based on resonance - resonance = field.detect_resonance(story, other) direction = other.position - story.position - distance = np.linalg.norm(direction) + 1e-6 # Prevent division by zero + distance = np.linalg.norm(direction) + 1e-6 + direction_normalized = direction / distance + + # Resonance-based attraction + resonance = field.detect_resonance(story, other) + attraction = self.attraction_strength * resonance * direction_normalized + + # Distance-based repulsion + repulsion = ( + -self.repulsion_strength * direction_normalized / (distance**2) + ) + if distance < self.min_distance: + repulsion *= 2.0 # Stronger repulsion when too close - # Scale force by distance with a minimum threshold - force = self.attraction_strength * resonance * direction / distance - net_force += force + net_force += attraction + repulsion - # Normalize and limit forces + # Containment force - quadratic increase with distance + displacement = story.position + distance_from_center = np.linalg.norm(displacement) + if distance_from_center > self.target_zone_radius: + containment = ( + -0.1 + * (distance_from_center / self.target_zone_radius) ** 2 + * displacement + ) + net_force += containment + + # Random exploration force - varies with time + random_direction = np.random.randn(3) + random_direction /= np.linalg.norm(random_direction) + exploration_force = ( + self.random_force * random_direction * np.sin(timestep / 100) + ) + net_force += exploration_force + + # Balance z-axis movement + net_force[2] *= 0.3 # Reduce but don't eliminate z-axis movement + + # Apply force limits net_force = self._normalize_force(net_force) # Update velocity with damping @@ -491,33 +573,61 @@ def update_story_motion(self, story: Story, field: NarrativeField, timestep: int # Update position story.position += story.velocity - self.logger.debug( - f"Story {story.id} - " - f"Position: {story.position}, " - f"Velocity: {np.linalg.norm(story.velocity):.3f}" - ) + # Log significant movements + if timestep % 100 == 0: + self.logger.debug( + f"Story {story.id} at t={timestep}:\n" + f" Position: {story.position}\n" + f" Velocity: {np.linalg.norm(story.velocity):.3f}\n" + f" Force: {np.linalg.norm(net_force):.3f}" + ) + + def _normalize_force(self, force: np.ndarray) -> np.ndarray: + """Normalize force vector to prevent exponential growth""" + magnitude = np.linalg.norm(force) + if magnitude > self.max_force: + return (force / magnitude) * self.max_force + return force + + def _limit_velocity(self, velocity: np.ndarray) -> np.ndarray: + """Limit velocity magnitude""" + magnitude = np.linalg.norm(velocity) + if magnitude > self.max_velocity: + return (velocity / magnitude) * self.max_velocity + return velocity + + def apply_field_constraints(self, stories: List[Story]): + """Apply global constraints to all stories""" + # Find center of mass + com = np.mean([s.position for s in stories], axis=0) + # If stories are drifting too far as a group, pull them back + if np.linalg.norm(com) > 5.0: + for story in stories: + # Apply centering force proportional to distance from origin + centering = -0.1 * story.position + story.velocity += self._normalize_force(centering) + story.velocity = self._limit_velocity(story.velocity) -class EnhancedCollectiveStoryEngine: + +class EnhancedCollectiveStoryEngine(BaseClass): """Enhanced version with more sophisticated pattern detection""" def __init__(self, field: NarrativeField): + super().__init__() self.field = field self.collective_memories = [] self.story_states: Dict[str, StoryState] = {} - self.logger = logging.getLogger(__name__) self.collective_story = "" # Add this line to initialize the collective story async def update_story_states(self): for story in self.field.stories: recent_memories = story.memory_layer[-5:] # Get the 5 most recent memories if recent_memories: - avg_resonance = np.mean( - [m.get("resonance", 0) for m in recent_memories] - ) + avg_resonance = np.mean([m.resonance for m in recent_memories]) perspective_shifts = [] for m in recent_memories: - shift = m.get("perspective_shift", 0) + shift = m.emotional_impact if asyncio.iscoroutine(shift): shift = await shift perspective_shifts.append(shift) @@ -586,97 +696,85 @@ def summarize_story(self, story: Story) -> str: return f"Story {story.id} resonates with {', '.join(story.themes[:3])}, its journey marked by {len(story.memory_layer)} memories." -class ThemeEvolutionEngine: - """Handles theme evolution and perspective shifts""" - +class ThemeManager(BaseClass): def __init__(self): - self.logger = logging.getLogger(__name__) - self.theme_resonance = {} # Track theme relationships - self.logger = logging.getLogger(__name__) - - -class EnhancedInteractionEngine: - # ... (other methods remain the same) - - async def process_interaction(self, story1: Story, story2: Story): - if story1.id == story2.id: - return # Prevent a story from interacting with itself - - interaction_type = await self.determine_interaction_type(story1, story2) - resonance = self.field.detect_resonance(story1, story2) - - if resonance > self.field.resonance_threshold: - theme_impact = self.calculate_theme_impact(story1, story2) - emotional_change = self._calculate_emotional_influence(story1, story2) + super().__init__() + self.relationship_map = ThemeRelationshipMap() + self.evolution_engine = ThemeEvolutionEngine() - perspective_shift = await story1.update_perspective( - story2, theme_impact, resonance, emotional_change, interaction_type - ) + async def process_theme_interaction(self, story1: Story, story2: Story): + shared_themes = set(story1.themes) & set(story2.themes) + theme_impact = len(shared_themes) / max(len(story1.themes), len(story2.themes)) - await story1.update_emotional_state( - story2, interaction_type, resonance, self.llm - ) + for theme in shared_themes: + resonance = self.relationship_map.get_theme_resonance(theme, theme) + self.evolution_engine.update_theme_resonance(theme, resonance) - # Add memory of interaction - memory = { - "type": "interaction", - "partner_id": story2.id, - "resonance": resonance, - "interaction_type": interaction_type, - "perspective_shift": perspective_shift, - "timestamp": self.field.time, - } - story1.memory_layer.append(memory) + return theme_impact - return resonance, interaction_type - return 0, None + def get_theme_resonance(self, theme1: str, theme2: str) -> float: + return self.relationship_map.get_theme_resonance(theme1, theme2) - # ... (rest of the class remains the same) + def evolve_themes(self, story: Story, interaction_history): + self.evolution_engine.evolve_themes(story, interaction_history) -class StoryInteractionEngine: +class BaseInteractionEngine(BaseClass): def __init__(self, field: NarrativeField): + super().__init__() self.field = field - self.logger = logging.getLogger(__name__) + async def process_interaction(self, story1: Story, story2: Story): + # Base implementation + raise NotImplementedError("Subclasses must implement process_interaction") + + +class StoryInteractionEngine(BaseInteractionEngine): async def process_interaction(self, story1: Story, story2: Story): # Basic interaction processing resonance = self.field.detect_resonance(story1, story2) if resonance > self.field.resonance_threshold: # Perform basic interaction logic here pass + return resonance, None -class EnhancedInteractionEngine(StoryInteractionEngine): +class EnhancedInteractionEngine(BaseInteractionEngine): def __init__(self, field: NarrativeField, llm: LanguageModel): super().__init__(field) self.llm = llm - self.logger = logging.getLogger(__name__) async def determine_interaction_type(self, story1: Story, story2: Story) -> str: - prompt = f""" - Story 1 Themes: {', '.join(story1.themes)} - Story 1 Emotional State: {story1.emotional_state} - - Story 2 Themes: {', '.join(story2.themes)} - Story 2 Emotional State: {story2.emotional_state} - - Based on the themes and emotional states of these two stories, what type of interaction might occur between them? - Choose from: collaboration, conflict, inspiration, reflection, transformation, challenge, or synthesis. - - Respond only with one chosen interaction type as a SINGLE WORD response. - """ - return await self.llm.generate(prompt) + # Calculate the similarity between the stories' themes + shared_themes = set(story1.themes) & set(story2.themes) + theme_similarity = len(shared_themes) / max( + len(story1.themes), len(story2.themes) + ) + + # Calculate emotional state similarity + emotional_similarity = self.field._calculate_emotional_similarity( + story1, story2 + ) + + # Determine interaction type based on similarities + if theme_similarity > 0.5 and emotional_similarity > 0.5: + return "collaboration" + elif theme_similarity < 0.2 and emotional_similarity < 0.2: + return "conflict" + else: + return "neutral" async def process_interaction(self, story1: Story, story2: Story): if story1.id == story2.id: - return # Prevent a story from interacting with itself + return 0, None # Prevent a story from interacting with itself interaction_type = await self.determine_interaction_type(story1, story2) resonance = self.field.detect_resonance(story1, story2) if resonance > self.field.resonance_threshold: - theme_impact = self.calculate_theme_impact(story1, story2) + theme_impact = await self.field.theme_manager.process_theme_interaction( + story1, story2 + ) emotional_change = self._calculate_emotional_influence(story1, story2) perspective_shift = await story1.update_perspective( @@ -687,16 +785,16 @@ async def process_interaction(self, story1: Story, story2: Story): story2, interaction_type, resonance, self.llm ) - # Add memory of interaction - memory = { - "type": "interaction", - "partner_id": story2.id, - "resonance": resonance, - "interaction_type": interaction_type, - "perspective_shift": perspective_shift, # This is now awaited - "timestamp": self.field.time, - } - story1.memory_layer.append(memory) + # Create and add memory + memory = Memory( + timestamp=self.field.time, + interaction_type=interaction_type, + resonance=resonance, + themes=list(set(story1.themes) & set(story2.themes)), + emotional_impact=emotional_change, + partner_id=story2.id, + ) + story1.add_memory(memory) return resonance, interaction_type return 0, None @@ -711,166 +809,21 @@ def _calculate_emotional_influence(self, story1: Story, story2: Story) -> float: return 0.5 # Return a value between 0 and 1 -class StoryPhysics: - """Handles physical behavior of stories in the field""" - - def __init__(self): - # Physics parameters - self.damping = 0.95 # Slightly reduced damping - self.attraction_strength = 0.2 # Stronger attraction - self.repulsion_strength = 0.1 # Add repulsion to prevent collapse - self.min_distance = 0.5 # Minimum distance between stories - self.interaction_range = 2.0 # Range for story interactions - self.random_force = 0.05 # Small random force for exploration - - # Movement limits - self.max_force = 0.3 - self.max_velocity = 0.2 - self.target_zone_radius = 10.0 # Desired story movement range - - self.logger = logging.getLogger(__name__) - - def update_story_motion(self, story: Story, field: NarrativeField, timestep: int): - """Update story position and velocity with balanced forces""" - net_force = np.zeros(3) - - # Forces from other stories - for other in field.stories: - if other.id != story.id: - direction = other.position - story.position - distance = np.linalg.norm(direction) + 1e-6 - direction_normalized = direction / distance - - # Resonance-based attraction - resonance = field.detect_resonance(story, other) - attraction = self.attraction_strength * resonance * direction_normalized - - # Distance-based repulsion - repulsion = ( - -self.repulsion_strength * direction_normalized / (distance**2) - ) - if distance < self.min_distance: - repulsion *= 2.0 # Stronger repulsion when too close - - net_force += attraction + repulsion - - # Containment force - quadratic increase with distance - displacement = story.position - distance_from_center = np.linalg.norm(displacement) - if distance_from_center > self.target_zone_radius: - containment = ( - -0.1 - * (distance_from_center / self.target_zone_radius) ** 2 - * displacement - ) - net_force += containment - - # Random exploration force - varies with time - random_direction = np.random.randn(3) - random_direction /= np.linalg.norm(random_direction) - exploration_force = ( - self.random_force * random_direction * np.sin(timestep / 100) - ) - net_force += exploration_force - - # Balance z-axis movement - net_force[2] *= 0.3 # Reduce but don't eliminate z-axis movement - - # Apply force limits - net_force = self._normalize_force(net_force) - - # Update velocity with damping - story.velocity = self._limit_velocity( - (1 - self.damping) * story.velocity + net_force - ) - - # Update position - story.position += story.velocity - - # Log significant movements - if timestep % 100 == 0: - self.logger.debug( - f"Story {story.id} at t={timestep}:\n" - f" Position: {story.position}\n" - f" Velocity: {np.linalg.norm(story.velocity):.3f}\n" - f" Force: {np.linalg.norm(net_force):.3f}" - ) - - def _normalize_force(self, force: np.ndarray) -> np.ndarray: - """Normalize force vector to prevent exponential growth""" - magnitude = np.linalg.norm(force) - if magnitude > self.max_force: - return (force / magnitude) * self.max_force - return force - - def _limit_velocity(self, velocity: np.ndarray) -> np.ndarray: - """Limit velocity magnitude""" - magnitude = np.linalg.norm(velocity) - if magnitude > self.max_velocity: - return (velocity / magnitude) * self.max_velocity - return velocity - - def apply_field_constraints(self, stories: List[Story]): - """Apply global constraints to all stories""" - # Find center of mass - com = np.mean([s.position for s in stories], axis=0) - - # If stories are drifting too far as a group, pull them back - if np.linalg.norm(com) > 5.0: - for story in stories: - # Apply centering force proportional to distance from origin - centering = -0.1 * story.position - story.velocity += self._normalize_force(centering) - story.velocity = self._limit_velocity(story.velocity) - - def _normalize_force(self, force: np.ndarray) -> np.ndarray: - """Normalize force vector to prevent exponential growth""" - magnitude = np.linalg.norm(force) - if magnitude > self.max_force: - return (force / magnitude) * self.max_force - return force - - def _limit_velocity(self, velocity: np.ndarray) -> np.ndarray: - """Limit velocity magnitude""" - magnitude = np.linalg.norm(velocity) - if magnitude > self.max_velocity: - return (velocity / magnitude) * self.max_velocity - return velocity - - def apply_field_constraints(self, stories: List[Story]): - """Apply global constraints to all stories""" - # Find center of mass - com = np.mean([s.position for s in stories], axis=0) - - # If stories are drifting too far as a group, pull them back - if np.linalg.norm(com) > 5.0: - for story in stories: - # Apply centering force proportional to distance from origin - centering = -0.1 * story.position - story.velocity += self._normalize_force(centering) - story.velocity = self._limit_velocity(story.velocity) - - -class StoryJourneyLogger: - """Tracks and logs the journey of stories through the narrative field""" +class EnhancedJourneyLogger(BaseClass): + """Enhanced logger to track and analyze story journeys through the narrative field""" def __init__(self): - self.logger = logging.getLogger(__name__) + super().__init__() self.journey_log = {} - self.total_distances = {} # Track cumulative distance for each story - self.significant_events = [] # Track important moments + self.total_distances = {} + self.significant_events = [] + self.emotional_history = {} def log_interaction( self, story1: Story, story2: Story, resonance: float, interaction_type: str ): - latest_memory = story1.memory_layer[-1] if story1.memory_layer else {} - perspective_shift = latest_memory.get("perspective_shift", 0) - - # If perspective_shift is a coroutine, we need to run it in an event loop - if asyncio.iscoroutine(perspective_shift): - perspective_shift = asyncio.get_event_loop().run_until_complete( - perspective_shift - ) + latest_memory = story1.memory_layer[-1] if story1.memory_layer else None + perspective_shift = latest_memory.emotional_impact if latest_memory else 0 log_entry = ( f"Interaction between {story1.id} and {story2.id}:\n" @@ -885,8 +838,11 @@ def log_interaction( ) self.logger.info(log_entry) + # Log emotional states + self.log_emotional_state(story1) + self.log_emotional_state(story2) + def log_story_state(self, story: Story, timestep: float): - """Log detailed story state and track journey metrics""" if story.id not in self.journey_log: self.journey_log[story.id] = [] self.total_distances[story.id] = 0.0 @@ -921,8 +877,19 @@ def log_story_state(self, story: Story, timestep: float): } self.journey_log[story.id].append(state) + def log_emotional_state(self, story: Story): + if story.id not in self.emotional_history: + self.emotional_history[story.id] = [] + + self.emotional_history[story.id].append( + { + "timestep": story.field.time, + "emotional_state": story.emotional_state.description, + "embedding": story.emotional_state.embedding.tolist(), + } + ) + def summarize_journey(self, story: Story): - """Enhanced journey summary with accumulated perspective shifts""" journey = self.journey_log.get(story.id, []) if not journey: return @@ -949,12 +916,16 @@ def summarize_journey(self, story: Story): # Safely get unique interactions unique_interactions = { - m["interacted_with"] - for m in story.memory_layer - if "interacted_with" in m + m["partner_id"] for m in story.memory_layer if "partner_id" in m } num_unique_interactions = len(unique_interactions) + # Emotional journey analysis + emotional_states = self.emotional_history.get(story.id, []) + emotional_changes = ( + len(emotional_states) - 1 if len(emotional_states) > 1 else 0 + ) + self.logger.info( f"\n=== Journey Summary for {story.id} ===\n" f"Movement Metrics:\n" @@ -967,131 +938,47 @@ def summarize_journey(self, story: Story): f" Total Perspective Shift: {story.total_perspective_shift:.4f}\n" f" Average Shift Magnitude: {avg_shift:.4f}\n" f" Significant Perspective Changes: {len(significant_shifts)}\n" + f"\nEmotional Journey:\n" + f" Emotional State Changes: {emotional_changes}\n" + f" Initial Emotional State: {emotional_states[0]['emotional_state'] if emotional_states else 'N/A'}\n" + f" Final Emotional State: {emotional_states[-1]['emotional_state'] if emotional_states else 'N/A'}\n" f"\nFinal State:\n" f" Position: {end_state['position']}\n" f" Velocity: {end_state['velocity']}\n" f"\nSignificant Events: {len(story.memory_layer)}" ) - -class JourneyLogger: - def log_interaction( - self, story1: Story, story2: Story, resonance: float, interaction_type: str - ): - latest_memory = story1.memory_layer[-1] if story1.memory_layer else {} - perspective_shift = latest_memory.get("perspective_shift", 0) - - # If perspective_shift is a coroutine, we need to run it in an event loop - if asyncio.iscoroutine(perspective_shift): - perspective_shift = asyncio.get_event_loop().run_until_complete( - perspective_shift - ) - - log_entry = ( - f"Interaction between {story1.id} and {story2.id}:\n" - f" Resonance: {resonance:.2f}\n" - f" Interaction Type: {interaction_type}\n" - f" Perspective Shift: {perspective_shift:.4f}\n" - f" Shared Themes: {set(story1.themes) & set(story2.themes)}\n" - f" Distance: {np.linalg.norm(story1.position - story2.position):.2f}\n" - f" Positions:\n" - f" {story1.id}: {story1.position}\n" - f" {story2.id}: {story2.position}\n" - ) - self.logger.info(log_entry) - - def _format_emotional_change(self, story: Story) -> str: - return ", ".join( - [f"{e}: {v:.2f}" for e, v in story.emotional_state.get_dominant_emotions()] - ) - - def log_story_state(self, story: Story, timestep: float): - """Log detailed story state and track journey metrics""" - if story.id not in self.journey_log: - self.journey_log[story.id] = [] - self.total_distances[story.id] = 0.0 - - # Calculate movement since last state - if self.journey_log[story.id]: - last_pos = self.journey_log[story.id][-1]["position"] - movement = np.linalg.norm(story.position - last_pos) - self.total_distances[story.id] += movement - - # Log significant movements - if movement > 0.5: # Threshold for significant movement - self.significant_events.append( - { - "type": "movement", - "time": timestep, - "story_id": story.id, - "distance": movement, - "direction": story.velocity - / (np.linalg.norm(story.velocity) + 1e-6), - } - ) - - # Store current state - state = { - "timestep": timestep, - "position": story.position.copy(), - "velocity": story.velocity.copy(), - "memory_count": len(story.memory_layer), - "perspective_sum": story.perspective_filter.sum(), - "total_distance": self.total_distances[story.id], - } - self.journey_log[story.id].append(state) - - def summarize_journey(self, story: Story): - """Enhanced journey summary with accumulated perspective shifts""" + def get_journey_analytics(self, story: Story): journey = self.journey_log.get(story.id, []) if not journey: - return - - start_state = journey[0] - end_state = journey[-1] - - # Calculate metrics - total_distance = self.total_distances[story.id] - direct_distance = np.linalg.norm( - end_state["position"] - start_state["position"] - ) - wandering_ratio = total_distance / (direct_distance + 1e-6) - - # Perspective analysis - significant_shifts = [ - s for s in story.perspective_shifts if s["magnitude"] > 0.01 - ] - avg_shift = ( - np.mean([s["magnitude"] for s in significant_shifts]) - if significant_shifts - else 0 - ) - - # Safely get unique interactions - unique_interactions = { - m["interacted_with"] - for m in story.memory_layer - if "interacted_with" in m + return {} + + theme_counts = {} + for memory in story.memory_layer: + for theme in memory.themes: + theme_counts[theme] = theme_counts.get(theme, 0) + 1 + + most_influential_themes = sorted( + story.perspective.theme_influences.items(), key=lambda x: x[1], reverse=True + )[:3] + + return { + "total_memories": len(story.memory_layer), + "unique_interactions": len( + { + m.partner_id + for m in story.memory_layer + if m.partner_id is not None + } + ), + "theme_exposure": theme_counts, + "total_perspective_shift": story.total_perspective_shift, + "most_influential_themes": most_influential_themes, + "perspective_shifts": len( + [s for s in story.perspective_shifts if s["magnitude"] > 0.01] + ), + "emotional_changes": len(self.emotional_history.get(story.id, [])) - 1, } - num_unique_interactions = len(unique_interactions) - - self.logger.info( - f"\n=== Journey Summary for {story.id} ===\n" - f"Movement Metrics:\n" - f" Total Distance Traveled: {total_distance:.2f}\n" - f" Direct Distance (start to end): {direct_distance:.2f}\n" - f" Wandering Ratio: {wandering_ratio:.2f}\n" - f"\nInteraction Metrics:\n" - f" Memories Formed: {len(story.memory_layer)}\n" - f" Unique Interactions: {num_unique_interactions}\n" - f" Total Perspective Shift: {story.total_perspective_shift:.4f}\n" - f" Average Shift Magnitude: {avg_shift:.4f}\n" - f" Significant Perspective Changes: {len(significant_shifts)}\n" - f"\nFinal State:\n" - f" Position: {end_state['position']}\n" - f" Velocity: {end_state['velocity']}\n" - f"\nSignificant Events: {len(story.memory_layer)}" - ) async def create_story_cluster(): @@ -1105,38 +992,11 @@ async def create_story_cluster(): return base_positions + np.random.randn(3, 3) * 0.2 -def summarize_story_journey(story: Story): - """Enhanced journey summary with perspective analysis""" - theme_counts = {} - for memory in story.memory_layer: - for theme in memory["themes"]: - theme_counts[theme] = theme_counts.get(theme, 0) + 1 - - most_influential_themes = sorted( - story.perspective.theme_influences.items(), key=lambda x: x[1], reverse=True - )[:3] - - total_perspective_shift = story.perspective.total_shift - - return { - "total_memories": len(story.memory_layer), - "unique_interactions": len( - set(m["interacted_with"] for m in story.memory_layer) - ), - "theme_exposure": theme_counts, - "total_perspective_shift": total_perspective_shift, - "most_influential_themes": most_influential_themes, - "perspective_shifts": len( - [s for s in story.perspective.shift_history if s["magnitude"] > 0.01] - ), - } - - -class DynamicThemeGenerator: +class DynamicThemeGenerator(BaseClass): def __init__(self, llm: LanguageModel): + super().__init__() self.llm = llm self.theme_cache = set() - self.logger = logging.getLogger(__name__) async def generate_themes(self, context: str, num_themes: int = 3) -> List[str]: prompt = f"""Given the context '{context}', generate {num_themes} unique, single-word themes that could be present in a story. Output ONLY a valid JSON array of strings, nothing else. Example: @@ -1174,16 +1034,16 @@ def get_random_themes(self, num_themes: int = 3) -> List[str]: ) -class DynamicStoryGenerator: +class DynamicStoryGenerator(BaseClass): def __init__(self, llm: LanguageModel, theme_generator: DynamicThemeGenerator): + super().__init__() self.llm = llm self.theme_generator = theme_generator - self.logger = logging.getLogger(__name__) async def generate_story(self, field: NarrativeField) -> Story: themes = await self.theme_generator.generate_themes("Create a new story") - prompt = f"Write a short positive, neutral, or negative story (5-8 sentences) incorporating the themes: {', '.join(themes)}. Make it interesting and engaging. Make it personal and emotional. Make it unique and memorable, with one clearly defined protagonist. Start the story by introducing the protagonist's name." + prompt = f"Write a short positive, neutral, or negative story (2 - 3 sentences) incorporating the themes: {', '.join(themes)}. Make it interesting and engaging. Make it personal and emotional. Name names." content = await self.llm.generate(prompt) # Extract the protagonist's name from the first sentence @@ -1210,16 +1070,16 @@ async def generate_story(self, field: NarrativeField) -> Story: ) async def generate_emotional_state(self, content: str) -> EmotionalState: - prompt = f"""Given the story: '{content}', describe its positive, neutral, or negative emotional state in 2-3 sentences: """ + prompt = f"""Given the story: '{content}', describe its positive, neutral, or negative emotional state in 2-3 sentences. How does it feel?: """ description = await self.llm.generate(prompt) embedding = await self.llm.generate_embedding(description) return EmotionalState(description, np.array(embedding)) -class EnvironmentalEventGenerator: +class EnvironmentalEventGenerator(BaseClass): def __init__(self, llm: LanguageModel): + super().__init__() self.llm = llm - self.logger = logging.getLogger(__name__) async def generate_event(self) -> Dict[str, Any]: prompt = "Generate a random positive or negative environmental event for a narrative field. Include an event name, description, and intensity (0.0 to 1.0)." @@ -1281,12 +1141,12 @@ async def simulate_field(): interaction_engine = EnhancedInteractionEngine(field, llm) # Generate initial stories - for _ in range(1): + for _ in range(9): story = await story_generator.generate_story(field) field.add_story(story) # Add journey logger - journey_logger = StoryJourneyLogger() + journey_logger = EnhancedJourneyLogger() # Simulation loop for t in range(100): @@ -1391,6 +1251,8 @@ async def simulate_field(): logger.info("\n=== Final Journey Summaries ===") for story in field.stories: journey_logger.summarize_journey(story) + analytics = journey_logger.get_journey_analytics(story) + # Use analytics as needed logger.info("Narrative field simulation completed")