diff --git a/README.md b/README.md index cbeae98..56755c7 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,7 @@ Friction Flow is an advanced Python-based research project aimed at developing a ## Technical Stack **Python**: Core programming language (version >= 3.12 recommended) + - **PyTorch**: For neural network components and tensor operations - **Transformers**: For integration with pre-trained language models - **Ray**: For distributed computing @@ -51,12 +52,49 @@ Friction Flow is an advanced Python-based research project aimed at developing a - `src/`: Contains the core source code - `nfs_story_waves.py`: Simulation components for narrative field dynamics + - `nfd_three_story_evolve.py`: Enhanced simulation with three interacting stories - `nfs_simple_lab_scenario.py`: Example scenario implementation - `language_models.py`: Interfaces for various language models - `config.py`: Configuration settings - `tests/`: Unit and integration tests - `pocs/`: Proof of concept implementations -- `.github/`: Issue templates and CI/CD workflows + +## Latest Proof of Concept Implementations + +### nfs_story_waves.py + +This PoC implements a quantum-inspired approach to modeling narrative dynamics. Key features include: + +- `NarrativeWave`: Represents stories as quantum wave functions in a narrative field. +- `NarrativeFieldSimulator`: Main class for simulating the evolution of the narrative field. +- Quantum-inspired operations for story interactions and field evolution. +- Pattern detection and emergence analysis. +- Frequency analysis of narrative field dynamics. + +This implementation explores the potential of quantum concepts in modeling complex narrative interactions and emergent behaviors. + +### nfd_three_story_evolve.py + +This PoC focuses on the detailed evolution of three interacting stories within a narrative field. Key components include: + +- `Story`: Enhanced representation of individual narratives with emotional states and memory. +- `StoryPhysics`: Simulates the motion and interactions of stories in the narrative field. +- `EnhancedCollectiveStoryEngine`: Manages the collective narrative emerging from story interactions. +- `EnhancedInteractionEngine`: Processes detailed interactions between stories. + +This implementation provides a more granular look at story evolution and interaction dynamics. + +### nfs_simple_lab_scenario.py + +This PoC demonstrates a practical application of the Narrative Field System in a simulated lab environment. Key features include: + +- `NarrativeField`: Main class orchestrating the entire system. +- `FieldAnalyzer`: Analyzes the impact of stories on the field state. +- `ResonanceDetector`: Finds similarities between stories. +- `VectorStore`: Manages storage and retrieval of story embeddings. +- A demo scenario simulating events in a research lab, showcasing how the system can model real-world narrative dynamics. + +This implementation showcases how the Narrative Field System can be applied to analyze and track narrative dynamics in a specific context. ## Development Guidelines @@ -77,6 +115,7 @@ pytest tests/ ## CI/CD The project uses GitHub Actions for continuous integration. The workflow includes: + - Setting up Python 3.12.6 - Installing dependencies - Running tests diff --git a/src/nfd_three_story_evolve.md b/src/nfd_three_story_evolve.md index cdef5c7..de5d99e 100644 --- a/src/nfd_three_story_evolve.md +++ b/src/nfd_three_story_evolve.md @@ -2,19 +2,21 @@ In the quiet hum of the narrative field, three stories drift, each with its own essence, its own silent history. They move, not in straight lines but in subtle, winding trajectories, always pulled by something unseen. This is their story. -**The First Story: “The Waiting Lighthouse”** +## Stories -The first story has a lighthouse as its core. Its themes are **loneliness** and **enduring duty**. It waits, steady and vigilant, with a soft glow in the field—its core narrative like a beam of light, reaching out. This light, encoded in its **core embedding**, represents its themes of solitude and patience. A quiet sense of **hope** lingers in its memory layer, an echo of past encounters, though that hope has started to fade, losing brightness with time. +### The First Story: "The Waiting Lighthouse" + +The first story has a lighthouse as its core. Its themes are loneliness and enduring duty. It waits, steady and vigilant, with a soft glow in the field—its core narrative like a beam of light, reaching out. This light, encoded in its **core embedding**, represents its themes of solitude and patience. A quiet sense of **hope** lingers in its memory layer, an echo of past encounters, though that hope has started to fade, losing brightness with time. The lighthouse moves slowly, unhurried. Its **perspective filter** tunes it to notice other stories that carry tones of seeking or loss—stories that might, even if briefly, cross its light. It moves like the light it casts: careful, steady, not actively seeking but ever open to something beyond itself. -**The Second Story: “The Wandering Forest Path”** +### The Second Story: "The Wandering Forest Path" The second story is the tale of a traveler lost in a forest, its core theme an **exploration tinged with fear and curiosity**. Its embedding is close to the lighthouse, drawn by a kind of gravitational pull, for the themes of seeking and being found resonate between them. This path has traces of memories from past interactions in the forest—glimpses of fleeting encounters, echoes of other travelers. Its perspective filter is tuned to notice stability and guidance, a longing for direction in its wandering path. The path moves more restlessly, pulled by hints of guidance in the field. It propels itself through the field as if following shadows, creating ripples in the nearby stories. Each ripple shifts its own embedding slightly, pushing it forward with every new encounter. -**The Third Story: “The Dream of Flight”** +### The Third Story: "The Dream of Flight" The third story is light and free—a child’s dream of flying, untethered by gravity. Its embedding holds themes of **freedom, innocence, and boundless wonder**. It hovers nearby, drifting like a leaf on the wind, pulled by currents of curiosity. But in its memories are faint imprints of past encounters with stories grounded and heavy; a slight trace of yearning tints its core. @@ -22,7 +24,7 @@ Its movement is sporadic, darting and circling, unconcerned with direction but d --- -## First Encounter: The Lighthouse and the Path** +## First Encounter: The Lighthouse and the Path The field, attuned to the proximity of these stories, senses a resonance between the lighthouse and the path, softly adjusting their trajectories. Slowly, they drift closer. The lighthouse, aware of a nearby story with themes of seeking, extends its light, casting a gentle sense of comfort toward the path. The path feels this and pauses, sensing direction in the lighthouse’s presence. @@ -35,7 +37,7 @@ The encounter shifts their **embeddings** slightly: - The path’s position in semantic space shifts a bit closer to stability, while the lighthouse gains a touch of motion in its otherwise steadfast embedding. -## Second Encounter: The Path and the Dream** +## Second Encounter: The Path and the Dream The path, now feeling a faint pull of comfort, continues forward in the field until it senses something playful nearby: the dream of flight. The path is drawn in by the lightness of the dream, and the two stories circle each other, curious but hesitant. The dream of flight feels the path’s echo of hope and seeking, which awakens a new sensation—a hint of groundedness in its own memory layer. @@ -49,7 +51,7 @@ These memories subtly alter their **perspective filters**: - The path begins to notice stories with a sense of lightness and play, softening its need for guidance. - The dream, meanwhile, gains a slight “weight,” now attuned to stories carrying themes of longing and groundedness. -## Third Encounter: The Collective Shockwave** +## Third Encounter: The Collective Shockwave Time passes, and a **pulse** moves through the field. This is a collective event generated by the **Collective Story Engine**—a moment where the narrative field takes on a dominant theme of **seeking connection**. This pulse temporarily heightens each story’s awareness, expanding their perception to even distant stories in the field. diff --git a/src/nfd_three_story_evolve.py b/src/nfd_three_story_evolve.py index 4c141eb..bd6ccbe 100644 --- a/src/nfd_three_story_evolve.py +++ b/src/nfd_three_story_evolve.py @@ -3,7 +3,6 @@ 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 @@ -33,6 +32,18 @@ @dataclass class Memory: + """ + Represents a memory of an interaction or event in the narrative field. + + Attributes: + timestamp (float): The time when the memory was created. + interaction_type (str): The type of interaction (e.g., "collaboration", "conflict"). + resonance (float): The strength of the interaction's impact. + themes (List[str]): Themes associated with the memory. + emotional_impact (float): The emotional intensity of the memory. + partner_id (Optional[str]): The ID of the interacting story, if applicable. + """ + timestamp: float interaction_type: str resonance: float @@ -43,25 +54,51 @@ class Memory: @dataclass class StoryState: - """Captures the evolving state of a story over time""" + """ + Captures the evolving state of a story over time. + + Attributes: + resonance_level (float): The current resonance level of the story. + active_themes (List[str]): Currently active themes in the story. + interaction_count (int): Number of interactions the story has had. + """ resonance_level: float = 0.0 active_themes: List[str] = field(default_factory=list) interaction_count: int = 0 def update(self, memory: Memory): + """ + Updates the story state based on a new memory. + + Args: + memory (Memory): The new memory to incorporate into the state. + """ + # Update resonance level with a weighted average self.resonance_level = 0.8 * self.resonance_level + 0.2 * memory.resonance + # Add new themes from the memory self.active_themes.extend(memory.themes) + # Increment the interaction count self.interaction_count += 1 class BaseClass: + """ + A base class that sets up logging for derived classes. + """ + def __init__(self): + # Create a logger with the name of the module and the class self.logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}") class ThemeRelationshipMap(BaseClass): - """Manages theme relationships and their evolution""" + """ + Manages theme relationships and their evolution in the narrative field. + + This class maintains a map of primary theme relationships and theme groups, + allowing for the calculation of resonance between different themes. + """ def __init__(self): super().__init__() @@ -80,7 +117,7 @@ def __init__(self): ("journey", "nature"): 0.7, } - # Secondary theme groups + # Secondary theme groups for indirect relationships self.theme_groups = { "exploration": {"journey", "discovery", "nature"}, "inner_world": {"imagination", "subconscious", "freedom"}, @@ -89,32 +126,59 @@ def __init__(self): } def get_theme_resonance(self, theme1: str, theme2: str) -> float: - """Get resonance between two themes""" - # Check direct relationship + """ + Calculate the resonance between two themes. + + Args: + theme1 (str): The first theme. + theme2 (str): The second theme. + + Returns: + float: The resonance value between the two themes. + """ + # Check for direct relationship in primary relationships if (theme1, theme2) in self.primary_relationships: return self.primary_relationships[(theme1, theme2)] if (theme2, theme1) in self.primary_relationships: return self.primary_relationships[(theme2, theme1)] + # Check for indirect relationships through theme groups shared_groups = sum( - 1 - for group in self.theme_groups.values() - if theme1 in group and theme2 in group + theme1 in group and theme2 in group for group in self.theme_groups.values() ) return 0.3 * shared_groups if shared_groups > 0 else 0.1 class ThemeEvolutionEngine(BaseClass): - """Handles theme evolution and perspective shifts""" + """ + Handles theme evolution and perspective shifts in the narrative field. + + This class is responsible for updating theme resonances and evolving + themes based on story interactions. + """ def __init__(self): super().__init__() self.theme_resonance = {} # Track theme relationships def update_theme_resonance(self, theme: str, resonance: float): + """ + Update the resonance value for a given theme. + + Args: + theme (str): The theme to update. + resonance (float): The new resonance value. + """ self.theme_resonance[theme] = resonance def evolve_themes(self, story: "Story", interaction_history): + """ + Evolve the themes of a story based on recent interactions. + + Args: + story (Story): The story whose themes are being evolved. + interaction_history: Recent interactions of the story. + """ new_themes = set() for interaction in interaction_history[-10:]: # Consider last 10 interactions if interaction["resonance"] > 0.5: @@ -123,7 +187,12 @@ def evolve_themes(self, story: "Story", interaction_history): class StoryPerspective(BaseClass): - """Manages a story's evolving perspective""" + """ + Manages a story's evolving perspective in the narrative field. + + This class handles the updating of a story's perspective based on + interactions with other stories and tracks the history of perspective shifts. + """ def __init__(self, initial_filter: np.ndarray): super().__init__() @@ -139,7 +208,18 @@ def update( indirect_resonance: float, theme_relationships: List[tuple], ): - """Update perspective with detailed tracking""" + """ + Update the story's perspective based on interaction with another story. + + Args: + other_filter (np.ndarray): The perspective filter of the interacting story. + shared_themes (set): Themes shared between the stories. + indirect_resonance (float): Indirect resonance between the stories. + theme_relationships (List[tuple]): Related themes between the stories. + + Returns: + float: The magnitude of the perspective shift. + """ # Calculate influence factors direct_influence = len(shared_themes) * 0.15 indirect_influence = indirect_resonance * 0.1 @@ -185,6 +265,13 @@ def update( class EmotionalState(BaseClass): + """ + Represents and manages the emotional state of a story. + + This class handles the updating and evolution of a story's emotional state + based on interactions and external influences. + """ + def __init__(self, description: str, embedding: np.ndarray): super().__init__() self.description = description @@ -196,6 +283,14 @@ async def update( interaction_strength: float = 0.1, llm: Optional[LanguageModel] = None, ): + """ + Update the emotional state based on interaction with another story or external factors. + + Args: + other (Optional[EmotionalState]): The emotional state of the interacting story. + interaction_strength (float): The strength of the interaction. + llm (Optional[LanguageModel]): Language model for generating new emotional states. + """ if other is None or llm is None: # Simple update without interaction self.embedding += np.random.normal( @@ -232,6 +327,13 @@ def __str__(self): class Story(BaseClass): + """ + Represents a story in the narrative field. + + This class encapsulates all the properties and behaviors of a story, + including its content, themes, position in the field, and interactions. + """ + def __init__( self, id: str, @@ -265,13 +367,25 @@ def __init__( self._field = None if field: self.set_field(field) - self.theme_influences = {} # Add this line to track theme influences + self.theme_influences = {} # Track theme influences def set_field(self, field: "NarrativeField"): + """ + Set the narrative field for this story. + + Args: + field (NarrativeField): The narrative field to which this story belongs. + """ self._field = field @property def field(self) -> Optional["NarrativeField"]: + """ + Get the narrative field to which this story belongs. + + Returns: + Optional[NarrativeField]: The narrative field, if set. + """ return self._field def __str__(self): @@ -285,6 +399,19 @@ async def update_perspective( emotional_change: float, interaction_type: str, ) -> float: + """ + Update the story's perspective based on interaction with another story. + + Args: + other_story (Story): The story this story is interacting with. + theme_impact (float): The impact of shared themes. + resonance (float): The resonance between the stories. + emotional_change (float): The change in emotional state. + interaction_type (str): The type of interaction (e.g., "conflict", "collaboration"). + + Returns: + float: The magnitude of the perspective shift. + """ # Calculate perspective shift based on interaction factors shift = theme_impact * resonance * emotional_change * 0.1 @@ -300,6 +427,15 @@ async def update_perspective( return float(shift) # Ensure we return a float def _calculate_emotional_influence(self, other: "Story") -> float: + """ + Calculate the emotional influence of another story on this story. + + Args: + other (Story): The other story. + + Returns: + float: The calculated emotional influence. + """ similarity = self.field._calculate_emotional_similarity(self, other) intensity = np.mean( [getattr(self.emotional_state, e) for e in vars(self.emotional_state)] @@ -313,33 +449,75 @@ async def update_emotional_state( resonance: float, llm: LanguageModel, ): + """ + Update the story's emotional state based on interaction with another story. + + Args: + other (Story): The story this story is interacting with. + interaction_type (str): The type of interaction. + resonance (float): The resonance between the stories. + llm (LanguageModel): The language model for generating new emotional states. + """ self.previous_emotional_state = EmotionalState( self.emotional_state.description, self.emotional_state.embedding.copy() ) await self.emotional_state.update(other.emotional_state, resonance, llm) def decay_emotions(self, decay_rate=0.01): + """ + Apply emotional decay over time. + + Args: + decay_rate (float): The rate at which emotions decay. + """ for emotion in vars(self.emotional_state): current_value = getattr(self.emotional_state, emotion) decayed_value = max(0, current_value - decay_rate) setattr(self.emotional_state, emotion, decayed_value) def check_emotional_thresholds(self): + """ + Check if emotional thresholds have been crossed and trigger events if necessary. + """ if self.emotional_state.joy > 0.9: self.trigger_joyful_event() elif self.emotional_state.sadness > 0.8: self.trigger_melancholy_event() def evolve_themes(self, interaction_history): + """ + Evolve the story's themes based on recent interactions. + + Args: + interaction_history: The history of recent interactions. + """ self.field.theme_manager.evolve_themes(self, interaction_history) def calculate_interaction_strength(self, other_story): + """ + Calculate the strength of interaction with another story. + + Args: + other_story (Story): The story to interact with. + + Returns: + float: The calculated interaction strength. + """ base_strength = self.field.detect_resonance(self, other_story) interaction_history = self.get_interaction_history(other_story) familiarity_bonus = min(0.2, len(interaction_history) * 0.02) return base_strength + familiarity_bonus async def respond_to_environmental_event(self, event: Dict[str, Any]) -> float: + """ + Respond to an environmental event in the narrative field. + + Args: + event (Dict[str, Any]): The environmental event details. + + Returns: + float: The intensity of the response. + """ # Simple response to environmental events intensity = event.get("intensity", 0.5) @@ -359,6 +537,16 @@ async def respond_to_environmental_event(self, event: Dict[str, Any]) -> float: return intensity async def update_state(self, avg_resonance: float, avg_shift: float) -> float: + """ + Update the story's state based on recent interactions. + + Args: + avg_resonance (float): The average resonance from recent interactions. + avg_shift (float): The average perspective shift from recent interactions. + + Returns: + float: The magnitude of the state update. + """ # Update the story's state based on recent interactions await self.emotional_state.update(interaction_strength=avg_resonance * 0.1) shift = avg_shift * 0.1 @@ -368,15 +556,33 @@ async def update_state(self, avg_resonance: float, avg_shift: float) -> float: return shift def add_memory(self, memory: Memory): + """ + Add a new memory to the story's memory layer. + + Args: + memory (Memory): The memory to add. + """ self.memory_layer.append(memory) self.memory_state.update(memory) def update_theme_influence(self, theme: str, influence: float): + """ + Update the influence of a theme on this story. + + Args: + theme (str): The theme to update. + influence (float): The influence value to add. + """ self.theme_influences[theme] = self.theme_influences.get(theme, 0) + influence class NarrativeFieldViz(BaseClass): - """Handles visualization of field state""" + """ + Handles visualization of the narrative field state. + + This class captures and stores the state of the narrative field at different + time steps for later visualization or analysis. + """ def __init__(self, field_size: int = 1024): super().__init__() @@ -384,7 +590,13 @@ def __init__(self, field_size: int = 1024): self.history: List[Dict] = [] async def capture_state(self, field, timestep: int): - """Capture current field state for visualization""" + """ + Capture the current state of the narrative field. + + Args: + field (NarrativeField): The narrative field to capture. + timestep (int): The current time step of the simulation. + """ state = { "timestep": timestep, "story_positions": { @@ -403,7 +615,15 @@ async def capture_state(self, field, timestep: int): self.logger.debug(f"Captured field state at timestep {timestep}") def _compute_resonance_map(self, field) -> Dict: - """Compute resonance between all story pairs""" + """ + Compute the resonance between all pairs of stories in the field. + + Args: + field (NarrativeField): The narrative field. + + Returns: + Dict: A map of resonance values between story pairs. + """ resonance_map = {} for i, story1 in enumerate(field.stories): for story2 in field.stories[i + 1 :]: @@ -413,6 +633,13 @@ def _compute_resonance_map(self, field) -> Dict: class NarrativeField(BaseClass): + """ + Represents the narrative field in which stories interact and evolve. + + This class manages the overall state of the narrative field, including + all stories, their interactions, and the field's properties. + """ + def __init__(self, dimension: int = 1024): super().__init__() self.dimension = dimension @@ -428,7 +655,16 @@ def __init__(self, dimension: int = 1024): self.theme_manager = ThemeManager() def detect_resonance(self, story1: Story, story2: Story) -> float: - """Calculate resonance between two stories""" + """ + Calculate the resonance between two stories in the field. + + Args: + story1 (Story): The first story. + story2 (Story): The second story. + + Returns: + float: The calculated resonance between the stories. + """ # Base resonance on embedding similarity embedding_similarity = F.cosine_similarity( torch.tensor(story1.embedding), torch.tensor(story2.embedding), dim=0 @@ -464,6 +700,16 @@ def detect_resonance(self, story1: Story, story2: Story) -> float: ) def _calculate_emotional_similarity(self, story1: Story, story2: Story) -> float: + """ + Calculate the emotional similarity between two stories. + + Args: + story1 (Story): The first story. + story2 (Story): The second story. + + Returns: + float: The calculated emotional similarity. + """ # Calculate cosine similarity between emotional state embeddings similarity = F.cosine_similarity( torch.tensor(story1.emotional_state.embedding), @@ -473,14 +719,21 @@ def _calculate_emotional_similarity(self, story1: Story, story2: Story) -> float return float(similarity) def add_story(self, story: Story): - """Add a new story to the field""" + """ + Add a new story to the narrative field. + + Args: + story (Story): The story to add to the field. + """ story.set_field(self) self.stories.append(story) self._update_field_potential() self.logger.info(f"Added new story: {story.id}") def _update_field_potential(self): - """Update the field's potential energy based on story positions""" + """ + Update the field's potential energy based on story positions. + """ self.field_potential = np.zeros(self.dimension) for story in self.stories: # Each story contributes to the field potential @@ -489,14 +742,24 @@ def _update_field_potential(self): ) async def apply_environmental_event(self, event: Dict[str, Any]): - """Apply an environmental event to all stories in the field""" + """ + Apply an environmental event to all stories in the field. + + Args: + event (Dict[str, Any]): The environmental event details. + """ for story in self.stories: await story.respond_to_environmental_event(event) self.field_memory.append(event) class StoryPhysics(BaseClass): - """Handles physical behavior of stories in the field""" + """ + Handles the physical behavior of stories in the narrative field. + + This class manages the movement and interactions of stories based on + physical principles like attraction, repulsion, and containment. + """ def __init__( self, @@ -525,7 +788,14 @@ def __init__( self.target_zone_radius = target_zone_radius def update_story_motion(self, story: Story, field: NarrativeField, timestep: int): - """Update story position and velocity with balanced forces""" + """ + Update the motion of a story within the narrative field. + + Args: + story (Story): The story to update. + field (NarrativeField): The narrative field. + timestep (int): The current time step of the simulation. + """ net_force = np.zeros(3) # Forces from other stories @@ -591,21 +861,42 @@ def update_story_motion(self, story: Story, field: NarrativeField, timestep: int ) def _normalize_force(self, force: np.ndarray) -> np.ndarray: - """Normalize force vector to prevent exponential growth""" + """ + Normalize force vector to prevent exponential growth. + + Args: + force (np.ndarray): The force vector to normalize. + + Returns: + np.ndarray: The normalized force vector. + """ 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""" + """ + Limit velocity magnitude to prevent excessive speeds. + + Args: + velocity (np.ndarray): The velocity vector to limit. + + Returns: + np.ndarray: The limited velocity vector. + """ 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""" + """ + Apply global constraints to all stories in the field. + + Args: + stories (List[Story]): The list of stories to constrain. + """ # Find center of mass com = np.mean([s.position for s in stories], axis=0) @@ -619,19 +910,26 @@ def apply_field_constraints(self, stories: List[Story]): class EnhancedCollectiveStoryEngine(BaseClass): - """Enhanced version with more sophisticated pattern detection""" + """ + Enhanced version of the collective story engine with more sophisticated pattern detection. + + This class manages the collective state of all stories in the narrative field, + detecting emergent themes and generating field-wide effects. + """ def __init__(self, field: NarrativeField): super().__init__() self.field = field self.collective_memories = [] self.story_states: Dict[str, StoryState] = {} - self.collective_story = "" # Add this line to initialize the collective story + self.collective_story = "" # Initialize the collective story async def update_story_states(self): + """ + Update the states of all stories based on recent interactions. + """ for story in self.field.stories: - recent_memories = story.memory_layer[-5:] # Get the 5 most recent memories - if recent_memories: + if recent_memories := story.memory_layer[-5:]: avg_resonance = np.mean([m.resonance for m in recent_memories]) perspective_shifts = [] for m in recent_memories: @@ -647,7 +945,12 @@ async def update_story_states(self): story.total_perspective_shift += shift def detect_emergent_themes(self) -> List[str]: - """Detect themes that are becoming more prominent""" + """ + Detect themes that are becoming more prominent across all stories. + + Returns: + List[str]: The top emergent themes. + """ theme_weights = {} for story_id, state in self.story_states.items(): @@ -661,7 +964,13 @@ def detect_emergent_themes(self) -> List[str]: return [theme for theme, _ in sorted_themes[:3]] def generate_field_pulse(self, theme: str, intensity: float): - """Generate a collective field pulse around a theme""" + """ + Generate a collective field pulse around a specific theme. + + Args: + theme (str): The theme of the pulse. + intensity (float): The intensity of the pulse. + """ self.logger.info( f"Generating field pulse with theme '{theme}' and intensity {intensity}" ) @@ -684,7 +993,9 @@ def generate_field_pulse(self, theme: str, intensity: float): self.collective_memories.append(pulse) def write_collective_story(self): - """Generate and update the collective story based on current field state""" + """ + Generate and update the collective story based on current field state. + """ emergent_themes = self.detect_emergent_themes() story_summaries = [self.summarize_story(story) for story in self.field.stories] @@ -700,17 +1011,39 @@ def write_collective_story(self): self.logger.info(f"Updated collective story: {self.collective_story[:100]}...") def summarize_story(self, story: Story) -> str: - """Generate a brief summary of a story's current state""" + """ + Generate a brief summary of a story's current state. + + Args: + story (Story): The story to summarize. + + Returns: + str: A brief summary of the story's state. + """ return f"Story {story.id} resonates with {', '.join(story.themes[:3])}, its journey marked by {len(story.memory_layer)} memories." class ThemeManager(BaseClass): + """ + Manages theme relationships and their evolution in the narrative field. + """ + def __init__(self): super().__init__() self.relationship_map = ThemeRelationshipMap() self.evolution_engine = ThemeEvolutionEngine() async def process_theme_interaction(self, story1: Story, story2: Story): + """ + Process the theme interaction between two stories. + + Args: + story1 (Story): The first story. + story2 (Story): The second story. + + Returns: + float: The theme impact of the interaction. + """ shared_themes = set(story1.themes) & set(story2.themes) theme_impact = len(shared_themes) / max(len(story1.themes), len(story2.themes)) @@ -721,38 +1054,98 @@ async def process_theme_interaction(self, story1: Story, story2: Story): return theme_impact def get_theme_resonance(self, theme1: str, theme2: str) -> float: + """ + Get the resonance between two themes. + + Args: + theme1 (str): The first theme. + theme2 (str): The second theme. + + Returns: + float: The resonance between the themes. + """ return self.relationship_map.get_theme_resonance(theme1, theme2) def evolve_themes(self, story: Story, interaction_history): + """ + Evolve the themes of a story based on its interaction history. + + Args: + story (Story): The story whose themes are being evolved. + interaction_history: The interaction history of the story. + """ self.evolution_engine.evolve_themes(story, interaction_history) class BaseInteractionEngine(BaseClass): + """ + Base class for interaction engines in the narrative field. + """ + def __init__(self, field: NarrativeField): super().__init__() self.field = field async def process_interaction(self, story1: Story, story2: Story): + """ + Process the interaction between two stories. + + Args: + story1 (Story): The first story. + story2 (Story): The second story. + + Raises: + NotImplementedError: This method should be implemented by subclasses. + """ # Base implementation raise NotImplementedError("Subclasses must implement process_interaction") class StoryInteractionEngine(BaseInteractionEngine): + """ + Basic interaction engine for processing story interactions. + """ + async def process_interaction(self, story1: Story, story2: Story): + # sourcery skip: remove-empty-nested-block, remove-redundant-if + """ + Process the interaction between two stories. + + Args: + story1 (Story): The first story. + story2 (Story): The second story. + + Returns: + Tuple[float, None]: The resonance value and None (placeholder for interaction type). + """ # Basic interaction processing resonance = self.field.detect_resonance(story1, story2) if resonance > self.field.resonance_threshold: - # Perform basic interaction logic here + # TODO:Perform basic interaction logic here pass return resonance, None class EnhancedInteractionEngine(BaseInteractionEngine): + """ + Enhanced interaction engine with more sophisticated interaction processing. + """ + def __init__(self, field: NarrativeField, llm: LanguageModel): super().__init__(field) self.llm = llm async def determine_interaction_type(self, story1: Story, story2: Story) -> str: + """ + Determine the type of interaction between two stories. + + Args: + story1 (Story): The first story. + story2 (Story): The second story. + + Returns: + str: The determined interaction type. + """ # Calculate the similarity between the stories' themes shared_themes = set(story1.themes) & set(story2.themes) theme_similarity = len(shared_themes) / max( @@ -773,6 +1166,16 @@ async def determine_interaction_type(self, story1: Story, story2: Story) -> str: return "neutral" async def process_interaction(self, story1: Story, story2: Story): + """ + Process the interaction between two stories. + + Args: + story1 (Story): The first story. + story2 (Story): The second story. + + Returns: + Tuple[float, str]: The resonance value and interaction type. + """ if story1.id == story2.id: return 0, None # Prevent a story from interacting with itself @@ -814,17 +1217,39 @@ async def process_interaction(self, story1: Story, story2: Story): return 0, None def calculate_theme_impact(self, story1: Story, story2: Story) -> float: + """ + Calculate the theme impact between two stories. + + Args: + story1 (Story): The first story. + story2 (Story): The second story. + + Returns: + float: The calculated theme impact. + """ shared_themes = set(story1.themes) & set(story2.themes) return len(shared_themes) / max(len(story1.themes), len(story2.themes)) def _calculate_emotional_influence(self, story1: Story, story2: Story) -> float: + """ + Calculate the emotional influence between two stories. + + Args: + story1 (Story): The first story. + story2 (Story): The second story. + + Returns: + float: The calculated emotional influence. + """ # Implement the emotional influence calculation here # This is a placeholder implementation return 0.5 # Return a value between 0 and 1 class EnhancedJourneyLogger(BaseClass): - """Enhanced logger to track and analyze story journeys through the narrative field""" + """ + Enhanced logger to track and analyze story journeys through the narrative field. + """ def __init__(self): super().__init__() @@ -836,6 +1261,15 @@ def __init__(self): def log_interaction( self, story1: Story, story2: Story, resonance: float, interaction_type: str ): + """ + Log an interaction between two stories. + + Args: + story1 (Story): The first story. + story2 (Story): The second story. + resonance (float): The resonance of the interaction. + interaction_type (str): The type of interaction. + """ latest_memory = story1.memory_layer[-1] if story1.memory_layer else None perspective_shift = latest_memory.emotional_impact if latest_memory else 0 @@ -857,6 +1291,13 @@ def log_interaction( self.log_emotional_state(story2) def log_story_state(self, story: Story, timestep: float): + """ + Log the current state of a story. + + Args: + story (Story): The story to log. + timestep (float): The current timestep of the simulation. + """ if story.id not in self.journey_log: self.journey_log[story.id] = [] self.total_distances[story.id] = 0.0 @@ -892,6 +1333,12 @@ def log_story_state(self, story: Story, timestep: float): self.journey_log[story.id].append(state) def log_emotional_state(self, story: Story): + """ + Log the emotional state of a story. + + Args: + story (Story): The story whose emotional state is being logged. + """ if story.id not in self.emotional_history: self.emotional_history[story.id] = [] @@ -904,6 +1351,12 @@ def log_emotional_state(self, story: Story): ) def summarize_journey(self, story: Story): + """ + Summarize the journey of a story through the narrative field. + + Args: + story (Story): The story whose journey is being summarized. + """ journey = self.journey_log.get(story.id, []) if not journey: return @@ -928,11 +1381,11 @@ def summarize_journey(self, story: Story): else 0 ) - # Safely get unique interactions - unique_interactions = set() - for memory in story.memory_layer: - if hasattr(memory, 'partner_id') and memory.partner_id is not None: - unique_interactions.add(memory.partner_id) + unique_interactions = { + memory.partner_id + for memory in story.memory_layer + if hasattr(memory, "partner_id") and memory.partner_id is not None + } num_unique_interactions = len(unique_interactions) # Emotional journey analysis @@ -964,6 +1417,15 @@ def summarize_journey(self, story: Story): ) def get_journey_analytics(self, story: Story): + """ + Get analytics for a story's journey through the narrative field. + + Args: + story (Story): The story to analyze. + + Returns: + dict: A dictionary containing various analytics about the story's journey. + """ journey = self.journey_log.get(story.id, []) if not journey: return {} @@ -993,7 +1455,12 @@ def get_journey_analytics(self, story: Story): async def create_story_cluster(): - """Create initial story positions in a balanced configuration""" + """ + Create initial story positions in a balanced configuration. + + Returns: + np.ndarray: An array of initial story positions. + """ # Position stories in a triangle with some random offset base_positions = np.array( [[1.0, 0.0, 0.0], [-0.5, 0.866, 0.0], [-0.5, -0.866, 0.0]] @@ -1004,12 +1471,26 @@ async def create_story_cluster(): class DynamicThemeGenerator(BaseClass): + """ + Generates dynamic themes for stories using a language model. + """ + def __init__(self, llm: LanguageModel): super().__init__() self.llm = llm self.theme_cache = set() async def generate_themes(self, context: str, num_themes: int = 3) -> List[str]: + """ + Generate themes based on a given context. + + Args: + context (str): The context for theme generation. + num_themes (int): The number of themes to generate. + + Returns: + List[str]: A list of generated themes. + """ 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: ["hope", "journey", "transformation"] """ @@ -1017,14 +1498,11 @@ async def generate_themes(self, context: str, num_themes: int = 3) -> List[str]: response = response.strip().lower() try: - # Extract JSON array from the response - json_match = re.search(r"\[.*\]", response, re.DOTALL) - if json_match: - json_str = json_match.group(0) - new_themes = json.loads(json_str) - else: + if not (json_match := re.search(r"\[.*\]", response, re.DOTALL)): raise ValueError("No JSON array found in response") + json_str = json_match[0] + new_themes = json.loads(json_str) # Ensure we have the correct number of themes new_themes = [theme.lower() for theme in new_themes[:num_themes]] while len(new_themes) < num_themes: @@ -1040,18 +1518,40 @@ async def generate_themes(self, context: str, num_themes: int = 3) -> List[str]: return [f"theme_{i+1}" for i in range(num_themes)] def get_random_themes(self, num_themes: int = 3) -> List[str]: + """ + Get random themes from the theme cache. + + Args: + num_themes (int): The number of themes to retrieve. + + Returns: + List[str]: A list of random themes. + """ return random.sample( list(self.theme_cache), min(num_themes, len(self.theme_cache)) ) class DynamicStoryGenerator(BaseClass): + """ + Generates dynamic stories using a language model and theme generator. + """ + def __init__(self, llm: LanguageModel, theme_generator: DynamicThemeGenerator): super().__init__() self.llm = llm self.theme_generator = theme_generator async def generate_story(self, field: NarrativeField) -> Story: + """ + Generate a new story for the narrative field. + + Args: + field (NarrativeField): The narrative field the story will be added to. + + Returns: + Story: A newly generated story. + """ themes = await self.theme_generator.generate_themes("Create a new story") 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." @@ -1081,6 +1581,15 @@ async def generate_story(self, field: NarrativeField) -> Story: ) async def generate_emotional_state(self, content: str) -> EmotionalState: + """ + Generate an emotional state for a story based on its content. + + Args: + content (str): The content of the story. + + Returns: + EmotionalState: The generated emotional state. + """ 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) @@ -1088,11 +1597,21 @@ async def generate_emotional_state(self, content: str) -> EmotionalState: class EnvironmentalEventGenerator(BaseClass): + """ + Generates environmental events for the narrative field. + """ + def __init__(self, llm: LanguageModel): super().__init__() self.llm = llm async def generate_event(self) -> Dict[str, Any]: + """ + Generate a random environmental event. + + Returns: + Dict[str, Any]: A dictionary describing the generated event. + """ 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)." response = await self.llm.generate(prompt) @@ -1120,6 +1639,12 @@ async def generate_event(self) -> Dict[str, Any]: return event_data async def apply_event(self, field: NarrativeField): + """ + Apply a generated environmental event to the narrative field. + + Args: + field (NarrativeField): The narrative field to apply the event to. + """ try: event = await self.generate_event() self.logger.info( @@ -1133,8 +1658,10 @@ async def apply_event(self, field: NarrativeField): self.logger.error(f"Failed to generate or apply environmental event: {e}") -async def simulate_field(): - """Run a simulation of the narrative field""" +async def simulate_field(): # sourcery skip: low-code-quality + """ + Run a simulation of the narrative field. + """ # Set up logging logger = logging.getLogger(__name__) @@ -1152,7 +1679,7 @@ async def simulate_field(): interaction_engine = EnhancedInteractionEngine(field, llm) # Generate initial stories - for _ in range(15): + for _ in range(2): story = await story_generator.generate_story(field) field.add_story(story) @@ -1160,7 +1687,7 @@ async def simulate_field(): journey_logger = EnhancedJourneyLogger() # Simulation loop - for t in range(10): + for t in range(100): field.time = t # Update physics diff --git a/src/nfs_simple_lab_scenario.py b/src/nfs_simple_lab_scenario.py index 4600c2b..d9c3df6 100644 --- a/src/nfs_simple_lab_scenario.py +++ b/src/nfs_simple_lab_scenario.py @@ -1,6 +1,18 @@ """ Narrative Field System -A framework for analyzing and tracking narrative dynamics in complex social systems. + +This module implements a framework for analyzing and tracking narrative dynamics in complex social systems. +It uses language models and vector stores to process stories, detect patterns, and analyze the impact of narratives +on a simulated social environment. + +Key components: +- Story: Represents individual narratives within the system +- FieldState: Represents the current state of the narrative field +- VectorStore: Abstract base class for storing and retrieving story embeddings +- FieldAnalyzer: Analyzes the impact of stories on the field state +- ResonanceDetector: Finds similarities between stories +- NarrativeField: Main class that orchestrates the entire system +- PerformanceMetrics and PerformanceMonitor: Track and report on system performance """ from __future__ import annotations @@ -35,20 +47,56 @@ class VectorStore(ABC): + """ + Abstract base class for vector storage and retrieval operations. + Implementations of this class should provide methods for storing story embeddings + and finding similar stories based on their embeddings. + """ + @abstractmethod async def store(self, story: Story, embedding: List[float]) -> None: + """ + Store a story and its embedding in the vector store. + + Args: + story (Story): The story object to store. + embedding (List[float]): The embedding vector of the story. + """ pass @abstractmethod async def find_similar( self, embedding: List[float], threshold: float, limit: int ) -> List[Dict]: + """ + Find similar stories based on the given embedding. + + Args: + embedding (List[float]): The query embedding vector. + threshold (float): The similarity threshold for matching. + limit (int): The maximum number of similar stories to return. + + Returns: + List[Dict]: A list of dictionaries containing similar stories and their metadata. + """ pass -# Data Classes @dataclass class Story: + """ + Represents a single narrative or story within the system. + + Attributes: + content (str): The main text content of the story. + context (str): Additional context or metadata for the story. + id (StoryID): Unique identifier for the story. + timestamp (datetime): When the story was created or added to the system. + metadata (Optional[Dict[str, Any]]): Additional metadata for the story. + resonances (List[str]): IDs of other stories that resonate with this one. + field_effects (List[Dict]): Analyses of how this story affects the narrative field. + """ + content: str context: str id: StoryID = field(default_factory=lambda: StoryID(str(uuid4()))) @@ -60,6 +108,17 @@ class Story: @dataclass class FieldState: + """ + Represents the current state of the narrative field. + + Attributes: + description (str): A textual description of the current field state. + patterns (List[Dict[str, Any]]): Detected patterns in the narrative field. + active_resonances (List[Dict[str, Any]]): Currently active resonances between stories. + emergence_points (List[Dict[str, Any]]): Points where new narratives or patterns emerge. + timestamp (datetime): When this field state was last updated. + """ + description: str patterns: List[Dict[str, Any]] = field(default_factory=list) active_resonances: List[Dict[str, Any]] = field(default_factory=list) @@ -67,10 +126,24 @@ class FieldState: timestamp: datetime = field(default_factory=datetime.now) -# Prompt Management class FieldAnalysisPrompts: + """ + A collection of static methods that generate prompts for various field analysis tasks. + These prompts are used to guide the language model in analyzing the narrative field. + """ + @staticmethod def get_impact_analysis_prompt(story: Story, current_state: FieldState) -> str: + """ + Generate a prompt for analyzing the impact of a new story on the current field state. + + Args: + story (Story): The new story to analyze. + current_state (FieldState): The current state of the narrative field. + + Returns: + str: A formatted prompt for impact analysis. + """ return f"""Analyze how this new narrative affects the existing field state. Current Field State: @@ -102,6 +175,16 @@ def get_impact_analysis_prompt(story: Story, current_state: FieldState) -> str: def get_pattern_detection_prompt( stories: List[Story], current_state: FieldState ) -> str: + """ + Generate a prompt for detecting patterns across recent stories. + + Args: + stories (List[Story]): A list of recent stories to analyze. + current_state (FieldState): The current state of the narrative field. + + Returns: + str: A formatted prompt for pattern detection. + """ story_summaries = "\n".join(f"- {s.content}" for s in stories[-5:]) return f"""Analyze patterns and themes across these recent narratives. @@ -131,6 +214,16 @@ def get_pattern_detection_prompt( @staticmethod def get_resonance_analysis_prompt(story1: Story, story2: Story) -> str: + """ + Generate a prompt for analyzing the resonance between two stories. + + Args: + story1 (Story): The first story to compare. + story2 (Story): The second story to compare. + + Returns: + str: A formatted prompt for resonance analysis. + """ return f"""Analyze how these two narratives connect and influence each other. First Narrative: @@ -161,11 +254,25 @@ def get_resonance_analysis_prompt(story1: Story, story2: Story) -> str: class PerformanceMetrics: + """ + Tracks and reports performance metrics for various operations in the system. + + This class provides methods to start and stop timers, calculate average durations, + and log system resource usage. + """ + def __init__(self): + """Initialize the PerformanceMetrics object with empty metrics and a logger.""" self.metrics: Dict[str, Dict[str, Any]] = {} self.logger = logging.getLogger(__name__) def start_timer(self, operation: str): + """ + Start a timer for a specific operation. + + Args: + operation (str): The name of the operation to time. + """ if operation not in self.metrics: self.metrics[operation] = { "start_time": time.perf_counter(), @@ -175,6 +282,15 @@ def start_timer(self, operation: str): self.metrics[operation]["start_time"] = time.perf_counter() def stop_timer(self, operation: str) -> float: + """ + Stop the timer for a specific operation and record its duration. + + Args: + operation (str): The name of the operation to stop timing. + + Returns: + float: The duration of the operation in seconds. + """ if operation in self.metrics: duration = time.perf_counter() - self.metrics[operation]["start_time"] self.metrics[operation]["durations"].append(duration) @@ -182,6 +298,15 @@ def stop_timer(self, operation: str) -> float: return 0.0 def get_average_duration(self, operation: str) -> float: + """ + Calculate the average duration of a specific operation. + + Args: + operation (str): The name of the operation to calculate the average for. + + Returns: + float: The average duration of the operation in seconds. + """ if operation in self.metrics and self.metrics[operation]["durations"]: return sum(self.metrics[operation]["durations"]) / len( self.metrics[operation]["durations"] @@ -189,6 +314,7 @@ def get_average_duration(self, operation: str) -> float: return 0.0 def print_summary(self): + """Print a summary of all recorded performance metrics.""" print("\nPerformance Metrics Summary:") for operation, data in self.metrics.items(): if durations := data["durations"]: @@ -204,6 +330,7 @@ def print_summary(self): print(f"{operation}: No data") def log_system_resources(self): + """Log current CPU and memory usage.""" cpu_percent = psutil.cpu_percent() memory_info = psutil.virtual_memory() self.logger.info(f"CPU Usage: {cpu_percent}%") @@ -211,12 +338,29 @@ def log_system_resources(self): class PerformanceMonitor: + """ + Monitors and reports on the performance of language model operations. + + This class provides methods to track generation time and memory usage for LLM operations. + """ + def __init__(self): + """Initialize the PerformanceMonitor with an empty list of metrics.""" self.metrics = [] async def monitor_generation( self, llm: LanguageModel, prompt: str ) -> Tuple[str, Dict[str, float]]: + """ + Monitor the performance of a language model generation task. + + Args: + llm (LanguageModel): The language model interface to use. + prompt (str): The prompt to generate a response for. + + Returns: + Tuple[str, Dict[str, float]]: The generated response and a dictionary of performance metrics. + """ start_time = time.perf_counter() memory_before = psutil.virtual_memory().used @@ -234,6 +378,12 @@ async def monitor_generation( return response, metrics def get_performance_report(self) -> Dict[str, float]: + """ + Generate a report of average performance metrics. + + Returns: + Dict[str, float]: A dictionary containing average generation time and memory usage change. + """ if not self.metrics: return {"avg_generation_time": 0, "avg_memory_usage_change": 0} @@ -249,17 +399,50 @@ def get_performance_report(self) -> Dict[str, float]: @dataclass class BatchMetrics: + """ + Stores metrics for batch processing operations. + + Attributes: + batch_sizes (List[int]): List of batch sizes processed. + batch_times (List[float]): List of processing times for each batch. + memory_usage (List[float]): List of memory usage for each batch. + """ + batch_sizes: List[int] = field(default_factory=list) batch_times: List[float] = field(default_factory=list) memory_usage: List[float] = field(default_factory=list) class BatchProcessor: + """ + Handles batch processing of prompts using a language model. + + This class optimizes batch sizes based on memory usage and processes prompts in batches. + """ + def __init__(self, llm: LanguageModel): + """ + Initialize the BatchProcessor. + + Args: + llm (LanguageModel): The language model interface to use for processing. + """ self.llm = llm self.optimal_batch_size = 4 # Will be adjusted dynamically async def process_batch(self, prompts: List[str]) -> List[str]: + """ + Process a batch of prompts using the language model. + + This method dynamically adjusts the batch size based on memory usage and + processes the prompts in optimal batches. + + Args: + prompts (List[str]): List of prompts to process. + + Returns: + List[str]: List of generated responses for each prompt. + """ # Dynamic batch size adjustment based on memory usage memory_usage = psutil.Process().memory_info().rss / 1024 / 1024 if memory_usage > 0.8 * psutil.virtual_memory().total / 1024 / 1024: @@ -277,7 +460,20 @@ async def process_batch(self, prompts: List[str]) -> List[str]: class ChromaStore(VectorStore): + """ + Implementation of VectorStore using Chroma DB for storing and retrieving story embeddings. + + This class provides methods to store story embeddings and find similar stories based on + cosine similarity of their embeddings. + """ + def __init__(self, collection_name: str = "narrative_field"): + """ + Initialize the ChromaStore. + + Args: + collection_name (str): Name of the Chroma DB collection to use. + """ self.client = chromadb.Client(Settings(anonymized_telemetry=False)) self.logger = logging.getLogger(__name__) @@ -289,6 +485,13 @@ def __init__(self, collection_name: str = "narrative_field"): ) async def store(self, story: Story, embedding: List[float]) -> None: + """ + Store a story and its embedding in the Chroma DB. + + Args: + story (Story): The story object to store. + embedding (List[float]): The embedding vector of the story. + """ metadata = { "content": story.content, "context": story.context, @@ -320,6 +523,17 @@ async def find_similar( threshold: float = DEFAULT_SIMILARITY_THRESHOLD, limit: int = DEFAULT_RESONANCE_LIMIT, ) -> List[Dict]: + """ + Find similar stories based on the given embedding. + + Args: + embedding (List[float]): The query embedding vector. + threshold (float): The similarity threshold for matching. + limit (int): The maximum number of similar stories to return. + + Returns: + List[Dict]: A list of dictionaries containing similar stories and their metadata. + """ count = self.collection.count() if count == 0: return [] @@ -345,7 +559,19 @@ async def find_similar( class FieldAnalyzer: + """ + Analyzes the impact of stories on the narrative field and detects patterns. + + This class uses a language model to generate analyses based on prompts. + """ + def __init__(self, llm_interface: LanguageModel): + """ + Initialize the FieldAnalyzer. + + Args: + llm_interface (LanguageModel): The language model interface to use for analysis. + """ self.llm = llm_interface self.logger = logging.getLogger(__name__) self.prompts = FieldAnalysisPrompts() @@ -353,6 +579,16 @@ def __init__(self, llm_interface: LanguageModel): async def analyze_impact( self, story: Story, current_state: FieldState ) -> Dict[str, Any]: + """ + Analyze the impact of a new story on the current field state. + + Args: + story (Story): The new story to analyze. + current_state (FieldState): The current state of the narrative field. + + Returns: + Dict[str, Any]: A dictionary containing the analysis, timestamp, and story ID. + """ prompt = self.prompts.get_impact_analysis_prompt(story, current_state) analysis = await self.llm.generate(prompt) @@ -361,12 +597,36 @@ async def analyze_impact( async def detect_patterns( self, stories: List[Story], current_state: FieldState ) -> str: + """ + Detect patterns across a list of stories in the context of the current field state. + + Args: + stories (List[Story]): The list of stories to analyze for patterns. + current_state (FieldState): The current state of the narrative field. + + Returns: + str: A string describing the detected patterns. + """ prompt = self.prompts.get_pattern_detection_prompt(stories, current_state) return await self.llm.generate(prompt) class ResonanceDetector: + """ + Detects resonances between stories in the narrative field. + + This class uses a vector store to find similar stories and a language model + to analyze the nature of the resonance between stories. + """ + def __init__(self, vector_store: VectorStore, llm_interface: LanguageModel): + """ + Initialize the ResonanceDetector. + + Args: + vector_store (VectorStore): The vector store to use for finding similar stories. + llm_interface (LanguageModel): The language model interface to use for analysis. + """ self.vector_store = vector_store self.llm = llm_interface self.logger = logging.getLogger(__name__) @@ -378,6 +638,17 @@ async def find_resonances( threshold: float = DEFAULT_SIMILARITY_THRESHOLD, limit: int = DEFAULT_RESONANCE_LIMIT, ) -> List[Dict[str, Any]]: + """ + Find resonances between a given story and other stories in the vector store. + + Args: + story (Story): The story to find resonances for. + threshold (float): The similarity threshold for considering a resonance. + limit (int): The maximum number of resonances to return. + + Returns: + List[Dict[str, Any]]: A list of dictionaries describing the resonances found. + """ try: self.logger.debug(f"Generating embedding for story: {story.id}") # Ensure embedding is generated before using it @@ -417,6 +688,16 @@ async def find_resonances( async def determine_resonance_type( self, story1: Story, story2: Story ) -> Dict[str, Any]: + """ + Determine the type and nature of resonance between two stories. + + Args: + story1 (Story): The first story in the resonance pair. + story2 (Story): The second story in the resonance pair. + + Returns: + Dict[str, Any]: A dictionary describing the resonance between the stories. + """ prompt = self.prompts.get_resonance_analysis_prompt(story1, story2) analysis = await self.llm.generate(prompt) @@ -440,7 +721,21 @@ async def determine_resonance_type( class NarrativeField: + """ + Main class that orchestrates the entire Narrative Field System. + + This class manages the addition of new stories, updates to the field state, + and coordinates the analysis and resonance detection processes. + """ + def __init__(self, llm_interface: LanguageModel, vector_store: VectorStore): + """ + Initialize the NarrativeField. + + Args: + llm_interface (LanguageModel): The language model interface to use for analysis. + vector_store (VectorStore): The vector store to use for story storage and retrieval. + """ self._analyzer = FieldAnalyzer(llm_interface) self._resonance_detector = ResonanceDetector(vector_store, llm_interface) self._vector_store = vector_store @@ -451,13 +746,28 @@ def __init__(self, llm_interface: LanguageModel, vector_store: VectorStore): @property def state(self) -> FieldState: + """Get the current state of the narrative field.""" return self._state @property def stories(self) -> Dict[StoryID, Story]: + """Get a copy of all stories in the narrative field.""" return self._stories.copy() async def add_story(self, content: str, context: str) -> Story: + """ + Add a new story to the narrative field and analyze its impact. + + This method creates a new story, analyzes its impact on the field, + finds resonances with other stories, and updates the field state. + + Args: + content (str): The main content of the story. + context (str): The context or additional information about the story. + + Returns: + Story: The newly created and analyzed story object. + """ self._performance_metrics.start_timer("add_story") self._performance_metrics.start_timer("create_story") @@ -495,6 +805,12 @@ async def add_story(self, content: str, context: str) -> Story: return story async def _store_story(self, story: Story) -> None: + """ + Store a story in the vector store and local dictionary. + + Args: + story (Story): The story to store. + """ embedding = await self._resonance_detector.llm.generate_embedding( f"{story.content} {story.context}" ) @@ -504,6 +820,14 @@ async def _store_story(self, story: Story) -> None: async def _update_field_state( self, story: Story, impact: Dict, resonances: List[Dict] ) -> None: + """ + Update the field state based on a new story, its impact, and resonances. + + Args: + story (Story): The new story added to the field. + impact (Dict): The impact analysis of the story. + resonances (List[Dict]): The resonances found for the story. + """ patterns = await self._analyzer.detect_patterns( list(self._stories.values()), self.state ) @@ -527,6 +851,11 @@ async def _update_field_state( # Global cleanup function def global_cleanup(): + """ + Perform global cleanup operations. + + This function is called at program exit to free up resources and clear caches. + """ gc.collect() if torch.cuda.is_available(): torch.cuda.empty_cache() @@ -537,6 +866,12 @@ def global_cleanup(): async def demo_scenario(): + """ + Run a demonstration scenario of the Narrative Field System. + + This function sets up the system, processes a series of stories, + and logs performance metrics throughout the process. + """ logger = logging.getLogger(__name__) logger.info("Starting narrative field demonstration...") @@ -566,7 +901,7 @@ async def demo_scenario(): }, # Event 2: Robbert's tough advice { - "content": "After work, Robbert and Leon walked back to the lab together. Leon expressed his worries about Danny's accident and the AI minor. However, Robbert seemed more preoccupied with his own research and was not interested in discussing the minor. \"I know you're concerned, but you need to man up and stop whining,\" Robbert said bluntly. His tough advice left Leon feeling isolated and unsupported.", + "content": "After work, Robbert and Leon walked back to the lab together. Leon expressed his worries about Danny's accident and the AI minor. However, Robbert seemed more preoccupied with his own research and was not interested in discussing the minor. \"I know you're concerned, but you need to man up and stop whining,\" Robbert said bluntly. His tough advice left Leon feeling isolated.", "context": "Robbert dismisses Leon's concerns, focusing instead on his own research priorities.", }, # Event 4: Sarah's contribution diff --git a/src/nfs_story_waves.py b/src/nfs_story_waves.py index c7eb124..16261cc 100644 --- a/src/nfs_story_waves.py +++ b/src/nfs_story_waves.py @@ -2,7 +2,7 @@ from dataclasses import dataclass, field import torch import torch.autograd -from typing import List, Dict, Any, Set +from typing import List, Dict, Any, Set, Tuple from transformers import AutoTokenizer, AutoModel import logging import torch.nn.functional as F @@ -288,14 +288,16 @@ def analyze_attractor(self) -> Dict: 'strength_stability': 0.0, 'dominant_frequency': 0.0 } - + recent_states = self.trajectory[-10:] - + field_variance = torch.var(torch.stack([s['field'] for s in recent_states])) - - # Handle case where pattern_strength might be empty - strength_tensors = [s['pattern_strength'] for s in recent_states if s['pattern_strength'].numel() > 0] - if strength_tensors: + + if strength_tensors := [ + s['pattern_strength'] + for s in recent_states + if s['pattern_strength'].numel() > 0 + ]: strength_variance = torch.var(torch.cat(strength_tensors)) strength_series = torch.cat(strength_tensors) if strength_series.numel() > 1: @@ -307,11 +309,11 @@ def analyze_attractor(self) -> Dict: else: strength_variance = torch.tensor(0.0) dominant_frequency = 0.0 - + # Avoid division by zero field_stability = float(1.0 / (1.0 + field_variance)) if field_variance != 0 else 1.0 strength_stability = float(1.0 / (1.0 + strength_variance)) if strength_variance != 0 else 1.0 - + return { 'field_stability': field_stability, 'strength_stability': strength_stability, @@ -447,35 +449,35 @@ def analyze_interactions(self, stories: Dict[str, NarrativeWave]) -> Dict: """ story_ids = list(stories.keys()) n_stories = len(story_ids) - + if n_stories < 2: return {} - + # Create interaction matrix interaction_matrix = torch.zeros((n_stories, n_stories)) - + for i in range(n_stories): for j in range(i+1, n_stories): story1 = stories[story_ids[i]] story2 = stories[story_ids[j]] - + semantic_similarity = torch.cosine_similarity( story1.embedding.unsqueeze(0), story2.embedding.unsqueeze(0) ) - + phase_coupling = torch.cos(story1.phase - story2.phase) amplitude_ratio = torch.min(story1.amplitude, story2.amplitude) / \ - torch.max(story1.amplitude, story2.amplitude) - + torch.max(story1.amplitude, story2.amplitude) + interaction_strength = ( semantic_similarity * phase_coupling * amplitude_ratio ).item() - + interaction_matrix[i, j] = interaction_matrix[j, i] = interaction_strength - + try: eigenvalues, eigenvectors = torch.linalg.eigh(interaction_matrix) except torch._C._LinAlgError: @@ -488,16 +490,14 @@ def analyze_interactions(self, stories: Dict[str, NarrativeWave]) -> Dict: threshold = 0.7 for i, vec in enumerate(eigenvectors.T): if eigenvalues[i] > threshold: - cluster_members = [ - story_ids[j] for j in range(n_stories) - if abs(vec[j]) > 0.3 - ] - if cluster_members: + if cluster_members := [ + story_ids[j] for j in range(n_stories) if abs(vec[j]) > 0.3 + ]: clusters.append({ 'strength': float(eigenvalues[i]), 'members': cluster_members }) - + return { 'interaction_matrix': interaction_matrix.tolist(), 'eigenvalues': eigenvalues.tolist(), @@ -819,7 +819,7 @@ def update_field_state(self): else: self.field_state = torch.rand_like(self.field_state) * 1e-6 - def detect_emergence(self) -> List[Dict]: + def detect_emergence(self) -> List[Dict]: # sourcery skip: low-code-quality """ Enhanced pattern detection with better uniqueness handling. @@ -1009,19 +1009,7 @@ def simulate_timestep(self, dt: float): # Analyze attractor properties periodically if len(self.phase_space_tracker.trajectory) % 100 == 0: - attractor_properties = self.phase_space_tracker.analyze_attractor() - field_energy = torch.norm(self.field_state) - - if torch.isnan(field_energy): - logger.warning("NaN detected in field energy calculation.") - - logger.info(f"Attractor properties: {attractor_properties}") - logger.info(f"Current field energy: {field_energy:.4f}") - logger.info(f"Number of active stories: {len(self.stories)}") - if self.stories: - avg_coherence = sum(story.coherence.item() for story in self.stories.values()) / len(self.stories) - logger.info(f"Average story coherence: {avg_coherence:.4f}") - + self._extracted_from_simulate_timestep_83() # Log the number of active stories at the end of the timestep logger.info(f"Number of active stories at end of timestep: {len(self.stories)}") @@ -1030,7 +1018,7 @@ def simulate_timestep(self, dt: float): story.amplitude = story.amplitude.clamp(min=0.1, max=10.0) story.coherence = story.coherence.clamp(min=0.1, max=1.0) story.embedding = torch.where(torch.isfinite(story.embedding), story.embedding, torch.rand_like(story.embedding) * 1e-6) - + self.field_state = torch.where(torch.isfinite(self.field_state), self.field_state, torch.rand_like(self.field_state) * 1e-6) self.field_state = self.field_state.clamp(min=-10, max=10) @@ -1038,15 +1026,30 @@ def simulate_timestep(self, dt: float): frequency_data = self.frequency_analyzer.analyze_frequencies(self.field_state, patterns) logger.info(f"Dominant frequencies: {frequency_data['dominant_frequencies']}") - # Interaction analysis - interaction_data = self.interaction_analyzer.analyze_interactions(self.stories) - if interaction_data: + if interaction_data := self.interaction_analyzer.analyze_interactions( + self.stories + ): logger.info(f"Detected {len(interaction_data['clusters'])} interaction clusters") # Comprehensive metrics metrics = self.field_metrics.update(self.field_state, self.stories, patterns) logger.info(f"Field metrics: {metrics}") + # TODO Rename this here and in `simulate_timestep` + def _extracted_from_simulate_timestep_83(self): + attractor_properties = self.phase_space_tracker.analyze_attractor() + field_energy = torch.norm(self.field_state) + + if torch.isnan(field_energy): + logger.warning("NaN detected in field energy calculation.") + + logger.info(f"Attractor properties: {attractor_properties}") + logger.info(f"Current field energy: {field_energy:.4f}") + logger.info(f"Number of active stories: {len(self.stories)}") + if self.stories: + avg_coherence = sum(story.coherence.item() for story in self.stories.values()) / len(self.stories) + logger.info(f"Average story coherence: {avg_coherence:.4f}") + # Example usage simulator = NarrativeFieldSimulator()